Tokio FramedRead.for_each无限期地调用单个响应

Question

在使用tokio_uds编写协议的过程中，我一直在使用tokio几周。以下代码存在以下几个问题：

framed.for_each一次又一次地被召唤出来。

套接字只发送1个真实消息，但是Decoder尽可能多地解码完全相同的事件，直到它填满有界信道。

没有任何东西通过频道接收（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)
        }
    }
}

Answer 1

我看到你解决了你的另一个问题，我真的很想知道你是如何解决这个问题的。以下是我在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)
        }
    }
}

希望这可以帮助！

编辑：

根据我在博客文章中包含此解决方案后评论的锈子版本的用户，src.clear()对我来说可能是错误的答案。我应该使用`src.advance（14 + payload_len）

链接reddit评论here