Fix several streaming issues #106

pkwarren · 2023-09-20T20:47:16Z

Update conformance tests to add a new test exercising client side streaming, which exposed several issues in streaming call implementations.

The first issue only affected client streaming (it stopped attempting to read a response from the server once the send side closed - it should have stopped only if the receive side closed).

The second issue resulted from not calling close on the channel after the completion message was received, which lead to hangs consuming from resultChannel() (it would never complete). After this fix, both examples (for java and javalite) were updated and fixed to correctly exit when finished.

Additionally, several cleanups were made to the API (since the current API for client streaming was non-functional - it would only return the initial Headers result and not the message or completion result).

This should help resolve reported streaming issues like #100.

API Updates

`com.connectrpc.BidirectionalStreamInterface`

Removed

close()
- Use sendClose() instead. This may have confused callers that the close() method would close both send and receive sides of the connection when it was only closing the send side.

`com.connectrpc.ClientOnlyStreamInterface`

Added

sendClose()
- This shouldn't typically need to be called as receiveAndClose() already closes the send side of the stream.
isSendClosed()

Changed

receiveAndClose()
- Changed to return a ResponseMessage instead of a StreamResult. This allows callers to easily get access to the response as if they were calling a unary method. Previously, the StreamResult would only return the first result retrieved by the call, which typically was a Headers result (leaving callers unable to access the Message or Completion contents).

Removed

close()
- Replaced with sendClose().

`com.connectrpc.ServerOnlyStreamInterface`

Added

receiveClose()
isReceiveClosed()

Removed

close()
- This closed both the send and receive side of the stream (unlike in other interfaces which just closed the send side). If needed, callers should invoke receiveClose() instead (although this isn't necessary in normal use).
send()
- Callers should invoke sendAndClose() instead. Otherwise, reading results from resultChannel() will hang since the send side of the stream should be closed before reading responses.

`com.connectrpc.StreamResult`

Removed

Removed the error field from the base StreamResult class. It was never used by the Headers or Message subclasses and only used on the Complete type. This should make it easier for callers to use Headers and Message types since they don't need to worry about handling error.

Update conformance tests to add a new test exercising client side streaming, which exposed several issues in streaming call implementations. The first issue only affected client streaming (it stopped attempting to read a response from the server one the send side closed). The second issue resulted from not calling close on the channel after the completion message was received, which lead to hangs consuming from resultChannel() (it would never complete). After this fix, both examples (for java and javalite) were updated and fixed to correctly exit when finished. Additionally, several cleanups were made to the API (since the current API for client streaming was non-functional - it would only return the initial Headers result and not the message or completion result). The API updates are as follows: com.connectrpc.BidirectionalStreamInterface Removed: * close() * Use sendClose() instead. This may have confused callers that the close() method would close both send and receive sides of the connection when it was only closing the send side. com.connectrpc.ClientOnlyStreamInterface Added: * sendClose() * This shouldn't typically need to be called as receiveAndClose() already closes the send side of the stream. * isSendClosed() Changed: * receiveAndClose() * Changed to return a ResponseMessage instead of a StreamResult. This allows callers to easily get access to the response as if they were calling a unary response. Previously, the StreamResult would only return the first result retrieved by the call, which typically was a Headers result (leaving callers unable to access the Message or Completion contents). Removed: * close() * Replaced with sendClose(). com.connectrpc.ServerOnlyStreamInterface Added: * receiveClose() * isReceiveClosed() Removed: * close() * This closed both the send and receive side of the stream (unlike close() in other interfaces which just closed the send side). If needed, callers should invoke receiveClose() instead (although this isn't necessary in normal use). * send() * Callers should invoke sendAndClose() instead. Otherwise, reading results from resultChannel() will hang since the send side of the stream should be closed before reading responses. com.connectrpc.StreamResult: * Removed the 'error' field from the base StreamResult class. It was never set by Headers or Message StreamResult types and only used on the Complete type. This should make it easier for callers to handle Headers/Message result types since they don't need to worry about additional error handling.

pkwarren · 2023-09-20T20:47:59Z

conformance/google-java/src/test/kotlin/com/connectrpc/conformance/Conformance.kt

@@ -754,6 +750,94 @@ class Conformance(
        assertThat(countDownLatch.count).isZero()
    }

+    @Test
+    fun clientStreaming(): Unit = runBlocking {


New test for client streaming (used to find several issues with streaming calls and API definitions).

pkwarren · 2023-09-20T20:49:10Z

conformance/google-java/src/test/kotlin/com/connectrpc/conformance/Conformance.kt

+        val error: Throwable?,
+    )
+
+    /*


We should really revisit the APIs around streaming calls to make this sort of code unnecessary. For now however, this will at least allow us to easily consume all of the data from a server/bidi streaming response and perform assertions on headers, messages, trailers, and errors.

pkwarren · 2023-09-20T20:50:14Z

examples/buf.gen.yaml

@@ -9,10 +9,17 @@ plugins:
  - plugin: java
    out: generated-google-java/build/generated/sources/bufgen
    protoc_path: .tmp/bin/protoc
+  - plugin: kotlin


Generate kotlin code so our examples don't look so verbose.

pkwarren · 2023-09-20T20:52:20Z

examples/kotlin-google-java/src/main/kotlin/com/connectrpc/examples/kotlin/Main.kt

+                try {
+                    connectStreaming(elizaServiceClient)
+                } finally {
+                    okHttpClient.dispatcher.executorService.shutdown()


This is necessary for the client to cleanly shut down (the dispatcher thread pool uses non-daemon threads, so we need to shut it down manually for the app to shut down cleanly.

pkwarren · 2023-09-20T20:52:53Z

examples/kotlin-google-java/src/main/kotlin/com/connectrpc/examples/kotlin/Main.kt

            }
        }

        private suspend fun connectStreaming(elizaServiceClient: ElizaServiceClient) {
            val stream = elizaServiceClient.converse()
            withContext(Dispatchers.IO) {
                // Add the message the user is sending to the views.
-                stream.send(ConverseRequest.newBuilder().setSentence("hello").build())
+                stream.send(converseRequest { sentence = "hello" })
+                stream.sendClose()


If we don't close the bidi side of the stream, we won't ever complete.

pkwarren · 2023-09-20T20:53:19Z

examples/kotlin-google-java/src/main/kotlin/com/connectrpc/examples/kotlin/Main.kt

+                                if (connectErr != null) {
+                                    throw connectErr
+                                }
+                                throw ConnectError(code = result.code, metadata = result.trailers)


Added this to at least show how errors need to be handled today in the current API. In a follow up, I'll see if this can be improved.

Maybe just print the error details as a response?

I wanted to illustrate how you might want to handle errors in real code (instead of printing code/cause/trailers).

I think it also points out opportunities to make it even easier for consumers - if the Code is ever not OK, we should always have a ConnectError (even if it wraps another cause).

pkwarren · 2023-09-20T20:53:43Z

library/src/main/kotlin/com/connectrpc/ResponseMessage.kt

-    ) : ResponseMessage<Output>(code, headers, trailers)
+    ) : ResponseMessage<Output>(code, headers, trailers) {
+        override fun toString(): String {
+            return "Success{message=$message,code=$code,headers=$headers,trailers=$trailers}"


Added for easier debugging.

pkwarren · 2023-09-20T21:01:38Z

library/src/main/kotlin/com/connectrpc/impl/ProtocolClient.kt

@@ -211,6 +211,9 @@ class ProtocolClient(
                }
            }
            channel.send(result)
+            if (isComplete) {
+                channel.close()


This was a major issue - not closing the channel would lead to hangs in streaming clients trying to read results.

pkwarren · 2023-09-20T21:02:48Z

okhttp/src/main/kotlin/com/connectrpc/okhttp/OkHttpStream.kt

@@ -61,7 +61,7 @@ internal fun OkHttpClient.initializeStream(
    }
    val callRequest = builder.build()
    val call = newCall(callRequest)
-    call.enqueue(ResponseCallback(onResult, isSendClosed))
+    call.enqueue(ResponseCallback(onResult, isReceiveClosed))


This was the other big error for client streaming calls - this was using isSendClosed instead of isReceiveClosed, so we would stop trying to read messages from a stream once the send side was closed.

conformance/google-java/src/test/kotlin/com/connectrpc/conformance/Conformance.kt

rebello95 · 2023-09-21T22:54:09Z

examples/kotlin-google-java/src/main/kotlin/com/connectrpc/examples/kotlin/Main.kt

+                                if (connectErr != null) {
+                                    throw connectErr
+                                }
+                                throw ConnectError(code = result.code, metadata = result.trailers)


Maybe just print the error details as a response?

Co-authored-by: Michael Rebello <me@michaelrebello.com>

pkwarren requested review from rebello95 and smaye81 September 20, 2023 20:47

pkwarren added 2 commits September 20, 2023 15:49

cleanup unnecessary code

83f5a03

more cleanup

79f533b

pkwarren commented Sep 20, 2023

View reviewed changes

a few more tweaks

2797372

rebello95 approved these changes Sep 21, 2023

View reviewed changes

Apply suggestions from code review

26f4e9c

Co-authored-by: Michael Rebello <me@michaelrebello.com>

pkwarren requested a review from rebello95 September 21, 2023 23:44

rebello95 approved these changes Sep 21, 2023

View reviewed changes

pkwarren merged commit bb102a8 into main Sep 22, 2023

pkwarren deleted the pkw/streaming-fixes branch September 22, 2023 00:03

pkwarren mentioned this pull request Sep 22, 2023

Fix compilation failure #107

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix several streaming issues #106

Fix several streaming issues #106

pkwarren commented Sep 20, 2023 •

edited

Loading

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

rebello95 Sep 21, 2023

pkwarren Sep 21, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

pkwarren Sep 20, 2023

rebello95 Sep 21, 2023

Fix several streaming issues #106

Fix several streaming issues #106

Conversation

pkwarren commented Sep 20, 2023 • edited Loading

API Updates

com.connectrpc.BidirectionalStreamInterface

Removed

com.connectrpc.ClientOnlyStreamInterface

Added

Changed

Removed

com.connectrpc.ServerOnlyStreamInterface

Added

Removed

com.connectrpc.StreamResult

Removed

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pkwarren commented Sep 20, 2023 •

edited

Loading

`com.connectrpc.BidirectionalStreamInterface`

`com.connectrpc.ClientOnlyStreamInterface`

`com.connectrpc.ServerOnlyStreamInterface`

`com.connectrpc.StreamResult`