forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Generic Streamable Registry (opensearch-project#7780)
This commit adds a WriteableRegistry to register generic read / write serialization methods for StreamInput.readGeneric and StreamOutput.writeGeneric. This enables modules, plugins, and libraries to register their own type serialization logic without forcing the class implementation to be in the :server module. Decoupling this logic also further decouples the StreamInput / StreamOutput transport logic from the server module such that it can be refactored to the core library to support cloud native and serverless implementations. Signed-off-by: Nicholas Walter Knize <[email protected]>
- Loading branch information
1 parent
d4ffd8e
commit d93bd95
Showing
9 changed files
with
164 additions
and
114 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
89 changes: 89 additions & 0 deletions
89
server/src/main/java/org/opensearch/common/io/stream/Streamables.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,89 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
|
|
||
| package org.opensearch.common.io.stream; | ||
|
|
||
| import org.joda.time.DateTimeZone; | ||
| import org.joda.time.ReadableInstant; | ||
| import org.opensearch.common.geo.GeoPoint; | ||
| import org.opensearch.common.time.DateUtils; | ||
| import org.opensearch.core.common.io.stream.BaseWriteable.WriteableRegistry; | ||
| import org.opensearch.core.common.io.stream.BaseWriteable; | ||
| import org.opensearch.script.JodaCompatibleZonedDateTime; | ||
|
|
||
| import java.time.Instant; | ||
| import java.time.ZoneId; | ||
|
|
||
| /** | ||
| * This utility class registers generic types for streaming over the wire using | ||
| * {@linkplain StreamOutput#writeGenericValue(Object)} and {@linkplain StreamInput#readGenericValue()} | ||
| * | ||
| * In this manner we can register any type across OpenSearch modules, plugins, or libraries without requiring | ||
| * the implementation reside in the server module. | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public final class Streamables { | ||
|
|
||
| // no instance: | ||
| private Streamables() {} | ||
|
|
||
| /** | ||
| * Called when {@linkplain org.opensearch.transport.TransportService} is loaded by the classloader | ||
| * We do this because streamables depend on the TransportService being loaded | ||
| */ | ||
| public static void registerStreamables() { | ||
| registerWriters(); | ||
| registerReaders(); | ||
| } | ||
|
|
||
| /** | ||
| * Registers writers by class type | ||
| */ | ||
| private static void registerWriters() { | ||
| /** {@link ReadableInstant} */ | ||
| WriteableRegistry.<BaseWriteable.Writer<StreamOutput, ?>>registerWriter(ReadableInstant.class, (o, v) -> { | ||
| o.writeByte((byte) 13); | ||
| final ReadableInstant instant = (ReadableInstant) v; | ||
| o.writeString(instant.getZone().getID()); | ||
| o.writeLong(instant.getMillis()); | ||
| }); | ||
| WriteableRegistry.registerClassAlias(ReadableInstant.class, ReadableInstant.class); | ||
| /** {@link JodaCompatibleZonedDateTime} */ | ||
| WriteableRegistry.<BaseWriteable.Writer<StreamOutput, ?>>registerWriter(JodaCompatibleZonedDateTime.class, (o, v) -> { | ||
| // write the joda compatibility datetime as joda datetime | ||
| o.writeByte((byte) 13); | ||
| final JodaCompatibleZonedDateTime zonedDateTime = (JodaCompatibleZonedDateTime) v; | ||
| String zoneId = zonedDateTime.getZonedDateTime().getZone().getId(); | ||
| // joda does not understand "Z" for utc, so we must special case | ||
| o.writeString(zoneId.equals("Z") ? DateTimeZone.UTC.getID() : zoneId); | ||
| o.writeLong(zonedDateTime.toInstant().toEpochMilli()); | ||
| }); | ||
| /** {@link GeoPoint} */ | ||
| BaseWriteable.WriteableRegistry.<BaseWriteable.Writer<StreamOutput, ?>>registerWriter(GeoPoint.class, (o, v) -> { | ||
| o.writeByte((byte) 22); | ||
| ((GeoPoint) v).writeTo(o); | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Registers a reader function mapped by ordinal values that are written by {@linkplain StreamOutput} | ||
| * | ||
| * NOTE: see {@code StreamOutput#WRITERS} for all registered ordinals | ||
| */ | ||
| private static void registerReaders() { | ||
| /** {@link JodaCompatibleZonedDateTime */ | ||
| WriteableRegistry.<BaseWriteable.Reader<StreamInput, ?>>registerReader(Byte.valueOf((byte) 13), (i) -> { | ||
| final ZoneId zoneId = DateUtils.dateTimeZoneToZoneId(DateTimeZone.forID(i.readString())); | ||
| long millis = i.readLong(); | ||
| return new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), zoneId); | ||
| }); | ||
| /** {@link GeoPoint} */ | ||
| WriteableRegistry.<BaseWriteable.Reader<StreamInput, ?>>registerReader(Byte.valueOf((byte) 22), GeoPoint::new); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.