forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Refactor] Stream Reader and Write Generics (opensearch-project#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 <[email protected]>
- Loading branch information
Showing
19 changed files
with
310 additions
and
94 deletions.
There are no files selected for viewing
19 changes: 19 additions & 0 deletions
19
libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamInput.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,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 {} |
19 changes: 19 additions & 0 deletions
19
libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseStreamOutput.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,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 {} |
130 changes: 130 additions & 0 deletions
130
libs/core/src/main/java/org/opensearch/core/common/io/stream/BaseWriteable.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,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<S extends BaseStreamOutput> { | ||
| /** | ||
| * 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<Class<?>, Writer<? extends BaseStreamOutput, ?>> WRITER_REGISTRY = new ConcurrentHashMap<>(); | ||
| private static final Map<Byte, Reader<? extends BaseStreamInput, ?>> READER_REGISTRY = new ConcurrentHashMap<>(); | ||
|
|
||
| /** | ||
| * registers a streamable writer | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public static <W extends Writer<? extends BaseStreamOutput, ?>> 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 <R extends Reader<? extends BaseStreamInput, ?>> 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 extends Writer<? extends BaseStreamOutput, ?>> 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 extends Reader<? extends BaseStreamInput, ?>> 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}. | ||
| * <p> | ||
| * 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! | ||
| * <p> | ||
| * Most classes should implement {@code Writeable} and the {@code Writeable#writeTo(BaseStreamOutput)} method should <em>use</em> | ||
| * {@link BaseStreamOutput} methods directly or this indirectly: | ||
| * <pre><code> | ||
| * public void writeTo(StreamOutput out) throws IOException { | ||
| * out.writeVInt(someValue); | ||
| * out.writeMapOfLists(someMap, StreamOutput::writeString, StreamOutput::writeString); | ||
| * } | ||
| * </code></pre> | ||
| */ | ||
| @FunctionalInterface | ||
| interface Writer<S extends BaseStreamOutput, V> { | ||
|
|
||
| /** | ||
| * 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)}. | ||
| * <p> | ||
| * 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: | ||
| * <pre><code> | ||
| * public MyClass(final StreamInput in) throws IOException { | ||
| * this.someValue = in.readVInt(); | ||
| * this.someMap = in.readMapOfLists(StreamInput::readString, StreamInput::readString); | ||
| * } | ||
| * </code></pre> | ||
| */ | ||
| @FunctionalInterface | ||
| interface Reader<S extends BaseStreamInput, V> { | ||
|
|
||
| /** | ||
| * Read {@code V}-type value from a stream. | ||
| * | ||
| * @param in Input to read the value from | ||
| */ | ||
| V read(final S in) throws IOException; | ||
| } | ||
| } |
9 changes: 9 additions & 0 deletions
9
libs/core/src/main/java/org/opensearch/core/common/io/stream/package-info.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,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; |
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
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.