You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: basic_pub_bencher.rs
    path of file: ./repos/amqprs/benchmarks/src
    
    the code of the file until where you have to start completion: use criterion_bencher_compat::{benchmark_group, benchmark_main, Bencher};
mod common;
use common::*;

/// benchmark functions for `amqprs` client
mod client_amqprs {
    use super::{get_size_list, rt, setup_tracing, Bencher};
    use amqprs::{
        callbacks::{DefaultChannelCallback, DefaultConnectionCallback},
        channel::{
            BasicPublishArguments, QueueBindArguments, QueueDeclareArguments, QueuePurgeArguments,
        },
        connecti
    
    pub fn lapin_basic_pub(bencher: &mut Bencher) {
        setup_tracing();

        let rt = rt();

        let uri = "amqp://user:bitnami@localhost:5672";
        let options = ConnectionProperties::default()
            // Use tokio executor and reactor.
            // At the moment the reactor is only available for unix.
            .with_executor(Tokio::default().with_handle(rt.handle().clone()))
            .with_reactor(tokio_reactor_trait::Tokio);

        let (connection, channel) = rt.block_on(async {
            let connection = Connection::connect(uri, options).await.unwrap();
            let channel = connection.create_channel().await.unwrap();

            (connection, channel)
        });

        let rounting_key = "bench.lapin.pub";
        let exchange_name = "amq.topic";
        let queue_name = "bench-lapin-q";

        rt.block_on(async {
            channel
                .queue_declare(
                    queue_name,
                    QueueDeclareOptions::default(),
                    FieldTable::default(),
                )
                .await
                .unwrap();
            channel
                .queue_bind(
                    queue_name,
                    exchange_name,
                    rounting_key,
                    QueueBindOptions::default(),
                    FieldTable::default(),
                )
                .await
                .unwrap();
        });

        let pubopts = BasicPublishOptions::default();
        let declopts = QueueDeclareOptions {
            passive: true,
            ..Default::default()
        };

        let msg_size_list = get_size_list(connection.configuration().frame_max() as usize);

        let task = || async {
            let count = msg_size_list.len();
            // purge queue
            channel
                .queue_purge(queue_name, QueuePurgeOptions::default())
                .await
                .unwrap();
            let q_state = channel
                .queue_declare(queue_name, declopts, FieldTable::default())
                .await
                .unwrap();

            assert_eq!(0, q_state.message_count());
            // publish  messages of variable sizes
            for &i in msg_size_list.iter().take(count) {
                let _confirm = channel
                    .basic_publish(
                        exchange_name,
                        rounting_key,
                        pubopts,
                        &vec![0xc5; i],
                        BasicProperties::default(),
                    )
                    .await
                    .unwrap()
                    .await
                    .unwrap();
            }
            // check all messages arrived at queue
            loop {
                let q_state = channel
                    .queue_declare(queue_name, declopts, FieldTable::default())
                    .await
                    .unwrap();
                if count == q_state.message_count() as usize {
                    break;
                }
            }
        };
        // start benchmark
        bencher.iter(|| {
            rt.block_on(task());
        });

        rt.block_on(async {
            channel.close(0, "").await.unwrap();
            connection.close(0, "").await.unwrap();
        });
    }