szetszwo commented on code in PR #1481:
URL: https://github.com/apache/ratis/pull/1481#discussion_r3444615322
##########
ratis-client/src/main/java/org/apache/ratis/client/DataStreamClientRpc.java:
##########
@@ -36,4 +38,11 @@ default CompletableFuture<DataStreamReply>
streamAsync(DataStreamRequest request
throw new UnsupportedOperationException(getClass() + " does not support "
+ JavaUtils.getCurrentStackTraceElement().getMethodName());
}
+
+ /** Async call to send a request and receive multiple replies for the
request. */
+ default CompletableFuture<DataStreamReply> streamAsync(
+ DataStreamRequest request, DataStreamObserver<DataStreamReplyByteBuf>
replyHandler) {
+ throw new UnsupportedOperationException(getClass() + " does not support "
+ + JavaUtils.getCurrentStackTraceElement().getMethodName());
+ }
Review Comment:
We should use reference-counted:
```java
/**
* Async call to send a request to receive a stream of intermediate
replies and a final reply.
*
* @param request the request
* @param replyHandler to handle intermediate replies
* @return a future the final reply
*/
default CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest
request,
DataStreamObserver<ReferenceCountedObject<DataStreamReply>>
replyHandler) {
throw new UnsupportedOperationException(getClass() + " does not support "
+ JavaUtils.getCurrentStackTraceElement().getMethodName());
}
```
##########
ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java:
##########
@@ -236,6 +245,132 @@ private CompletableFuture<DataStreamReply>
sendForward(DataStreamReply writeRepl
}
}
+ public final class DataStreamInputImpl implements DataStreamInput {
+ private final RaftClientRequest header;
+ private final CompletableFuture<DataStreamReply> replyFuture;
+ private final Queue<DataStreamReply> replies = new ArrayDeque<>();
+ private final Queue<CompletableFuture<DataStreamReply>> pendingReads = new
ArrayDeque<>();
+ private Throwable readException;
+ private boolean endOfStream;
+ private boolean closed;
+
+ private DataStreamInputImpl(RaftClientRequest request) {
+ this.header = request;
+ final ByteBuffer buffer =
ClientProtoUtils.toRaftClientRequestProtoByteBuffer(header);
+ final DataStreamRequestHeader h = new
DataStreamRequestHeader(header.getClientId(), Type.STREAM_HEADER,
+ header.getCallId(), 0, buffer.remaining(),
StandardWriteOption.FLUSH, StandardWriteOption.CLOSE);
+ this.replyFuture = dataStreamClientRpc.streamAsync(new
DataStreamRequestByteBuffer(h, buffer),
+ new DataStreamObserver<DataStreamReplyByteBuf>() {
+ @Override
+ public void onNext(DataStreamReplyByteBuf reply) {
+ receive(reply.copy());
Review Comment:
We should use reference-counted in the queues and avoid copying the reply.
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
Review Comment:
This will release the reply even before user application reading it. We
should use reference-counted:
```java
final DataStreamReplyByteBuf reply = (DataStreamReplyByteBuf) msg;
final ReferenceCountedObject<DataStreamReply> ref =
ReferenceCountedObject.<DataStreamReply>newBuilder()
.setValue((DataStreamReplyByteBuf) msg)
.setReleaseMethod(r -> {
if (r != null) {
Preconditions.assertSame(reply, r, "reply");
reply.release();
}
}).build();
try (UncheckedAutoCloseableSupplier<DataStreamReply> ignored =
ref.retainAndReleaseOnClose()) {
process(ref);
}
```
##########
ratis-common/src/main/java/org/apache/ratis/datastream/DataStreamObserver.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.datastream;
+
+/** An interface similar to gRPC {@link
org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver}. */
+@FunctionalInterface
+public interface DataStreamObserver<V> {
+ void onNext(V value);
+
+ default void onError(Throwable t) {
+ }
+
+ default void onCompleted() {
+ }
Review Comment:
If we add them, we should enforce implementing them. Let's remove default.
```java
void onCompleted();
void onError(Throwable throwable);
```
##########
ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamInput.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.client.api;
+
+import org.apache.ratis.protocol.DataStreamReply;
+
+import java.io.Closeable;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * An asynchronous input stream supporting zero buffer copying.
+ */
+public interface DataStreamInput extends Closeable {
+ /**
+ * Read the next chunk in the stream asynchronously.
+ * The caller owns the returned {@link DataStreamReply} and should call
+ * {@link DataStreamReply#release()} after consuming it.
+ *
+ * @return a future of the reply.
+ */
+ CompletableFuture<DataStreamReply> readAsync();
Review Comment:
We should use reference-counted:
```java
/**
* Read the next chunk in the stream asynchronously.
* The caller owns the returned {@link DataStreamReply} which is a {@link
ReferenceCountedObject}.
* It must call {@link ReferenceCountedObject#release()} after consuming
it.
*
* @return a future of the reference-counted reply.
*/
CompletableFuture<ReferenceCountedObject<DataStreamReply>> readAsync();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
