How to test a bidirectional streaming server

[CodedOre]

So, I am currently working on a subscription service. It takes a stream, from which requests for subscription creation and updates are delivered, and delivers updates as responses, for as long as the request stream is kept open.

The service implementation looks similar to this:

pub struct SubscriptionService {
    // Internals
}

impl SubscriptionApi for SubscriptionService {
    type SubscribeStream =
        Pin<Box<dyn Stream<Item = Result<SubscriptionResponse, Status>> + Send + 'static>>;

    async fn subscribe(
        &self,
        request: Request<Streaming<SubscriptionRequest>>,
    ) -> Result<Response<Self::SubscribeStream>, Status> {
        // Business Logic
    }
}

Now, I wanted to implement a few tests to ensure the functionality of the service.
The general structure of the test is supposed to be as follows: I first create the service and it's internal structures. Then I run the first request, check the response, but also the internal systems. Then I close the stream and check the proper cleanup in the internal system.

Therefore, the test needs to be able to send requests in intervalls. As I see it, the logical way would be to use a tokio_stream::wrappers::ReceiverStream. So, this is the structure I have in mind:

#[cfg(test)]
mod tests {
    fn setup_service() -> (SubscriptionService, InternalSystem);

    #[futures_test::test]
    async fn test_subscription() {
        let (api, internal) = setup_service();

        let (request_tx, request_rx) = mpsc::channel(1);
        let request_stream = ReceiverStream::new(request_rx);
        let request = Request::new(request_stream);

        let response_stream = api.subscribe(request).await;

        let request = SubscriptionRequest{
            // ...
        };
        request_tx
            .send(request)
            .await
            .expect("internal: failed to send request to service");

        let response = response_stream.message().await;
        // Check response

        // Check internals

        drop(request_tx);

        // Check internals
    }
}

However, this is not possible, as it leads to an compile issue:

error[E0308]: mismatched types
   --> server/src/data/service/v2.rs:484:23
    |
484 |         api.subscribe(request).await;
    |             --------- ^^^^^^^ expected `Request<Streaming<SubscriptionRequest>>`, found `Request<ReceiverStream<SubscriptionRequest>>`
    |             |
    |             arguments to this method are incorrect
    |
    = note: expected struct `tonic::Request<Streaming<SubscriptionRequest>>`
               found struct `tonic::Request<tokio_stream::wrappers::ReceiverStream<SubscriptionRequest>>`

I have looked around to find a solution. So far I have seen that you can convert for example an tokio_stream::iter into a streaming request. However, this doesn't seem to be a viable solution in my case. As far as I understand, a Stream like tokio_stream::iter would be closed once all items are processed, which would in turn close the service.
In order for all tests to be done, I need to be able to control when the request stream is closed, and run requests and response checks in a specific order.

So, is there a good way to implement such a test?