How to write the subscribed bytes back to the client via Tcpstream

[SudoAptGetYou]

I'm learning about Rust,and I tried to use mqtt as a proxy.I can already send the client's data to mqtt,but I don’t know how to write the subscribed bytes back to the client and keep the TCP connection continuously.

Code is:

let url = "tcp://localhost:1883";
let username = "username";
let password = "password";
let topic_pub = "topicpub";
let topic_sub = "topicsub";
let client_id = "test";
let addr = "127.0.0.1:1234".to_string();
let listener = TcpListener::bind(&addr).await?;
let create_opts = mqtt::CreateOptionsBuilder::new()
            .server_uri(url)
            .client_id(client_id)
            .finalize();

let mut m_cli = mqtt::Client::new(None).unwrap();
let conn_opts = mqtt::ConnectOptionsBuilder::new()
        .clean_session(true)
        .user_name(username)
        .password(password)
        .finalize();
if let Err(e) = m_cli.connect(conn_opts) {
    println!("Unable to connect:\n\t{:?}", e);
    process::exit(1);
    } else {
    println!("Connect mqtt success!"); 
}
let requests = m_cli.start_consuming();
m_cli.subscribe(topic_sub, QOS).unwrap();

let m_cli = Arc::new(Mutex::new(m_cli)); 

loop {
    let (socket, addr) = listener.accept().await?;
    println!("New connection from: {}", addr);
    let (r, w) = socket.into_split();
    // let mut sink = FramedWrite::new(w, BytesCodec::new());
    let mut stream = FramedRead::new(r, BytesCodec::new());
    let client_clone = m_cli.clone();
    tokio::spawn(async move {
        while let Some(message) = stream.next().await {
            match message {
                Ok(bytes) => {
                    println!("Recieved bytes...");
                    let bytes = bytes.to_vec();
                    println!("read bytes: {:?}", bytes.len());
                    let msg = gen_publish_message(bytes, topic_pub);
                    if let Err(e) = client_clone.lock().unwrap().publish(msg) {
                        println!("Unable to publish messager:\n\t{:?}", e);
                    }
                },
                Err(err) => println!("Error: {:?}", err),
            }
        }
        println!("Socket received FIN packet and closed connection");
    });
}

I tried to use FramedWrite, but I don’t know how to start...

[Darksonn]

You can split a TcpStream into a read and write half using the into_split or split methods.

[SudoAptGetYou]

I have used into_split function to achieve reading module, but I am a little confused about the writer document, and I don’t know how to implement it.

[Darksonn]

You want to forward messages received from mqtt? I would spawn a task that reads messages from mqtt in a loop, which then sends that message on a broadcast channel. Then, every connection should also spawn a task, and that task should loop receiving messages from the broadcast channel, and write them to the TcpStream write half.