Avoid race in duplex pipe for streaming calls

[oldergod]

Fixes #3370

From Claude

Root Cause

GrpcServerStreamingCall was implemented by delegating to GrpcStreamingCall, which uses a PipeDuplexRequestBody, an asynchronous pipe where the request body is written by a background coroutine on Dispatchers.IO. This means the request message and the HTTP/2 END_STREAM flag are sent asynchronously, after the HTTP call is already enqueued.
Many gRPC server implementations wait for the client's half-close (END_STREAM) before starting to stream responses. With the duplex pipe approach, there's a race between when the server receives END_STREAM and when it decides to time out the idle stream, causing the reported ~5-minute delay.

Fix

RealGrpcServerStreamingCall now has its own independent implementation using newRequestBody() (a non-duplex request body), identical to how RealGrpcCall works. The complete request message + END_STREAM is written synchronously in OkHttp's writeTo() call, ensuring the server receives the full request before any response reading begins.
The old delegation-based class is preserved as GrpcStreamingCallServerStreamingAdapter, still used by the test-double factory (GrpcServerStreamingCall { ... } in GrpcCalls.kt).

Supporting changes:

• readFromResponseBodyCallback() in grpc.kt: added an overload accepting onResponseMetadata: (Map<String, String>) -> Unit so both RealGrpcStreamingCall and RealGrpcServerStreamingCall can use it
• BlockingMessageSource: replaced grpcCall: RealGrpcStreamingCall with onResponseMetadata: (Map<String, String>) -> Unit for the same reason
• GrpcClient.newServerStreamingCall() — now creates RealGrpcServerStreamingCall directly instead of wrapping a streaming call.

[oldergod]

Does it look legit to you @swankjesse ?