Sulkta-OSS/pallas
3
0
Fork 0
Code Issues 1 Pull requests 4 Projects Releases Packages Wiki Activity Actions

fix(multiplexer): Use buffers that own the inner channel (#113)

Browse source
This commit is contained in:
Santiago Carmuega 2022-06-10 11:54:14 -03:00 • committed by GitHub
parent c6c5f7b1c0
commit 8b54dbeeac
7 changed files with 45 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

18
pallas-multiplexer/src/agents.rs
View file

@ -46,13 +46,13 @@ where
}
/// A channel abstraction to hide the complexity of partial payloads
pub struct ChannelBuffer<'c, C: Channel> {
channel: &'c mut C,
pub struct ChannelBuffer<C: Channel> {
channel: C,
temp: Vec<u8>,
}
impl<'c, C: Channel> ChannelBuffer<'c, C> {
pub fn new(channel: &'c mut C) -> Self {
impl<C: Channel> ChannelBuffer<C> {
pub fn new(channel: C) -> Self {
Self {
channel,
temp: Vec::new(),
@ -105,4 +105,14 @@ impl<'c, C: Channel> ChannelBuffer<'c, C> {
}
}
}
pub fn unwrap(self) -> C {
self.channel
}
}
impl<C: Channel> From<C> for ChannelBuffer<C> {
fn from(channel: C) -> Self {
ChannelBuffer::new(channel)
}
}

7
pallas-multiplexer/src/std.rs
View file

@ -1,4 +1,7 @@
use crate::{agents, demux, mux, Payload};
use crate::{
agents::{self, ChannelBuffer},
demux, mux, Payload,
};
use std::{
sync::{
@ -103,6 +106,8 @@ impl demux::Demuxer<StdEgress> {
pub type StdChannel = (Sender<Payload>, Receiver<Payload>);
pub type StdChannelBuffer = ChannelBuffer<StdChannel>;
impl agents::Channel for StdChannel {
fn enqueue_chunk(&mut self, payload: Payload) -> Result<(), agents::ChannelError> {
match self.0.send(payload) {

8
pallas-multiplexer/tests/integration.rs
View file

@ -72,10 +72,10 @@ fn multiple_messages_in_same_payload() {
minicbor::encode(in_part1, &mut input).unwrap();
minicbor::encode(in_part2, &mut input).unwrap();
let mut channel = std::sync::mpsc::channel();
let channel = std::sync::mpsc::channel();
channel.0.send(input).unwrap();
let mut buf = ChannelBuffer::new(&mut channel);
let mut buf = ChannelBuffer::new(channel);
let out_part1 = buf.recv_full_msg::<(u8, u8, u8)>().unwrap();
let out_part2 = buf.recv_full_msg::<(u8, u8, u8)>().unwrap();
@ -90,14 +90,14 @@ fn fragmented_message_in_multiple_payloads() {
let msg = (11u8, 12u8, 13u8, 14u8, 15u8, 16u8, 17u8);
minicbor::encode(msg, &mut input).unwrap();
let mut channel = std::sync::mpsc::channel();
let channel = std::sync::mpsc::channel();
while !input.is_empty() {
let chunk = Vec::from(input.drain(0..2).as_slice());
channel.0.send(chunk).unwrap();
}
let mut buf = ChannelBuffer::new(&mut channel);
let mut buf = ChannelBuffer::new(channel);
let out_msg = buf.recv_full_msg::<(u8, u8, u8, u8, u8, u8, u8)>().unwrap();