szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3306458278
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -510,6 +558,72 @@ private void readImpl(DataStreamRequestByteBuf request,
ChannelHandlerContext ct
});
}
+ private static RaftClientRequest
toRaftClientRequest(DataStreamRequestByteBuf request) {
+ try {
+ return
ClientProtoUtils.toRaftClientRequest(RaftClientRequestProto.parseFrom(request.slice().nioBuffer()));
+ } catch (Throwable e) {
+ throw new CompletionException(e);
+ }
+ }
+
+ private CompletableFuture<Void>
submitReadOnlyRequest(DataStreamRequestByteBuf request,
+ RaftClientRequest raftClientRequest, ChannelHandlerContext ctx) {
+ try {
+ final StateMachine.DataChannel readOnlyDataStream = new
StateMachine.DataChannel() {
+ private long streamOffset;
+ private boolean closed;
+
+ @Override
+ public synchronized boolean isOpen() {
+ return !closed;
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ }
+
+ @Override
+ public synchronized void force(boolean metadata) throws IOException {
+ if (!isOpen()) {
+ throw new AlreadyClosedException("Channel closed at offset " +
streamOffset);
+ }
+ ctx.flush();
+ }
+
+ @Override
+ public synchronized int write(ByteBuffer buffer) throws IOException {
+ if (!isOpen()) {
+ throw new AlreadyClosedException("Channel closed at offset " +
streamOffset);
+ }
+ final int length = buffer.remaining();
+ final DataStreamReplyByteBuffer reply =
newDataStreamReadOnlyReplyByteBuffer(request, streamOffset, buffer);
+ final ChannelFuture future = ctx.writeAndFlush(reply);
+ try {
+ future.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException(
+ "Interrupted while writing " + length + " bytes at offset " +
streamOffset);
+ }
+ if (!future.isSuccess()) {
+ final Throwable cause = future.cause();
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
Review Comment:
Let's create a new IOException in both cases since it can add the offset to
the message.
##########
ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java:
##########
@@ -150,6 +153,18 @@ default RaftGroup getGroup() {
/** @return the data stream rpc service. */
DataStreamServerRpc getDataStreamServerRpc();
+ /**
+ * Submit a read-only request whose response may be streamed through the
data stream RPC.
+ *
+ * @param request the read-only request
+ * @param stream the stream for response data chunks
+ * @return a future for the terminal reply
+ */
+ default CompletableFuture<RaftClientReply> streamReadOnlyAsync(
+ RaftClientRequest request, StateMachine.DataChannel stream) throws
IOException {
+ throw new UnsupportedOperationException("This method is NOT supported.");
+ }
Review Comment:
This new method seems not needed since we may:
- Phase 1: Directly call DataApi.streamReadOnly(..) and ignore all
linearizable checks.
- Phase 2: Reuse RaftClientAsynchronousProtocol.submitClientRequestAsync(..)
to submit a dummy read request for linearizable checks and then call
DataApi.streamReadOnly(..).
Of course, we should start with Phase 1 for simpilcity.
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream>
stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write
zero or more data
+ * chunks before completing the returned future with the terminal reply
message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest
request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except that
* {@link #query(Message)} returns the result in a future
* while this method sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the results
*/
default void query(Message request, DataChannel stream) {
}
```
--
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]
