Executing code after a stream is finished.

[frederikbosch]

Hello,

I have got the following proto

service ProjectDeployment {
  rpc InitializeDeployments(ClientInitializationRequest) returns (stream Deployment) {}
}

The Tonic service looks as follows.

#[tonic::async_trait]
impl ProjectDeployment for ProjectDeploymentService {
    type InitializeDeploymentsStream =
        Pin<Box<dyn Stream<Item = Result<RpcDeployment, Status>> + Send + Sync + 'static>>;

    async fn initialize_deployments(
        &self,
        request: Request<ClientInitializationRequest>,
    ) -> Result<Response<Self::InitializeDeploymentsStream>, Status> {
        let output = async_stream::try_stream! {
            // some code here
        }

        Ok(Response::new(Box::pin(output)
            as Pin<
                Box<dyn Stream<Item = Result<RpcDeployment, Status>> + Send + Sync + 'static>,
            >))
    }
}

Want I want is execute some code after all incoming streamed messages have been converted into a response per item. If I put it between letand Ok() it is not executed after the stream is finished. How should I achieve this?

[rahadbhuiya]

The reason code placed between let output = ... and Ok(...) executes immediately is because initialize_deployments only creates and returns the stream response wrapper; the actual stream has not been polled or consumed by the gRPC client yet at that point.

To execute code after all items in the stream have been emitted, you can place your code directly at the end of the try_stream! block.

The try_stream! generator runs lazily as the client polls for items. Any code placed after your message-yielding loop inside try_stream! will execute once all messages have finished streaming:

let output = async_stream::try_stream! {
    // 1. Stream your items to the client
    for item in items {
        // ...
        yield rpc_deployment;
    }

    // 2. This executes AFTER all items above have been yielded and consumed
    println!("Stream finished successfully, executing post-processing...");
    do_cleanup().await;
};

(Note: If you also need cleanup to run even when a client cancels or disconnects mid-stream before all items are sent, you can hold an RAII drop guard or send a completion signal via a tokio::sync::oneshot channel captured inside the stream).