Sending multiple response to a received request

[alishir]

It is not a bug report, it is a help request.
I want to implement simple network protocol like following diagram:

+--------+           +--------+
| Client |           | Server |
+---+----+           +---+----+
    |                    |
    |       PING         |
    |  +-------------->  |
    |                    |
    |                    |
    |       ACK          |
    |  <--------------+  |
    |                    |
    |                    |
    |                    |
    |                    |
    |       PONG         |
    |  <--------------+  |
    |                    |
    |                    |
    +                    +

I implemented simple server with simple codec that can decode ping messages, I can get the ping request and send the ack response, but I don't know how should I send the pong response.
I'm using netcat nc 127.0.0.1 6142 to interact with the server, server send ACK response well.
Here is the source code and repo:

extern crate bytes;
extern crate futures;
extern crate tokio;

use bytes::BufMut;
use bytes::BytesMut;
use tokio::codec::{Decoder, Encoder};
use tokio::io;
use tokio::net::TcpListener;
use tokio::prelude::*;

// Ping-ack-Pong Protocol Message
struct Ppm;

enum Msg {
    PING,
    PONG,
    ACK,
}

impl Encoder for Ppm {
    type Item = Msg;
    type Error = io::Error;

    fn encode(&mut self, item: Msg, dst: &mut BytesMut) -> io::Result<()> {
        let res = match item {
            Msg::PONG => "PONG",
            Msg::PING => "PING",
            Msg::ACK => "ACK",
        };
        dst.put(res);
        dst.put("\n");
        Ok(())
    }
}

impl Decoder for Ppm {
    type Item = Msg;
    type Error = io::Error;

    fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<Msg>> {
        match &src[..] {
            b"PING\n" => {
                src.clear();
                Ok(Some(Msg::PING))
            }
            _ => {
                src.clear();
                Ok(None)
            }
        }
    }
}

fn main() {
    let addr = "127.0.0.1:6142".parse().unwrap();
    let listener = TcpListener::bind(&addr).unwrap();
    let server = listener
        .incoming()
        .map_err(|_| println!("failed to accept socket"))
        .for_each(|socket| {
            println!("new connection");
            let (tx, rx) = Ppm.framed(socket).split();

            let task = rx
                .inspect(|_| {
                    println!("new message received");
                }).map(|msg| match msg {
                    Msg::PING => Msg::ACK,
                    _ => Msg::PING,
                }).forward(tx)
                .then(|_| Ok(()));

            tokio::spawn(task)
        });
    tokio::run(server);
}

[alishir]

I should thanks to @tobz for his kind helps, the solution is:

            let task = rx
                .inspect(|_| {
                    println!("new message received");
                }).map(|_| {
                    stream::iter_ok::<_, io::Error>(vec![Msg::ACK, Msg::PONG])
                }).flatten().forward(tx)
                .then(|_| Ok(()));

[alishir]

@carllerche this issue has been resolved, should I close it or I should make a doc for it?

[carllerche]

@alishir sorry for the delay, do you have thoughts on how the docs (Tokio guides) could be improved to cover this case?

[alishir]

@carllerche sorry for late reply, I think a cheat sheet that covers future to future or stream conversion and vise versa can solve the problem, during programming with tokio in many times I found that I should convert this future to other future or stream, or run multiple future together but I couldn't find any example.