Struct BidiStream 

pub struct BidiStream<B, Req, RespView> { /* private fields */ }

A bidirectional streaming RPC in progress.

Returned from call_bidi_stream. Provides a send/close_send/message API modeled on connect-go’s BidiStreamForClient.

Half-duplex vs full-duplex

The Connect spec supports both. Half-duplex (send all, then receive all) works on HTTP/1.1 and HTTP/2. Full-duplex (interleaved send/receive) requires HTTP/2. This type does not distinguish — it’s the caller’s responsibility to respect the protocol in use. On HTTP/1.1, calling message() before close_send() will block until the request body is complete.

To drive the two sides from separate tasks, split the stream into independently owned halves with into_split().

Cancellation

Dropping the BidiStream cancels the call: any in-flight initialization task is aborted, which resets the underlying transport stream. Request messages accepted by send() but not yet transmitted may never reach the server — a caller that needs the request delivered must drive the call to completion via message() before dropping. Cancelling an individual message() future is safe and resumable — see message().

Example

let mut stream = call_bidi_stream(&transport, &config, "svc", "method", CallOptions::default()).await?;
stream.send(request1).await?;
stream.send(request2).await?;
stream.close_send();
// `Ok(None)` means a clean end; a failed RPC surfaces as `Err`,
// so `?` is the complete error handling.
while let Some(msg) = stream.message().await? {
    println!("got: {msg:?}");
}

Implementations

impl<B, Req, RespView> BidiStream<B, Req, RespView>

pub fn into_split(self) -> (BidiSendHalf<Req>, BidiRecvHalf<B, RespView>)

Split the stream into independently owned send and receive halves, so the two sides can be driven from separate tasks (full duplex).

Interleaved, response-dependent use — receiving an answer before sending the next message — requires an HTTP/2 transport, exactly as with an unsplit stream: on HTTP/1.1 no response arrives until the request body is complete, so a task waiting on the other half’s progress deadlocks. Prefer moving each half into its own spawned task (as below) over storing them in named struct fields — the halves’ full type parameters include the transport body type, which task-local inference names for you.

The halves are plain moves of the stream’s two sides — no locking is added — and there is no way to reassemble them. Semantics carried by each half:

• Dropping the BidiSendHalf (or calling close_send()) ends the request body cleanly; the RPC continues until the receive half finishes.
• Dropping the BidiRecvHalf cancels the RPC — as when dropping a whole BidiStream — after which sends on the other half fail.
• When send() fails because the server closed the stream, the server’s error is retrieved from the receive half via message().

Example

let (mut send, mut recv) = stream.into_split();
let reader = tokio::spawn(async move {
    while let Some(msg) = recv.message().await? {
        println!("got: {msg:?}");
    }
    Ok::<_, connectrpc::ConnectError>(())
});
for req in requests {
    send.send(req).await?;
}
send.close_send();
reader.await.expect("reader task")?;

impl<B, Req, RespView> BidiStream<B, Req, RespView>

where B: Body<Data = Bytes> + Send + Unpin, B::Error: Display, Req: Message + JsonSerialize, RespView: MessageView<'static> + Send, RespView::Owned: Message + JsonDeserialize,

pub async fn send(&mut self, msg: Req) -> Result<(), ConnectError>

Send a request message.

Errors

See BidiSendHalf::send for the error contract.

pub fn close_send(&mut self)

Close the send side of the stream. Idempotent. See BidiSendHalf::close_send.

pub async fn message<M>( &mut self, ) -> Result<Option<StreamMessage<M>>, ConnectError>

where B: 'static, RespView: MessageView<'static, Owned = M> + 'static, M: HasMessageView<View<'static> = RespView>,

Receive the next response message.

Errors

See BidiRecvHalf::message for the full contract.

pub fn headers(&self) -> Option<&HeaderMap>

Response headers. See BidiRecvHalf::headers.

pub fn trailers(&self) -> Option<&HeaderMap>

Trailing metadata. See BidiRecvHalf::trailers.

pub fn error(&self) -> Option<&ConnectError>

Terminal error that ended the stream, if any. See BidiRecvHalf::error.

Trait Implementations

impl<B, Req, RespView> Debug for BidiStream<B, Req, RespView>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.