How to correctly drain replies after unsubscribing?

[davide-scalzo]

TLDR: I have a test service that I am trying to wrap in a graceful shutdown wrapper. What I'm seeing is that after unsubscribing the last reply does not get sent the counterparty service that issued the request.

The wrapper spawns an application_task that requires a tokio TaskTracker and a shutdown signal. Once the shutdown signal is received (e.g. ctrl+c) the task tracker gets closed and awaited in order to complete any pending tasks (requests) handled by the application task.

The content of the application_task looks like this:

async fn application_task(tracker: TaskTracker, mut shutdown_rx: mpsc::Receiver<()>) {
    let client = async_nats::ConnectOptions::new()
        .connect("nats://localhost:4222")
        .await
        .unwrap();

    let mut requests = client
        .subscribe("foo")
        .await
        .expect("Could not subscribe to user.request.*");

    loop {
        tokio::select! {
            Some(msg) = requests.next() => {
                let client = client.clone();
                tracker.spawn(async move {
                    println!("start handling task");
                    let _ = dummy_handler().await;
                    println!("finish handling task");
                    client
                        .publish(msg.reply.unwrap(), Vec::new().into()) // dummy payload
                        .await
                        .unwrap();
                    println!("Client Status: {:?}", client.connection_state());
                    println!("Published response");
                });
            }
            _ = shutdown_rx.recv() => {
                println!("Unsubscribing");
                let _ = requests.unsubscribe().await;
                println!("Unsubscribed");
                break;
            }
        }
    }
}

The logs look like this:

start handling task <-- request 1
start handling task <-- request 2
Received shutdown signal <- shutdown signal sent
Unsubscribing
Unsubscribed
finish handling task
Client Status: Connected
Published response <-- this reply is received correctly by the counterparty
finish handling task
Client Status: Connected <-- this reply is never gets received on the NATS server and therefore the counterparty
Published response
Shutdown complete

Any suggestion, on what I'm doing wrong / how to handle un-subscription correctly?

[caspervonb]

Try to flush either after publishing, or at-atleast before shutting down. Client publish goes into a write buffer.