szetszwo commented on code in PR #1481:
URL: https://github.com/apache/ratis/pull/1481#discussion_r3431974886
##########
ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java:
##########
Review Comment:
Since we have changed the decoder, we could start using
DataStreamReplyByteBuf here.
```java
public void channelRead(ChannelHandlerContext ctx, Object msg) {
if (!(msg instanceof DataStreamReplyByteBuf)) {
LOG.error("{}: unexpected message {}", name, msg.getClass());
return;
}
try (DataStreamReplyByteBuf reply = (DataStreamReplyByteBuf) msg) {
process(reply);
}
}
```
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java:
##########
@@ -48,13 +55,15 @@ public class ReadStreamManagement {
static class ReadStream implements WritableByteChannel {
private final ClientId clientId;
private final long streamId;
+ private final RaftClientRequest request;
Review Comment:
Store the terminalReply instead of the request:
```java
private final DataStreamReplyByteBuffer terminalReply;
private long streamOffset;
ReadStream(RaftClientRequest request, long streamId,
ChannelHandlerContext ctx) {
this.clientId = request.getClientId();
this.streamId = streamId;
this.ctx = ctx;
final RaftClientReply reply = RaftClientReply.newBuilder()
.setRequest(request)
.setSuccess()
.build();
this.terminalReply = DataStreamReplyByteBuffer.newBuilder()
.setClientId(clientId)
.setType(Type.STREAM_HEADER)
.setStreamId(streamId)
.setStreamOffset(0)
.setBuffer(toRaftClientReplyProto(reply).toByteString().asReadOnlyByteBuffer())
.setSuccess(true)
.setBytesWritten(0)
.build();
}
```
##########
ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamReply.java:
##########
@@ -22,12 +22,23 @@
import java.util.Collection;
-public interface DataStreamReply extends DataStreamPacket {
+public interface DataStreamReply extends DataStreamPacket, AutoCloseable {
boolean isSuccess();
long getBytesWritten();
/** @return the commit information when the reply is created. */
Collection<CommitInfoProto> getCommitInfos();
-}
\ No newline at end of file
+
+ /**
+ * Release resources owned by this reply.
+ */
+ default void release() {
+ }
+
+ @Override
+ default void close() {
+ release();
+ }
Review Comment:
Since DataStreamReply is a public API, we cannot add release() and close()
methods; otherwise it is an incompatible change.
--
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]
