Токио FramedRead.for_each вызывается на неопределенный срок для одного ответа - PullRequest
Новая
       73

Токио FramedRead.for_each вызывается на неопределенный срок для одного ответа

0 голосов
/ 06 апреля 2019

Я несколько недель шутил с Токио, пытаясь написать протокол с использованием tokio_uds. Существует несколько проблем со следующим кодом:

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)
        }
    }
}

1 Ответ

1 голос
/ 07 апреля 2019

Я видел, что вы решили свою другую проблему, и мне было бы очень интересно посмотреть, как вы решили эту проблему.Вот как я исправил это в моем проекте 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 здесь

...