From 578db4c815dd957cd3b07fc24d3d54d6909f0010 Mon Sep 17 00:00:00 2001 From: Nick Knize Date: Thu, 11 May 2023 15:08:46 -0500 Subject: [PATCH 1/2] [Refactor] Stream Reader and Write Generics (#7465) StreamInput and StreamOutput provide the core functionality for marshalling / unmarshalling objects over the transport wire. The class is intended to be generic but it is tightly coupled to the types defined in the server module. Because of this tight coupling, the classes cannot be refactored to the core library, thus all new types are required to be hard coded in the server module. To decouple this logic and make it more generic across opensearch modules and plugins, this commit introduces a reader and writer registry in a new BaseWriteable interface. The StreamInput and StreamOutput also now inherits from new BaseStreamInput and BaseStreamOutput classes, respectively, located in the core library. This will be the new home for streaming primitives in a follow up commit. Signed-off-by: Nicholas Walter Knize --- .../common/io/stream/BaseStreamInput.java | 19 +++ .../common/io/stream/BaseStreamOutput.java | 19 +++ .../core/common/io/stream/BaseWriteable.java | 130 ++++++++++++++++++ .../core/common/io/stream/package-info.java | 9 ++ .../bucket/geogrid/InternalGeoGrid.java | 3 +- .../bucket/geogrid/InternalGeoHashGrid.java | 2 +- .../bucket/geogrid/InternalGeoTileGrid.java | 2 +- .../opensearch/common/geo/GeoBoundingBox.java | 8 +- .../org/opensearch/common/geo/GeoPoint.java | 26 ++++ .../common/io/stream/StreamInput.java | 41 ++---- .../common/io/stream/StreamOutput.java | 45 ++---- .../common/io/stream/Writeable.java | 27 +--- .../index/query/GeoDistanceQueryBuilder.java | 4 +- .../index/query/GeoPolygonQueryBuilder.java | 4 +- .../script/JodaCompatibleZonedDateTime.java | 26 ++++ .../transport/TransportService.java | 19 +++ .../common/io/stream/BaseStreamTests.java | 14 ++ .../common/io/stream/BytesStreamsTests.java | 4 +- .../opensearch/test/OpenSearchTestCase.java | 2 + 19 files changed, 310 insertions(+), 94 deletions(-) create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java create mode 100644 libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java new file mode 100644 index 0000000000000..a0e50722bf01d --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.java @@ -0,0 +1,19 @@ +/* + * 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.core.common.io.stream; + +import java.io.InputStream; + +/** + * Foundation class for reading core types off the transport stream + * + * todo: refactor {@code StreamInput} primitive readers to this class + * + * @opensearch.internal + */ +public abstract class BaseStreamInput extends InputStream {} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java new file mode 100644 index 0000000000000..f7a8862fa5f2c --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.java @@ -0,0 +1,19 @@ +/* + * 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.core.common.io.stream; + +import java.io.OutputStream; + +/** + * Foundation class for writing core types over the transport stream + * + * todo: refactor {@code StreamOutput} primitive writers to this class + * + * @opensearch.internal + */ +public abstract class BaseStreamOutput extends OutputStream {} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java new file mode 100644 index 0000000000000..56172e7c6a50e --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.java @@ -0,0 +1,130 @@ +/* + * 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.core.common.io.stream; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Implementers can be written to a {@code StreamOutput} and read from a {@code StreamInput}. This allows them to be "thrown + * across the wire" using OpenSearch's internal protocol. If the implementer also implements equals and hashCode then a copy made by + * serializing and deserializing must be equal and have the same hashCode. It isn't required that such a copy be entirely unchanged. + * + * @opensearch.internal + */ +public interface BaseWriteable { + /** + * A WriteableRegistry registers {@link Writer} methods for writing data types over a + * {@link BaseStreamOutput} channel and {@link Reader} methods for reading data from a + * {@link BaseStreamInput} channel. + * + * @opensearch.internal + */ + class WriteableRegistry { + private static final Map, Writer> WRITER_REGISTRY = new ConcurrentHashMap<>(); + private static final Map> READER_REGISTRY = new ConcurrentHashMap<>(); + + /** + * registers a streamable writer + * + * @opensearch.internal + */ + public static > void registerWriter(final Class clazz, final W writer) { + if (WRITER_REGISTRY.containsKey(clazz)) { + throw new IllegalArgumentException("Streamable writer already registered for type [" + clazz.getName() + "]"); + } + WRITER_REGISTRY.put(clazz, writer); + } + + /** + * registers a streamable reader + * + * @opensearch.internal + */ + public static > void registerReader(final byte ordinal, final R reader) { + if (READER_REGISTRY.containsKey(ordinal)) { + throw new IllegalArgumentException("Streamable reader already registered for ordinal [" + (int) ordinal + "]"); + } + READER_REGISTRY.put(ordinal, reader); + } + + /** + * Returns the registered writer keyed by the class type + */ + @SuppressWarnings("unchecked") + public static > W getWriter(final Class clazz) { + return (W) WRITER_REGISTRY.get(clazz); + } + + /** + * Returns the ristered reader keyed by the unique ordinal + */ + @SuppressWarnings("unchecked") + public static > R getReader(final byte b) { + return (R) READER_REGISTRY.get(b); + } + } + + /** + * Write this into the {@linkplain BaseStreamOutput}. + */ + void writeTo(final S out) throws IOException; + + /** + * Reference to a method that can write some object to a {@link BaseStreamOutput}. + *

+ * By convention this is a method from {@link BaseStreamOutput} itself (e.g., {@code StreamOutput#writeString}). If the value can be + * {@code null}, then the "optional" variant of methods should be used! + *

+ * Most classes should implement {@code Writeable} and the {@code Writeable#writeTo(BaseStreamOutput)} method should use + * {@link BaseStreamOutput} methods directly or this indirectly: + * 


+     * public void writeTo(StreamOutput out) throws IOException {
+     *     out.writeVInt(someValue);
+     *     out.writeMapOfLists(someMap, StreamOutput::writeString, StreamOutput::writeString);
+     * }
+     *
+ */ + @FunctionalInterface + interface Writer { + + /** + * Write {@code V}-type {@code value} to the {@code out}put stream. + * + * @param out Output to write the {@code value} too + * @param value The value to add + */ + void write(final S out, V value) throws IOException; + } + + /** + * Reference to a method that can read some object from a stream. By convention this is a constructor that takes + * {@linkplain BaseStreamInput} as an argument for most classes and a static method for things like enums. Returning null from one of these + * is always wrong - for that we use methods like {@code StreamInput#readOptionalWriteable(Reader)}. + *

+ * As most classes will implement this via a constructor (or a static method in the case of enumerations), it's something that should + * look like: + * 


+     * public MyClass(final StreamInput in) throws IOException {
+     *     this.someValue = in.readVInt();
+     *     this.someMap = in.readMapOfLists(StreamInput::readString, StreamInput::readString);
+     * }
+     *
+ */ + @FunctionalInterface + interface Reader { + + /** + * Read {@code V}-type value from a stream. + * + * @param in Input to read the value from + */ + V read(final S in) throws IOException; + } +} diff --git a/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java b/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java new file mode 100644 index 0000000000000..76d0842466b96 --- /dev/null +++ b/libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.java @@ -0,0 +1,9 @@ +/* + * 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. + */ +/** Core transport stream classes */ +package org.opensearch.core.common.io.stream; diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoGrid.java index 5c91bb4cfd8f8..4e8fd92d2d45e 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoGrid.java @@ -33,7 +33,6 @@ import org.opensearch.common.io.stream.StreamInput; import org.opensearch.common.io.stream.StreamOutput; -import org.opensearch.common.io.stream.Writeable; import org.opensearch.common.util.LongObjectPagedHashMap; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.search.aggregations.InternalAggregation; @@ -69,7 +68,7 @@ protected InternalGeoGrid(String name, int requiredSize, List getBucketReader(); + protected abstract Reader getBucketReader(); /** * Read from a stream. diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGrid.java index 9a1a81f6fd650..071f844c80b03 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoHashGrid.java @@ -76,7 +76,7 @@ protected InternalGeoHashGridBucket createBucket(long hashAsLong, long docCount, } @Override - protected Reader getBucketReader() { + protected Reader getBucketReader() { return InternalGeoHashGridBucket::new; } diff --git a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoTileGrid.java b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoTileGrid.java index c5c86f8a62069..559caeefd2e14 100644 --- a/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoTileGrid.java +++ b/modules/geo/src/main/java/org/opensearch/geo/search/aggregations/bucket/geogrid/InternalGeoTileGrid.java @@ -76,7 +76,7 @@ protected InternalGeoTileGridBucket createBucket(long hashAsLong, long docCount, } @Override - protected Reader getBucketReader() { + protected Reader getBucketReader() { return InternalGeoTileGridBucket::new; } diff --git a/server/src/main/java/org/opensearch/common/geo/GeoBoundingBox.java b/server/src/main/java/org/opensearch/common/geo/GeoBoundingBox.java index 8e3db45854f28..9609c10b6614f 100644 --- a/server/src/main/java/org/opensearch/common/geo/GeoBoundingBox.java +++ b/server/src/main/java/org/opensearch/common/geo/GeoBoundingBox.java @@ -79,8 +79,8 @@ public GeoBoundingBox(GeoPoint topLeft, GeoPoint bottomRight) { } public GeoBoundingBox(StreamInput input) throws IOException { - this.topLeft = input.readGeoPoint(); - this.bottomRight = input.readGeoPoint(); + this.topLeft = new GeoPoint(input); + this.bottomRight = new GeoPoint(input); } public boolean isUnbounded() { @@ -164,8 +164,8 @@ public boolean pointInBounds(double lon, double lat) { @Override public void writeTo(StreamOutput out) throws IOException { - out.writeGeoPoint(topLeft); - out.writeGeoPoint(bottomRight); + topLeft.writeTo(out); + bottomRight.writeTo(out); } @Override diff --git a/server/src/main/java/org/opensearch/common/geo/GeoPoint.java b/server/src/main/java/org/opensearch/common/geo/GeoPoint.java index b5c2d6a846f92..874f0ffb80be1 100644 --- a/server/src/main/java/org/opensearch/common/geo/GeoPoint.java +++ b/server/src/main/java/org/opensearch/common/geo/GeoPoint.java @@ -40,6 +40,11 @@ import org.apache.lucene.util.BytesRef; import org.opensearch.OpenSearchParseException; import org.opensearch.common.geo.GeoUtils.EffectivePoint; +import org.opensearch.common.io.stream.StreamInput; +import org.opensearch.common.io.stream.StreamOutput; +import org.opensearch.core.common.io.stream.BaseWriteable.Reader; +import org.opensearch.core.common.io.stream.BaseWriteable.Writer; +import org.opensearch.core.common.io.stream.BaseWriteable.WriteableRegistry; import org.opensearch.core.xcontent.ToXContentFragment; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.geometry.Geometry; @@ -87,6 +92,22 @@ public GeoPoint(GeoPoint template) { this(template.getLat(), template.getLon()); } + public GeoPoint(final StreamInput in) throws IOException { + this.lat = in.readDouble(); + this.lon = in.readDouble(); + } + + /** + * Register this type as a streamable so it can be serialized over the wire + */ + public static void registerStreamables() { + WriteableRegistry.>registerWriter(GeoPoint.class, (o, v) -> { + o.writeByte((byte) 22); + ((GeoPoint) v).writeTo(o); + }); + WriteableRegistry.>registerReader(Byte.valueOf((byte) 22), GeoPoint::new); + } + public GeoPoint reset(double lat, double lon) { this.lat = lat; this.lon = lon; @@ -210,6 +231,11 @@ public GeoPoint resetFromGeoHash(long geohashLong) { return this.resetFromIndexHash(BitUtil.flipFlop((geohashLong >>> 4) << ((level * 5) + 2))); } + public void writeTo(final StreamOutput out) throws IOException { + out.writeDouble(this.lat); + out.writeDouble(this.lon); + } + public double lat() { return this.lat; } diff --git a/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java b/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java index f1b4ffe2219aa..2b51d6c469fcf 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java +++ b/server/src/main/java/org/opensearch/common/io/stream/StreamInput.java @@ -50,20 +50,18 @@ import org.opensearch.common.Strings; import org.opensearch.common.bytes.BytesArray; import org.opensearch.common.bytes.BytesReference; -import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.settings.SecureString; import org.opensearch.common.text.Text; -import org.opensearch.common.time.DateUtils; import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.io.stream.BaseStreamInput; +import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; -import org.opensearch.script.JodaCompatibleZonedDateTime; import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.FileNotFoundException; import java.io.FilterInputStream; import java.io.IOException; -import java.io.InputStream; import java.math.BigInteger; import java.nio.file.AccessDeniedException; import java.nio.file.AtomicMoveNotSupportedException; @@ -108,7 +106,7 @@ * * @opensearch.internal */ -public abstract class StreamInput extends InputStream { +public abstract class StreamInput extends BaseStreamInput { private Version version = Version.CURRENT; @@ -686,6 +684,11 @@ public Map readMap() throws IOException { @Nullable public Object readGenericValue() throws IOException { byte type = readByte(); + BaseWriteable.Reader r = BaseWriteable.WriteableRegistry.getReader(type); + if (r != null) { + return r.read(this); + } + switch (type) { case -1: return null; @@ -715,8 +718,6 @@ public Object readGenericValue() throws IOException { return readByte(); case 12: return readDate(); - case 13: - return readDateTime(); case 14: return readBytesReference(); case 15: @@ -733,8 +734,6 @@ public Object readGenericValue() throws IOException { return readDoubleArray(); case 21: return readBytesRef(); - case 22: - return readGeoPoint(); case 23: return readZonedDateTime(); case 24: @@ -778,14 +777,6 @@ private List readArrayList() throws IOException { return list; } - private JodaCompatibleZonedDateTime readDateTime() throws IOException { - // we reuse DateTime to communicate with older nodes that don't know about the joda compat layer, but - // here we are on a new node so we always want a compat datetime - final ZoneId zoneId = DateUtils.dateTimeZoneToZoneId(DateTimeZone.forID(readString())); - long millis = readLong(); - return new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), zoneId); - } - private ZonedDateTime readZonedDateTime() throws IOException { final String timeZoneId = readString(); return ZonedDateTime.ofInstant(Instant.ofEpochMilli(readLong()), ZoneId.of(timeZoneId)); @@ -833,13 +824,6 @@ private Date readDate() throws IOException { return new Date(readLong()); } - /** - * Reads a {@link GeoPoint} from this stream input - */ - public GeoPoint readGeoPoint() throws IOException { - return new GeoPoint(readDouble(), readDouble()); - } - /** * Read a {@linkplain DateTimeZone}. */ @@ -1181,7 +1165,7 @@ public C readOptionalNamedWriteable(Class category * @return the list of objects * @throws IOException if an I/O exception occurs reading the list */ - public List readList(final Writeable.Reader reader) throws IOException { + public List readList(final BaseWriteable.Reader reader) throws IOException { return readCollection(reader, ArrayList::new, Collections.emptyList()); } @@ -1223,8 +1207,11 @@ public Set readSet(Writeable.Reader reader) throws IOException { /** * Reads a collection of objects */ - private > C readCollection(Writeable.Reader reader, IntFunction constructor, C empty) - throws IOException { + private > C readCollection( + BaseWriteable.Reader reader, + IntFunction constructor, + C empty + ) throws IOException { int count = readArraySize(); if (count == 0) { return empty; diff --git a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java index da17cddfcf97b..b0f4f6c8a6139 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java +++ b/server/src/main/java/org/opensearch/common/io/stream/StreamOutput.java @@ -51,18 +51,17 @@ import org.opensearch.common.Nullable; import org.opensearch.common.bytes.BytesArray; import org.opensearch.common.bytes.BytesReference; -import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.Writeable.Writer; import org.opensearch.common.settings.SecureString; import org.opensearch.common.text.Text; import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.common.io.stream.BaseStreamOutput; +import org.opensearch.core.common.io.stream.BaseWriteable; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; -import org.opensearch.script.JodaCompatibleZonedDateTime; import java.io.EOFException; import java.io.FileNotFoundException; import java.io.IOException; -import java.io.OutputStream; import java.math.BigInteger; import java.nio.file.AccessDeniedException; import java.nio.file.AtomicMoveNotSupportedException; @@ -103,7 +102,7 @@ * * @opensearch.internal */ -public abstract class StreamOutput extends OutputStream { +public abstract class StreamOutput extends BaseStreamOutput { private static final int MAX_NESTED_EXCEPTION_LEVEL = 100; @@ -663,10 +662,10 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep } } - private static final Map, Writer> WRITERS; + private static final Map, BaseWriteable.Writer> WRITERS; static { - Map, Writer> writers = new HashMap<>(); + Map, BaseWriteable.Writer> writers = new HashMap<>(); writers.put(String.class, (o, v) -> { o.writeByte((byte) 0); o.writeString((String) v); @@ -773,25 +772,12 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep o.writeByte((byte) 21); o.writeBytesRef((BytesRef) v); }); - writers.put(GeoPoint.class, (o, v) -> { - o.writeByte((byte) 22); - o.writeGeoPoint((GeoPoint) v); - }); writers.put(ZonedDateTime.class, (o, v) -> { o.writeByte((byte) 23); final ZonedDateTime zonedDateTime = (ZonedDateTime) v; o.writeString(zonedDateTime.getZone().getId()); o.writeLong(zonedDateTime.toInstant().toEpochMilli()); }); - writers.put(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()); - }); writers.put(Set.class, (o, v) -> { if (v instanceof LinkedHashSet) { o.writeByte((byte) 24); @@ -838,7 +824,12 @@ public void writeGenericValue(@Nullable Object value) throws IOException { return; } final Class type = getGenericType(value); - final Writer writer = WRITERS.get(type); + BaseWriteable.Writer writer = BaseWriteable.WriteableRegistry.getWriter(type); + if (writer == null) { + // fallback to this local hashmap + // todo: move all writers to the registry + writer = WRITERS.get(type); + } if (writer != null) { writer.write(this, value); } else { @@ -1145,14 +1136,6 @@ public void writeOptionalNamedWriteable(@Nullable NamedWriteable namedWriteable) } } - /** - * Writes the given {@link GeoPoint} to the stream - */ - public void writeGeoPoint(GeoPoint geoPoint) throws IOException { - writeDouble(geoPoint.lat()); - writeDouble(geoPoint.lon()); - } - /** * Write a {@linkplain DateTimeZone} to the stream. */ @@ -1193,7 +1176,7 @@ public void writeOptionalZoneId(@Nullable ZoneId timeZone) throws IOException { /** * Writes a collection to this stream. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection to write to this stream * @throws IOException if an I/O exception occurs writing the collection @@ -1224,7 +1207,7 @@ public void writeCollection(final Collection collection, final Writer /** * Writes a collection of a strings. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection of strings * @throws IOException if an I/O exception occurs writing the collection @@ -1235,7 +1218,7 @@ public void writeStringCollection(final Collection collection) throws IO /** * Writes an optional collection of a strings. The corresponding collection can be read from a stream input using - * {@link StreamInput#readList(Writeable.Reader)}. + * {@link StreamInput#readList(BaseWriteable.Reader)}. * * @param collection the collection of strings * @throws IOException if an I/O exception occurs writing the collection diff --git a/server/src/main/java/org/opensearch/common/io/stream/Writeable.java b/server/src/main/java/org/opensearch/common/io/stream/Writeable.java index 5fd227db6ca83..c04cd7977fdc0 100644 --- a/server/src/main/java/org/opensearch/common/io/stream/Writeable.java +++ b/server/src/main/java/org/opensearch/common/io/stream/Writeable.java @@ -32,6 +32,8 @@ package org.opensearch.common.io.stream; +import org.opensearch.core.common.io.stream.BaseWriteable; + import java.io.IOException; /** @@ -41,7 +43,7 @@ * * @opensearch.internal */ -public interface Writeable { +public interface Writeable extends BaseWriteable { /** * Write this into the {@linkplain StreamOutput}. @@ -64,17 +66,7 @@ public interface Writeable { * */ @FunctionalInterface - interface Writer { - - /** - * Write {@code V}-type {@code value} to the {@code out}put stream. - * - * @param out Output to write the {@code value} too - * @param value The value to add - */ - void write(StreamOutput out, V value) throws IOException; - - } + interface Writer extends BaseWriteable.Writer {} /** * Reference to a method that can read some object from a stream. By convention this is a constructor that takes @@ -91,15 +83,6 @@ interface Writer { * */ @FunctionalInterface - interface Reader { - - /** - * Read {@code V}-type value from a stream. - * - * @param in Input to read the value from - */ - V read(StreamInput in) throws IOException; - - } + interface Reader extends BaseWriteable.Reader {} } diff --git a/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java b/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java index e3ad2781f5546..bc52d8fe6a6df 100644 --- a/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java +++ b/server/src/main/java/org/opensearch/index/query/GeoDistanceQueryBuilder.java @@ -113,7 +113,7 @@ public GeoDistanceQueryBuilder(StreamInput in) throws IOException { fieldName = in.readString(); distance = in.readDouble(); validationMethod = GeoValidationMethod.readFromStream(in); - center = in.readGeoPoint(); + center = new GeoPoint(in); geoDistance = GeoDistance.readFromStream(in); ignoreUnmapped = in.readBoolean(); } @@ -123,7 +123,7 @@ protected void doWriteTo(StreamOutput out) throws IOException { out.writeString(fieldName); out.writeDouble(distance); validationMethod.writeTo(out); - out.writeGeoPoint(center); + center.writeTo(out); geoDistance.writeTo(out); out.writeBoolean(ignoreUnmapped); } diff --git a/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java b/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java index e0c7d44a08fac..f5e40fec78c25 100644 --- a/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java +++ b/server/src/main/java/org/opensearch/index/query/GeoPolygonQueryBuilder.java @@ -114,7 +114,7 @@ public GeoPolygonQueryBuilder(StreamInput in) throws IOException { int size = in.readVInt(); shell = new ArrayList<>(size); for (int i = 0; i < size; i++) { - shell.add(in.readGeoPoint()); + shell.add(new GeoPoint(in)); } validationMethod = GeoValidationMethod.readFromStream(in); ignoreUnmapped = in.readBoolean(); @@ -125,7 +125,7 @@ protected void doWriteTo(StreamOutput out) throws IOException { out.writeString(fieldName); out.writeVInt(shell.size()); for (GeoPoint point : shell) { - out.writeGeoPoint(point); + point.writeTo(out); } validationMethod.writeTo(out); out.writeBoolean(ignoreUnmapped); diff --git a/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java b/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java index 8f48da739359a..1359357dc1bc3 100644 --- a/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java +++ b/server/src/main/java/org/opensearch/script/JodaCompatibleZonedDateTime.java @@ -33,12 +33,18 @@ package org.opensearch.script; import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; import org.opensearch.common.SuppressForbidden; import org.opensearch.common.SuppressLoggerChecks; import org.opensearch.common.logging.DeprecationLogger; +import org.opensearch.common.io.stream.StreamInput; +import org.opensearch.common.io.stream.StreamOutput; +import org.opensearch.core.common.io.stream.BaseWriteable.Reader; +import org.opensearch.core.common.io.stream.BaseWriteable.Writer; import org.opensearch.common.time.DateFormatter; import org.opensearch.common.time.DateFormatters; import org.opensearch.common.time.DateUtils; +import org.opensearch.core.common.io.stream.BaseWriteable.WriteableRegistry; import java.security.AccessController; import java.security.PrivilegedAction; @@ -103,6 +109,26 @@ public JodaCompatibleZonedDateTime(Instant instant, ZoneId zone) { this.dt = ZonedDateTime.ofInstant(instant, zone); } + /** + * Register this type as a streamable so it can be serialized over the wire + */ + public static void registerStreamables() { + WriteableRegistry.>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()); + }); + WriteableRegistry.>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); + }); + } + // access the underlying ZonedDateTime public ZonedDateTime getZonedDateTime() { return dt; diff --git a/server/src/main/java/org/opensearch/transport/TransportService.java b/server/src/main/java/org/opensearch/transport/TransportService.java index f73cdd980b4e5..27493176465b8 100644 --- a/server/src/main/java/org/opensearch/transport/TransportService.java +++ b/server/src/main/java/org/opensearch/transport/TransportService.java @@ -46,6 +46,7 @@ import org.opensearch.common.Nullable; import org.opensearch.common.Strings; import org.opensearch.common.component.AbstractLifecycleComponent; +import org.opensearch.common.geo.GeoPoint; import org.opensearch.common.io.stream.StreamInput; import org.opensearch.common.io.stream.StreamOutput; import org.opensearch.common.io.stream.Writeable; @@ -63,6 +64,7 @@ import org.opensearch.common.util.io.IOUtils; import org.opensearch.node.NodeClosedException; import org.opensearch.node.ReportingService; +import org.opensearch.script.JodaCompatibleZonedDateTime; import org.opensearch.tasks.Task; import org.opensearch.tasks.TaskManager; import org.opensearch.threadpool.Scheduler; @@ -164,6 +166,14 @@ public boolean isClosed() { public void close() {} }; + static { + // registers server specific streamables + registerStreamables(); + } + + /** does nothing. easy way to ensure class is loaded */ + public static void ensureClassloaded() {} + /** * Build the service. * @@ -234,6 +244,15 @@ public TransportService( ); } + /** + * Registers server specific types as a streamables for serialization + * over the {@link StreamOutput} and {@link StreamInput} wire + */ + private static void registerStreamables() { + JodaCompatibleZonedDateTime.registerStreamables(); + GeoPoint.registerStreamables(); + } + public RemoteClusterService getRemoteClusterService() { return remoteClusterService; } diff --git a/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java b/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java index b92e59e43e0db..bd970be5e977d 100644 --- a/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java +++ b/server/src/test/java/org/opensearch/common/io/stream/BaseStreamTests.java @@ -40,12 +40,14 @@ import org.opensearch.common.bytes.BytesReference; import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.SecureString; +import org.opensearch.script.JodaCompatibleZonedDateTime; import org.opensearch.test.OpenSearchTestCase; import java.io.ByteArrayInputStream; import java.io.EOFException; import java.io.IOException; import java.time.Instant; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -396,6 +398,18 @@ public void testOptionalInstantSerialization() throws IOException { } } + public void testJodaDateTimeSerialization() throws IOException { + final BytesStreamOutput output = new BytesStreamOutput(); + long millis = randomIntBetween(0, Integer.MAX_VALUE); + JodaCompatibleZonedDateTime time = new JodaCompatibleZonedDateTime(Instant.ofEpochMilli(millis), ZoneOffset.ofHours(-7)); + output.writeGenericValue(time); + + final BytesReference bytesReference = output.bytes(); + final StreamInput input = getStreamInput(bytesReference); + Object inTime = input.readGenericValue(); + assertEquals(time, inTime); + } + static final class WriteableString implements Writeable { final String string; diff --git a/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java b/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java index bff35ad9fc975..36b84560bce41 100644 --- a/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java +++ b/server/src/test/java/org/opensearch/common/io/stream/BytesStreamsTests.java @@ -639,9 +639,9 @@ public void testReadWriteGeoPoint() throws IOException { try (BytesStreamOutput out = new BytesStreamOutput()) { GeoPoint geoPoint = new GeoPoint(randomDouble(), randomDouble()); - out.writeGeoPoint(geoPoint); + geoPoint.writeTo(out); StreamInput wrap = out.bytes().streamInput(); - GeoPoint point = wrap.readGeoPoint(); + GeoPoint point = new GeoPoint(wrap); assertEquals(point, geoPoint); } } diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java index 0ef44a5ae8db8..7722b59313b5f 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchTestCase.java @@ -128,6 +128,7 @@ import org.opensearch.test.junit.listeners.LoggingListener; import org.opensearch.test.junit.listeners.ReproduceInfoPrinter; import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; import org.opensearch.transport.nio.MockNioTransportPlugin; import org.joda.time.DateTimeZone; import org.junit.After; @@ -263,6 +264,7 @@ public void append(LogEvent event) { })); BootstrapForTesting.ensureInitialized(); + TransportService.ensureClassloaded(); // ensure server streamables are registered // filter out joda timezones that are deprecated for the java time migration List jodaTZIds = DateTimeZone.getAvailableIDs() From e888775f568ccb2b200a3c21916491b601e9af03 Mon Sep 17 00:00:00 2001 From: Nicholas Walter Knize Date: Wed, 5 Jul 2023 17:24:31 -0500 Subject: [PATCH 2/2] changelog Signed-off-by: Nicholas Walter Knize --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ab9c6b2d4166..1f133b71e1fad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -82,6 +82,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Check UTF16 string size before converting to String to avoid OOME ([#7963](https://github.com/opensearch-project/OpenSearch/pull/7963)) - [Refactor] remaining ImmutableOpenMap usage to j.u.Map and remove class ([#7309](https://github.com/opensearch-project/OpenSearch/pull/7309)) - [Refactor] Version and LegacyESVersion from server module to core lib ([#7328](https://github.com/opensearch-project/OpenSearch/pull/7328)) +- [Refactor] Stream Reader and Write Generics ([#7465](https://github.com/opensearch-project/OpenSearch/pull/7465)) - Move ZSTD compression codecs out of the sandbox ([#7908](https://github.com/opensearch-project/OpenSearch/pull/7908)) ### Deprecated