File Descriptor Leak in Producer

[tgockel]

Describe the bug

When you create a Producer and drop it, the underlying connection remains open, even though there are no references to the producer.

Reproduction steps

This is most easily reproduced with a unit test and observation with ss. First, add this test (e.g.: test/producer_test.rs):

#[tokio::test(flavor = "multi_thread")]
async fn connect_and_drop() {
    let env = TestEnvironment::create().await;
    for count in 0..1000 {
        println!("{count}...");
        let producer = env.env.producer().build(&env.stream).await.unwrap();
        drop(producer);
        tokio::time::sleep(Duration::from_millis(50)).await;
    }

    println!("All connections dropped, waiting 10 seconds...");
    tokio::time::sleep(Duration::from_secs(10)).await;
}

1. Start the RabbitMQ server with make rabbitmq-server
2. Watch the connection list with sudo nsenter -t $(docker inspect -f '{{.State.Pid}}' rabbitmq-tls-test) watch ss -anut
3. Run cargo test -- --nocapture connect_and_drop
4. Watch the thing you created in step 2 for a bunch of tcp ESTAB connections to be added but never dropped

Expected behavior

What should happen here is, since we're immediately calling drop(producer), there should be a graceful termination of the TCP connection. Instead, the connection for these producers stay in the established state as long as the process is running. If you SIGINT the process, the connections on the server side will go into the TIME-WAIT as the connections are non-gracefully dropped.

Leaving the client running will eventually cause EBUSY, as there are too many open connections:

thread 'connect_and_drop' panicked at tests/./common.rs:143:64:
called `Result::unwrap()` on an `Err` value: Client(Io(Os { code: 16, kind: ResourceBusy, message: "Device or resource busy" }))

Additional context

This is probably not a huge deal, since most usage is connecting once and going from there.