unix - Tokio FramedRead.for_each为单个响应无限期调用

为了使用tokio_uds编写协议(protocol)，我已经与tokio混为一谈了。以下代码存在多个问题:

从单个响应中反复调用framed.for_each。

套接字仅发送1条真实消息，但是Decoder尽可能多地解码完全相同的事件，直到它填满有边界的 channel 为止。

尽管在 channel 填满之前它似乎一直被写入，但在该 channel 上什么也没收到(rx.for_each从不打印任何内容)。

我需要使用UnixStream而不是UnixListener，因为必须先将一些数据放在套接字上才能“订阅”服务，并让其知道要发送的内容。

use byteorder::{ByteOrder, LittleEndian};
use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
use futures::prelude::*;
use futures::sync::mpsc::{self, Receiver, Sender};
use futures::Stream;
use tokio::prelude::*;
use tokio_codec::{Decoder, Encoder, FramedRead};
use tokio_uds::UnixStream;

fn subscribe(tx: Sender<event::Evt>, events: Vec<Event>) -> io::Result<()> {
    let fut = UnixStream::connect(socket_path()?)
        .and_then(move |stream| {
            // some setup
            tokio::io::write_all(stream, buf)
        })
        .and_then(|(stream, _buf)| {
            let buf = [0_u8; 30]; // <i3-ipc (6 bytes)><len (4 bytes)><type (4 bytes)><{success:true} 16 bytes>
            tokio::io::read_exact(stream, buf)
        })
        .and_then(|(stream, initial)| {
            if &initial[0..6] != MAGIC.as_bytes() {
                panic!("Magic str not received");
            }
            // decoding initial response and returning stream
            future::ok(stream)
        })
        .and_then(move |stream| {
            let framed = FramedRead::new(stream, EvtCodec);
            let sender = framed
                .for_each(move |evt| {
                    let tx = tx.clone();
                    tx.send(evt).wait(); // this line is called continuously until buffer fills
                    Ok(())
                })
                .map_err(|err| println!("{}", err));
            tokio::spawn(sender);
            Ok(())
        })
        .map(|_| ())
        .map_err(|e| eprintln!("{:?}", e));

    tokio::run(fut);
    Ok(())
}

    fn test_sub() -> io::Result<()> {
        let (tx, rx) = mpsc::channel(5);
        subscribe(tx, vec![Event::Window])?;
        let fut = rx.for_each(|e: event::Evt| {
            println!("received"); // never reaches
            future::ok(())
        });
        tokio::spawn(fut);
        Ok(())
    }

我的Decoder:

pub struct EvtCodec;

/// decoding: "<i3-ipc><payload len: u32><msg type: u32><payload>"
impl Decoder for EvtCodec {
    type Item = event::Evt;
    type Error = io::Error;
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, io::Error> {
        if src.len() > 14 {
            if &src[0..6] != MAGIC.as_bytes() {
                return Err(io::Error::new(
                    io::ErrorKind::Other,
                    format!("Expected 'i3-ipc' but received: {:?}", &src[0..6]),
                ));
            }
            let payload_len = LittleEndian::read_u32(&src[6..10]) as usize;
            let evt_type = LittleEndian::read_u32(&src[10..14]);
            dbg!(&src.len()); // 878
            dbg!(payload_len); // 864
            if src.len() < 14 + payload_len {
                Ok(None)
            } else {
                let evt = decode_evt(evt_type, src[14..].as_mut().to_vec())?;
                dbg!(&evt); // correctly prints out a well-formed event
                Ok(Some(evt))
            }
        } else {
            Ok(None)
        }
    }
}

最佳答案

我们看到您已经解决了其他问题，并且非常有兴趣了解您如何解决此问题。这是我在TCP Tokio端项目上修复它的方式:

use byteorder::{ByteOrder, LittleEndian};
use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf};
use futures::prelude::*;
use futures::sync::mpsc::{self, Receiver, Sender};
use futures::Stream;
use tokio::prelude::*;
use tokio_codec::{Decoder, Encoder, FramedRead};
use tokio_uds::UnixStream;

fn subscribe(tx: Sender<event::Evt>, rx: Receiver<event::Evt>, events: Vec<Event>) -> io::Result<()> {
    let fut = UnixStream::connect(socket_path()?)
        .and_then(move |stream| {
            // some setup
            tokio::io::write_all(stream, buf)
        })
        .and_then(|(stream, _buf)| {
            let buf = [0_u8; 30]; // <i3-ipc (6 bytes)><len (4 bytes)><type (4 bytes)><{success:true} 16 bytes>
            tokio::io::read_exact(stream, buf)
        })
        .and_then(|(stream, initial)| {
            if &initial[0..6] != MAGIC.as_bytes() {
                panic!("Magic str not received");
            }
            // decoding initial response and returning stream
            future::ok(stream)
        })
        .and_then(move |stream| {
            let framed = FramedRead::new(stream, EvtCodec);
            let (writer, reader) = framed.split();

            // Connect your framed reader to the channel
            let sink = rx.forward(writer.sink_map_err(|_| ()));
            tokio::spawn(sink.map(|_| ()));

            let sender = reader
                .for_each(move |evt| {
                    let tx = tx.clone();
                    tx.send(evt).wait(); // this line is called continuously until buffer fills
                    Ok(())
                })
                .map_err(|err| println!("{}", err));
            tokio::spawn(sender);
            Ok(())
        })
        .map(|_| ())
        .map_err(|e| eprintln!("{:?}", e));

    tokio::run(fut);
    Ok(())
}

    fn test_sub() -> io::Result<()> {
        let (tx, rx) = mpsc::channel(5);
        subscribe(tx, rx, vec![Event::Window])?;
        let fut = rx.for_each(|e: event::Evt| {
            println!("received"); // never reaches
            future::ok(())
        });
        tokio::spawn(fut);
        Ok(())
    }

然后将带有缓冲区的解码器清除:

pub struct EvtCodec;

/// decoding: "<i3-ipc><payload len: u32><msg type: u32><payload>"
impl Decoder for EvtCodec {
    type Item = event::Evt;
    type Error = io::Error;
    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, io::Error> {
        if src.len() > 14 {
            if &src[0..6] != MAGIC.as_bytes() {
                return Err(io::Error::new(
                    io::ErrorKind::Other,
                    format!("Expected 'i3-ipc' but received: {:?}", &src[0..6]),
                ));
            }
            let payload_len = LittleEndian::read_u32(&src[6..10]) as usize;
            let evt_type = LittleEndian::read_u32(&src[10..14]);
            dbg!(&src.len()); // 878
            dbg!(payload_len); // 864
            if src.len() < 14 + payload_len {
                Ok(None)
            } else {
                let evt = decode_evt(evt_type, src[14..].as_mut().to_vec())?;
                dbg!(&evt); // correctly prints out a well-formed event
                src.clear(); // Clears the buffer, so you don't have to keep decoding the same packet over and over.
                Ok(Some(evt))
            }
        } else {
            Ok(None)
        }
    }
}

希望这可以帮助!

编辑:

在我在博客文章中包含此解决方案后，对rust subreddit进行评论的用户表示，src.clear()可能对我来说是错误的答案。我应该改用`src.advance(14 + payload_len)

链接reddit评论here

关于unix - Tokio FramedRead.for_each为单个响应无限期调用，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/55552090/