kitsune_p2p_proxy 0.0.17

Proxy transport module for kitsune-p2p
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

use crate::*;
use futures::sink::SinkExt;
use futures::stream::StreamExt;
use ghost_actor::dependencies::tracing;
use kitsune_p2p_types::codec::Codec;

/// Wrap a TransportChannelRead in code that decodes ProxyWire items.
pub(crate) fn wrap_wire_read(
    mut read: TransportChannelRead,
) -> futures::channel::mpsc::Receiver<ProxyWire> {
    let (mut send, recv) = futures::channel::mpsc::channel(10);

    metric_task(async move {
        let mut buf = Vec::new();
        while let Some(data) = read.next().await {
            buf.extend_from_slice(&data);
            tracing::trace!("proxy read pending {} bytes", buf.len());
            while let Ok((read_size, wire)) = ProxyWire::decode_ref(&buf) {
                tracing::trace!("proxy read {:?}", wire);
                buf.drain(..(read_size as usize));
                send.send(wire).await.map_err(TransportError::other)?;
            }
        }
        TransportResult::Ok(())
    });

    recv
}