mprocs 0.9.6

TUI for running multiple processes
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

use std::marker::PhantomData;

use bytes::{Buf, BytesMut};
use futures::StreamExt;
use serde::de::DeserializeOwned;
use tokio::io::AsyncRead;
use tokio_util::codec::FramedRead;

struct MsgDecoder<T: DeserializeOwned> {
  state: DecoderState,
  t: PhantomData<T>,
}

enum DecoderState {
  Header,
  Data(usize),
}

impl<T: DeserializeOwned> MsgDecoder<T> {
  pub fn new() -> MsgDecoder<T> {
    MsgDecoder {
      state: DecoderState::Header,
      t: PhantomData,
    }
  }
}

impl<T: DeserializeOwned> tokio_util::codec::Decoder for MsgDecoder<T> {
  type Item = T;

  type Error = bincode::Error;

  fn decode(
    &mut self,
    src: &mut BytesMut,
  ) -> Result<Option<Self::Item>, Self::Error> {
    loop {
      let len = match self.state {
        DecoderState::Header => {
          if src.len() < 4 {
            return Ok(None);
          }
          let len = src.get_u32() as usize;
          self.state = DecoderState::Data(len);
          len
        }
        DecoderState::Data(len) => len,
      };

      if src.len() < len {
        return Ok(None);
      }

      let frame = src.split_to(len);
      self.state = DecoderState::Header;

      match bincode::deserialize::<T>(&frame) {
        Ok(msg) => return Ok(Some(msg)),
        Err(err) => {
          log::warn!("ipc: skipping undecodable frame ({len} bytes): {err}");
          continue;
        }
      }
    }
  }
}

type DynRead = dyn AsyncRead + Unpin + Send + 'static;

pub struct MsgReceiver<T: DeserializeOwned> {
  reader: FramedRead<Box<DynRead>, MsgDecoder<T>>,
}

impl<T: DeserializeOwned + Send + 'static> MsgReceiver<T> {
  pub fn new<R: AsyncRead + Unpin + Send + 'static>(read: R) -> Self {
    let reader = tokio_util::codec::FramedRead::new(
      Box::new(read) as Box<DynRead>,
      MsgDecoder::<T>::new(),
    );

    MsgReceiver { reader }
  }
}

impl<T: DeserializeOwned> MsgReceiver<T> {
  pub async fn recv(&mut self) -> Option<Result<T, bincode::Error>> {
    self.reader.next().await
  }
}

#[cfg(test)]
mod tests {
  use bytes::{BufMut, BytesMut};
  use tokio_util::codec::Decoder;

  use super::MsgDecoder;

  fn frame(payload: &[u8]) -> BytesMut {
    let mut b = BytesMut::new();
    b.put_u32(payload.len() as u32);
    b.extend_from_slice(payload);
    b
  }

  #[test]
  fn decode_skips_undecodable_frame() {
    let mut decoder = MsgDecoder::<bool>::new();
    let mut buf = BytesMut::new();
    buf.unsplit(frame(&bincode::serialize(&true).unwrap())); // good
    buf.unsplit(frame(&[0x05])); // bad: not a valid bool
    buf.unsplit(frame(&bincode::serialize(&false).unwrap())); // good

    assert_eq!(decoder.decode(&mut buf).unwrap(), Some(true));
    assert_eq!(decoder.decode(&mut buf).unwrap(), Some(false));
    assert_eq!(decoder.decode(&mut buf).unwrap(), None);
  }
}