GrpcStreamingCall client buffers packets and flushes only after timeout — not receiving events in real-time

[ShambhiReddyK]

I'm using com.squareup.wire.GrpcStreamingCall along with OkHttp in a Kotlin mobile app to stream gRPC events from a backend service. The issue I'm facing is that the client does not receive the streamed packets in real-time.

Instead, all packets are buffered internally and delivered together only when the server closes the connection or times out the session (e.g., after 5 minutes). This defeats the purpose of a streaming connection.

There is no any problem from backend side, API's are working fine in Postman.

Expected Behavior

• Packets should be delivered immediately to the client as the server sends them (real-time).
• The GrpcStreamingCall.response flow should emit each message as it arrives on the wire.

Actual Behavior

• Packets are only emitted from the flow after several minutes — when the session is closed/timed out.
• All buffered packets arrive together at once, as if they were delayed or cached.

Please help me out in fixing this problem

[swankjesse]

Yikes! That doesn’t seem to be a problem for our other users; I wonder why it’s breaking for you in particular.

Would you consider putting a breakpoint in GrpcMessageSource? We can use that to determine whether the problem is in Wire, or if something lower-level.

One other thing to check - you definitely need to close the outbound message sink when you’ve sent the last message, even if you’re sending only one message. This has tripped people up before.

[ShambhiReddyK]

Hi @swankjesse ,

Thanks for the reply. I have kept the breakpoint in GrpcMessageSource, it is reaching breakpoint after 5 minutes of session timeout only, not instantly. But when I verify the same in postman, client API is instantly receiving the data. Not sure where the problem is

I am sharing my code also here. Please check and update

1. Grpc Client Builder

val okHttpClient = OkHttpClient.Builder()
			.connectTimeout(30, TimeUnit.SECONDS)
			.readTimeout(0, TimeUnit.SECONDS) 
			.writeTimeout(30, TimeUnit.SECONDS)
			.addInterceptor(customHeadersInterceptor)
			.addInterceptor(httpToFileInterceptor)
			.addInterceptor(loggingInterceptor)
			.build()

GrpcClient.Builder()
			.client(okHttpClient)
			.baseUrl(baseUrl)
			.minMessageToCompress(Long.MAX_VALUE)
			.build()

2. Stream Events Implementation

override suspend fun startStreamingEvents(streamEventsRequest: StreamEventsRequest): Flow<StreamPacket> = callbackFlow {
		
		val streaming = call.authenticatedStream {  grpcStreamingClient.StreamEvents() }

		// Send the request
		streaming.request.send(streamEventsRequest)

		//Close the request stream!
		streaming.request.close() 

		val job = launch {
			try {
				for (packet in streaming.response) {
					send(packet)
				}
			} catch (e: Exception) {
				Timber.e(e, "Error while receiving packets")
				close(e)
			}
		}

		awaitClose {
			Timber.d("Stream closed")
			job.cancel()
		}
	}

3. Supporting code

suspend fun <Req : Any, Res : Any> authenticatedStream(call: () -> GrpcStreamingCall<Req, Res>): StreamingCallResult<Req, Res> {
		val (request, response) = call().apply {
			requestMetadata = headerProvider.getAuthHeaders()
		}.executeIn(CoroutineScope(dispatchers.io))
		return StreamingCallResult(request, response)
	}

data class StreamingCallResult<Req : Any, Res : Any>(
	val request: SendChannel<Req>,
	val response: ReceiveChannel<Res>
)

Could you please let me know where am I going wrong and point me to the right direction ?

[ShambhiReddyK]

Hi @swankjesse ,

Could you please check my code and revert back ? Where I am going wrong ?

Need your inputs. Thanks for the help