feat: udsink bidirectional streaming #141
Conversation
Signed-off-by: Yashash H L <[email protected]>
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #141 +/- ##
=======================================
Coverage ? 59.87%
Complexity ? 362
=======================================
Files ? 122
Lines ? 2472
Branches ? 175
=======================================
Hits ? 1480
Misses ? 863
Partials ? 129 ☔ View full report in Codecov by Sentry. |
| // make sure the handshake is done before processing the messages | ||
| if (!handshakeDone) { | ||
| if (!request.getHandshake().getSot()) { | ||
| responseObserver.onError(new Exception("Handshake request not received")); |
| if (request.hasHandshake() && request.getHandshake().getSot()) { | ||
| // make sure that the handshake is done before processing the ack requests | ||
| if (!handshakeDone) { | ||
| if (!request.getHandshake().getSot()) { |
There was a problem hiding this comment.
Should we keep the hasHandshake check? it could be null. Same for sink.
| responseObserver.onError(new Exception("Handshake request not received")); | ||
| return; | ||
| } | ||
| responseObserver.onNext(SinkOuterClass.SinkResponse.newBuilder() |
There was a problem hiding this comment.
In go sdk, we also set result status to success.
There was a problem hiding this comment.
we don't need to, I will change that in go sdk.
Signed-off-by: Yashash H L <[email protected]>
| @@ -113,11 +113,13 @@ public Client(String host, int port) { | |||
| * @return response from the server as a ResponseList | |||
| */ | |||
| public ResponseList sendRequest(DatumIterator datumIterator) { | |||
| CompletableFuture<SinkOuterClass.SinkResponse> future = new CompletableFuture<>(); | |||
There was a problem hiding this comment.
just curious, as to why SinkerTestKit lives in this package?
There was a problem hiding this comment.
this is for enabling users do "Component testing".
| responseListBuilder.addResponse(Response.responseOK(result.getId())); | ||
| } else if (result.getStatus() == SinkOuterClass.Status.FALLBACK) { | ||
| for (SinkOuterClass.SinkResponse result : outputResponses) { | ||
| if (result.getHandshake().getSot()) { |
There was a problem hiding this comment.
sot means start of the transmission. when set to true, it means the handshake is successful.
No description provided.