hopr_transport_session/
types.rs

1use std::{
2    convert::Into,
3    fmt::{Debug, Formatter},
4    pin::Pin,
5    task::{Context, Poll},
6    time::Duration,
7};
8
9use futures::{SinkExt, StreamExt, TryStreamExt};
10use hopr_api::types::{
11    internal::{prelude::HoprPseudonym, routing::DestinationRouting},
12    primitive::errors::GeneralError,
13};
14use hopr_protocol_app::prelude::{ApplicationData, ApplicationDataIn, ApplicationDataOut, ReservedTag, Tag};
15#[cfg(feature = "telemetry")]
16use hopr_protocol_session::NoopTracker;
17use hopr_protocol_session::{
18    AcknowledgementMode, AcknowledgementState, AcknowledgementStateConfig, ReliableSocket, SessionSocketConfig,
19    UnreliableSocket,
20};
21use hopr_protocol_start::StartProtocol;
22use hopr_utils::network_types::{
23    prelude::SealedHost,
24    utils::{AsyncWriteSink, DuplexIO},
25};
26use tracing::{debug, instrument};
27
28use crate::{Capabilities, Capability, errors::TransportSessionError};
29
30/// Wrapper for [`Capabilities`] that makes conversion to/from `u8` possible.
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub struct ByteCapabilities(pub Capabilities);
33
34impl TryFrom<u8> for ByteCapabilities {
35    type Error = GeneralError;
36
37    fn try_from(value: u8) -> Result<Self, Self::Error> {
38        Capabilities::new(value)
39            .map(Self)
40            .map_err(|_| GeneralError::ParseError("capabilities".into()))
41    }
42}
43
44impl From<ByteCapabilities> for u8 {
45    fn from(value: ByteCapabilities) -> Self {
46        *value.0.as_ref()
47    }
48}
49
50impl From<ByteCapabilities> for Capabilities {
51    fn from(value: ByteCapabilities) -> Self {
52        value.0
53    }
54}
55
56impl From<Capabilities> for ByteCapabilities {
57    fn from(value: Capabilities) -> Self {
58        Self(value)
59    }
60}
61
62impl AsRef<Capabilities> for ByteCapabilities {
63    fn as_ref(&self) -> &Capabilities {
64        &self.0
65    }
66}
67
68/// Start protocol instantiation for HOPR.
69pub type HoprStartProtocol = StartProtocol<SessionId, SessionTarget, ByteCapabilities>;
70
71/// Constant application tag used for all sessions.
72/// Previously tags were dynamically allocated per session.
73pub const SESSION_APPLICATION_TAG: Tag = Tag::Reserved(ReservedTag::Session as u64);
74
75/// Unique ID of a specific Session.
76///
77/// Now a simple type alias for HoprPseudonym since we use a constant
78/// application tag for all sessions instead of dynamically allocating tags.
79pub type SessionId = HoprPseudonym;
80
81pub(crate) fn caps_to_ack_mode(caps: Capabilities) -> AcknowledgementMode {
82    if caps.contains(Capability::RetransmissionAck | Capability::RetransmissionNack) {
83        AcknowledgementMode::Both
84    } else if caps.contains(Capability::RetransmissionAck) {
85        AcknowledgementMode::Full
86    } else {
87        AcknowledgementMode::Partial
88    }
89}
90
91/// Indicates the closure reason of a [`HoprSession`].
92#[derive(Clone, Copy, Debug, PartialEq, Eq, strum::Display)]
93pub enum ClosureReason {
94    /// Write-half of the Session has been closed.
95    WriteClosed,
96    /// Read-part of the Session has been closed (encountered empty read).
97    EmptyRead,
98    /// Session has been evicted from the cache due to inactivity or capacity reasons.
99    Eviction,
100}
101
102/// Helper trait to allow Box aliasing
103trait AsyncReadWrite: futures::AsyncWrite + futures::AsyncRead + Send + Unpin {}
104impl<T: futures::AsyncWrite + futures::AsyncRead + Send + Unpin> AsyncReadWrite for T {}
105
106/// Describes a node service target.
107/// These are specialized [`SessionTargets`](SessionTarget::ExitNode)
108/// that are local to the Exit node and have different purposes, such as Cover Traffic.
109///
110/// These targets cannot be [sealed](SealedHost) from the Entry node.
111pub type ServiceId = u32;
112
113/// Defines what should happen with the data at the recipient where the
114/// data from the established session are supposed to be forwarded to some `target`.
115#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
116pub enum SessionTarget {
117    /// Target is running over UDP with the given IP address and port.
118    UdpStream(SealedHost),
119    /// Target is running over TCP with the given address and port.
120    TcpStream(SealedHost),
121    /// Target is a service directly at the exit node with the given service ID.
122    ExitNode(ServiceId),
123}
124
125/// Wrapper for incoming [`HoprSession`] along with other information
126/// extracted from the Start protocol during the session establishment.
127#[derive(Debug)]
128pub struct IncomingSession {
129    /// Actual incoming session.
130    pub session: HoprSession,
131    /// Desired [target](SessionTarget) of the data received over the session.
132    pub target: SessionTarget,
133}
134
135/// Configures the Session protocol socket over HOPR.
136#[derive(Copy, Clone, Debug, PartialEq, Eq, smart_default::SmartDefault, serde::Serialize)]
137pub struct HoprSessionConfig {
138    /// Capabilities of the Session protocol socket.
139    ///
140    /// Default is no capabilities.
141    #[default(Capabilities::empty())]
142    pub capabilities: Capabilities,
143    /// Expected frame size of the Session protocol socket.
144    ///
145    /// Default is 1500.
146    #[default(1500)]
147    pub frame_mtu: usize,
148    /// Maximum amount of time an incomplete frame can be kept in the buffer.
149    ///
150    /// Default is 800 ms
151    #[default(Duration::from_millis(800))]
152    #[serde(with = "humantime_serde")]
153    pub frame_timeout: Duration,
154    /// Maximum number of segments to buffer in the downstream transport.
155    /// If 0 is given, the transport is unbuffered.
156    ///
157    /// Default is 0.
158    #[default(0)]
159    pub max_buffered_segments: usize,
160}
161
162/// Represents the Session protocol socket over HOPR.
163///
164/// This is essentially a HOPR-specific wrapper for [`ReliableSocket`] and [`UnreliableSocket`]
165/// Session protocol sockets.
166#[pin_project::pin_project]
167pub struct HoprSession {
168    id: SessionId,
169    #[pin]
170    inner: Box<dyn AsyncReadWrite>,
171    routing: DestinationRouting,
172    cfg: HoprSessionConfig,
173    on_close: Option<Box<dyn FnOnce(SessionId, ClosureReason) + Send + Sync>>,
174}
175
176pub(crate) const SESSION_SOCKET_CAPACITY: usize = 16384;
177
178impl HoprSession {
179    /// Creates a new HOPR Session.
180    ///
181    /// It builds an [`futures::io::AsyncRead`] + [`futures::io::AsyncWrite`] transport
182    /// from the given `hopr` interface and passing it to the appropriate [`UnreliableSocket`] or [`ReliableSocket`]
183    /// based on the given `capabilities`.
184    ///
185    /// The `on_close` closure can be optionally called when the Session has been closed via `poll_close`.
186    #[tracing::instrument(skip_all, fields(id, routing, cfg, session_id = %id))]
187    pub fn new<Tx, Rx>(
188        id: SessionId,
189        routing: DestinationRouting,
190        cfg: HoprSessionConfig,
191        hopr: (Tx, Rx),
192        on_close: Option<Box<dyn FnOnce(SessionId, ClosureReason) + Send + Sync>>,
193    ) -> Result<Self, TransportSessionError>
194    where
195        Tx: futures::Sink<(DestinationRouting, ApplicationDataOut)> + Send + Unpin + 'static,
196        Rx: futures::Stream<Item = ApplicationDataIn> + Send + Unpin + 'static,
197        Tx::Error: std::error::Error + Send + Sync,
198    {
199        let routing_clone = routing.clone();
200
201        #[cfg(feature = "telemetry")]
202        let (session_id_write, session_id_read) = (id, id);
203
204        // Wrap the HOPR transport so that it appears as regular transport to the SessionSocket
205        let transport = DuplexIO(
206            AsyncWriteSink::<{ ApplicationData::PAYLOAD_SIZE }, _>(hopr.0.sink_map_err(std::io::Error::other).with(
207                move |buf: Box<[u8]>| {
208                    #[cfg(feature = "telemetry")]
209                    crate::telemetry::record_session_write(&session_id_write, buf.len());
210                    // The Session protocol does not set any packet info on outgoing packets.
211                    // However, the SessionManager on top usually overrides this.
212                    futures::future::ready(
213                        ApplicationData::new(SESSION_APPLICATION_TAG, buf.into_vec())
214                            .map(|data| (routing_clone.clone(), ApplicationDataOut::with_no_packet_info(data)))
215                            .map_err(std::io::Error::other),
216                    )
217                },
218            )),
219            // The Session protocol ignores the packet info on incoming packets.
220            // It is typically SessionManager's job to interpret those.
221            hopr.1
222                .map(move |data| {
223                    #[cfg(feature = "telemetry")]
224                    crate::telemetry::record_session_read(&session_id_read, data.data.plain_text.len());
225                    Ok::<_, std::io::Error>(data.data.plain_text)
226                })
227                .into_async_read(),
228        );
229
230        // Based on the requested capabilities, see if we should use the Session protocol
231        let inner: Box<dyn AsyncReadWrite> = if cfg.capabilities.contains(Capability::Segmentation) {
232            let socket_cfg = SessionSocketConfig {
233                frame_size: cfg.frame_mtu,
234                frame_timeout: cfg.frame_timeout,
235                capacity: SESSION_SOCKET_CAPACITY,
236                flush_immediately: cfg.capabilities.contains(Capability::NoDelay),
237                max_buffered_segments: cfg.max_buffered_segments,
238                ..Default::default()
239            };
240
241            // Need to test the capabilities separately, because any Retransmission capability
242            // implies Segmentation, and therefore `is_disjoint` would fail
243            if cfg.capabilities.contains(Capability::RetransmissionAck)
244                || cfg.capabilities.contains(Capability::RetransmissionNack)
245            {
246                // TODO: update config values
247                let ack_cfg = AcknowledgementStateConfig {
248                    // This is a very coarse assumption, that a single 3-hop packet
249                    // takes on average 200 ms to deliver.
250                    // We can no longer base this timeout on the number of hops because
251                    // it is not known for SURB-based routing.
252                    expected_packet_latency: Duration::from_millis(200),
253                    mode: caps_to_ack_mode(cfg.capabilities),
254                    backoff_base: 0.2,
255                    max_incoming_frame_retries: 1,
256                    max_outgoing_frame_retries: 2,
257                    ..Default::default()
258                };
259
260                debug!(?socket_cfg, ?ack_cfg, "opening new stateful session socket");
261
262                Box::new(ReliableSocket::new(
263                    transport,
264                    AcknowledgementState::<{ ApplicationData::PAYLOAD_SIZE }>::new(id, ack_cfg),
265                    socket_cfg,
266                    #[cfg(feature = "telemetry")]
267                    NoopTracker,
268                )?)
269            } else {
270                debug!(?socket_cfg, "opening new stateless session socket");
271
272                Box::new(UnreliableSocket::<{ ApplicationData::PAYLOAD_SIZE }>::new_stateless(
273                    id,
274                    transport,
275                    socket_cfg,
276                    #[cfg(feature = "telemetry")]
277                    NoopTracker,
278                )?)
279            }
280        } else {
281            debug!("opening raw session socket");
282            Box::new(transport)
283        };
284
285        Ok(Self {
286            id,
287            inner,
288            routing,
289            cfg,
290            on_close,
291        })
292    }
293
294    /// ID of this Session.
295    pub fn id(&self) -> &SessionId {
296        &self.id
297    }
298
299    /// Routing options used to deliver data.
300    pub fn routing(&self) -> &DestinationRouting {
301        &self.routing
302    }
303
304    /// Configuration of this Session.
305    pub fn config(&self) -> &HoprSessionConfig {
306        &self.cfg
307    }
308}
309
310impl std::fmt::Debug for HoprSession {
311    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
312        f.debug_struct("Session")
313            .field("id", &self.id)
314            .field("routing", &self.routing)
315            .finish_non_exhaustive()
316    }
317}
318
319impl futures::AsyncRead for HoprSession {
320    #[instrument(name = "Session::poll_read", level = "trace", skip_all, fields(session_id = %self.id), ret)]
321    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
322        let this = self.project();
323        let read = futures::ready!(this.inner.poll_read(cx, buf))?;
324        if read == 0 {
325            tracing::trace!("hopr session empty read");
326            // Empty read signals end of the socket, notify if needed
327            if let Some(notifier) = this.on_close.take() {
328                tracing::trace!("notifying read half closure of session");
329                notifier(*this.id, ClosureReason::EmptyRead);
330            }
331        }
332        Poll::Ready(Ok(read))
333    }
334}
335
336impl futures::AsyncWrite for HoprSession {
337    #[instrument(name = "Session::poll_write", level = "trace", skip_all, fields(session_id = %self.id), ret)]
338    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
339        self.project().inner.poll_write(cx, buf)
340    }
341
342    #[instrument(name = "Session::poll_flush", level = "trace", skip_all, fields(session_id = %self.id), ret)]
343    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
344        self.project().inner.poll_flush(cx)
345    }
346
347    #[instrument(name = "Session::poll_close", level = "trace", skip_all, fields(session_id = %self.id), ret)]
348    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
349        let this = self.project();
350        futures::ready!(this.inner.poll_close(cx))?;
351        tracing::trace!("hopr session closed");
352
353        #[cfg(feature = "telemetry")]
354        crate::telemetry::set_session_state(this.id, crate::telemetry::SessionLifecycleState::Closing);
355
356        if let Some(notifier) = this.on_close.take() {
357            tracing::trace!("notifying write half closure of session");
358            notifier(*this.id, ClosureReason::WriteClosed);
359        }
360
361        Poll::Ready(Ok(()))
362    }
363}
364
365#[cfg(feature = "runtime-tokio")]
366impl tokio::io::AsyncRead for HoprSession {
367    fn poll_read(
368        mut self: Pin<&mut Self>,
369        cx: &mut Context<'_>,
370        buf: &mut tokio::io::ReadBuf<'_>,
371    ) -> Poll<std::io::Result<()>> {
372        let slice = buf.initialize_unfilled();
373        let n = std::task::ready!(futures::AsyncRead::poll_read(self.as_mut(), cx, slice))?;
374        buf.advance(n);
375        Poll::Ready(Ok(()))
376    }
377}
378
379#[cfg(feature = "runtime-tokio")]
380impl tokio::io::AsyncWrite for HoprSession {
381    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, std::io::Error>> {
382        futures::AsyncWrite::poll_write(self.as_mut(), cx, buf)
383    }
384
385    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
386        futures::AsyncWrite::poll_flush(self.as_mut(), cx)
387    }
388
389    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
390        futures::AsyncWrite::poll_close(self.as_mut(), cx)
391    }
392}
393
394#[cfg(test)]
395mod tests {
396    use anyhow::Context;
397    use futures::{AsyncReadExt, AsyncWriteExt};
398    use hopr_api::types::{
399        crypto::prelude::*, crypto_random::Randomizable, internal::routing::RoutingOptions, primitive::prelude::*,
400    };
401
402    use super::*;
403
404    // --- ByteCapabilities tests ---
405
406    #[test]
407    fn byte_capabilities_roundtrip_via_u8() -> anyhow::Result<()> {
408        let flags: Capabilities = Capability::Segmentation.into();
409        let caps = ByteCapabilities::from(flags);
410        let byte_val: u8 = caps.into();
411        let restored = ByteCapabilities::try_from(byte_val)?;
412        assert_eq!(caps, restored);
413        Ok(())
414    }
415
416    #[test]
417    fn byte_capabilities_invalid_bits_are_rejected() {
418        // 0xFF has bits set that don't correspond to any Capability
419        assert!(ByteCapabilities::try_from(0xFF_u8).is_err());
420    }
421
422    #[test]
423    fn byte_capabilities_empty_is_zero() {
424        let caps = ByteCapabilities::from(Capabilities::empty());
425        let byte_val: u8 = caps.into();
426        assert_eq!(byte_val, 0);
427    }
428
429    #[test]
430    fn byte_capabilities_combined_flags() -> anyhow::Result<()> {
431        let caps: Capabilities = Capability::Segmentation | Capability::NoRateControl;
432        let byte_caps = ByteCapabilities::from(caps);
433        let byte_val: u8 = byte_caps.into();
434        let restored = ByteCapabilities::try_from(byte_val)?;
435        assert_eq!(*restored.as_ref(), caps);
436        Ok(())
437    }
438
439    // --- caps_to_ack_mode tests ---
440
441    #[test]
442    fn caps_to_ack_mode_both_when_ack_and_nack() {
443        let caps: Capabilities = Capability::RetransmissionAck | Capability::RetransmissionNack;
444        assert_eq!(caps_to_ack_mode(caps), AcknowledgementMode::Both);
445    }
446
447    #[test]
448    fn caps_to_ack_mode_full_when_only_ack() {
449        let caps: Capabilities = Capability::RetransmissionAck.into();
450        assert_eq!(caps_to_ack_mode(caps), AcknowledgementMode::Full);
451    }
452
453    #[test]
454    fn caps_to_ack_mode_partial_when_no_retransmission() {
455        let caps: Capabilities = Capability::Segmentation.into();
456        assert_eq!(caps_to_ack_mode(caps), AcknowledgementMode::Partial);
457    }
458
459    #[test]
460    fn caps_to_ack_mode_partial_when_empty() {
461        assert_eq!(caps_to_ack_mode(Capabilities::empty()), AcknowledgementMode::Partial);
462    }
463
464    #[test]
465    fn caps_to_ack_mode_should_be_partial_when_only_nack() {
466        let caps: Capabilities = Capability::RetransmissionNack.into();
467        assert_eq!(caps_to_ack_mode(caps), AcknowledgementMode::Partial);
468    }
469
470    // --- ClosureReason tests ---
471
472    #[test]
473    fn closure_reason_display_values_are_stable() {
474        let reasons = [
475            ClosureReason::WriteClosed,
476            ClosureReason::EmptyRead,
477            ClosureReason::Eviction,
478        ];
479        insta::assert_debug_snapshot!(reasons);
480    }
481
482    // --- HoprSessionConfig tests ---
483
484    #[test]
485    fn hopr_session_config_default_snapshot() {
486        let cfg = HoprSessionConfig::default();
487        insta::assert_yaml_snapshot!(cfg);
488    }
489
490    // --- SessionTarget tests ---
491
492    #[test]
493    fn session_target_variants_debug_snapshot() -> anyhow::Result<()> {
494        let targets: Vec<SessionTarget> = vec![
495            SessionTarget::UdpStream(SealedHost::Plain(
496                "127.0.0.1:8080".parse().context("parsing UDP target")?,
497            )),
498            SessionTarget::TcpStream(SealedHost::Plain("10.0.0.1:443".parse().context("parsing TCP target")?)),
499            SessionTarget::ExitNode(42),
500        ];
501        insta::assert_debug_snapshot!(targets);
502        Ok(())
503    }
504
505    // --- SessionId edge cases ---
506
507    #[test]
508    fn session_id_display_and_debug_should_be_identical() {
509        let id = HoprPseudonym::random();
510        assert_eq!(format!("{id}"), format!("{id:?}"));
511    }
512
513    #[test]
514    fn session_id_hash_eq_consistency() {
515        use std::collections::HashSet;
516        let pseudonym = HoprPseudonym::random();
517        let id1: SessionId = pseudonym;
518        let id2: SessionId = pseudonym;
519        let id3: SessionId = HoprPseudonym::random();
520
521        let mut set = HashSet::new();
522        set.insert(id1);
523        assert!(set.contains(&id2));
524        assert!(!set.contains(&id3), "different pseudonym should not be in the set");
525    }
526
527    // --- Existing tests ---
528
529    #[test_log::test(tokio::test)]
530    async fn test_session_bidirectional_flow_without_segmentation() -> anyhow::Result<()> {
531        let dst: Address = (&ChainKeypair::random()).into();
532        let id: SessionId = HoprPseudonym::random();
533        const DATA_LEN: usize = 5000;
534
535        let (alice_tx, bob_rx) = futures::channel::mpsc::unbounded::<(DestinationRouting, ApplicationDataOut)>();
536        let (bob_tx, alice_rx) = futures::channel::mpsc::unbounded::<(DestinationRouting, ApplicationDataOut)>();
537
538        let mut alice_session = HoprSession::new(
539            id,
540            DestinationRouting::forward_only(dst, RoutingOptions::Hops(0.try_into()?)),
541            Default::default(),
542            (
543                alice_tx,
544                alice_rx
545                    .map(|(_, data)| ApplicationDataIn {
546                        data: data.data,
547                        packet_info: Default::default(),
548                    })
549                    .inspect(|d| debug!("alice rcvd: {}", d.data.total_len())),
550            ),
551            None,
552        )?;
553
554        let mut bob_session = HoprSession::new(
555            id,
556            DestinationRouting::Return(id.into()),
557            Default::default(),
558            (
559                bob_tx,
560                bob_rx
561                    .map(|(_, data)| ApplicationDataIn {
562                        data: data.data,
563                        packet_info: Default::default(),
564                    })
565                    .inspect(|d| debug!("bob rcvd: {}", d.data.total_len())),
566            ),
567            None,
568        )?;
569
570        let alice_sent = hopr_api::types::crypto_random::random_bytes::<DATA_LEN>();
571        let bob_sent = hopr_api::types::crypto_random::random_bytes::<DATA_LEN>();
572
573        let mut bob_recv = [0u8; DATA_LEN];
574        let mut alice_recv = [0u8; DATA_LEN];
575
576        tokio::time::timeout(Duration::from_secs(1), alice_session.write_all(&alice_sent))
577            .await
578            .context("alice write failed")?
579            .context("alice write timed out")?;
580        alice_session.flush().await?;
581
582        tokio::time::timeout(Duration::from_secs(1), bob_session.write_all(&bob_sent))
583            .await
584            .context("bob write failed")?
585            .context("bob write timed out")?;
586        bob_session.flush().await?;
587
588        tokio::time::timeout(Duration::from_secs(1), bob_session.read_exact(&mut bob_recv))
589            .await
590            .context("bob read failed")?
591            .context("bob read timed out")?;
592
593        tokio::time::timeout(Duration::from_secs(1), alice_session.read_exact(&mut alice_recv))
594            .await
595            .context("alice read failed")?
596            .context("alice read timed out")?;
597
598        assert_eq!(&alice_sent, bob_recv.as_slice());
599        assert_eq!(bob_sent, alice_recv);
600
601        Ok(())
602    }
603
604    #[test_log::test(tokio::test)]
605    async fn test_session_bidirectional_flow_with_segmentation() -> anyhow::Result<()> {
606        let dst: Address = (&ChainKeypair::random()).into();
607        let id: SessionId = HoprPseudonym::random();
608        const DATA_LEN: usize = 5000;
609
610        let (alice_tx, bob_rx) = futures::channel::mpsc::unbounded::<(DestinationRouting, ApplicationDataOut)>();
611        let (bob_tx, alice_rx) = futures::channel::mpsc::unbounded::<(DestinationRouting, ApplicationDataOut)>();
612
613        let mut alice_session = HoprSession::new(
614            id,
615            DestinationRouting::forward_only(dst, RoutingOptions::Hops(0.try_into()?)),
616            HoprSessionConfig {
617                capabilities: Capability::Segmentation.into(),
618                ..Default::default()
619            },
620            (
621                alice_tx,
622                alice_rx
623                    .map(|(_, data)| ApplicationDataIn {
624                        data: data.data,
625                        packet_info: Default::default(),
626                    })
627                    .inspect(|d| debug!("alice rcvd: {}", d.data.total_len())),
628            ),
629            None,
630        )?;
631
632        let mut bob_session = HoprSession::new(
633            id,
634            DestinationRouting::Return(id.into()),
635            HoprSessionConfig {
636                capabilities: Capability::Segmentation.into(),
637                ..Default::default()
638            },
639            (
640                bob_tx,
641                bob_rx
642                    .map(|(_, data)| ApplicationDataIn {
643                        data: data.data,
644                        packet_info: Default::default(),
645                    })
646                    .inspect(|d| debug!("bob rcvd: {}", d.data.total_len())),
647            ),
648            None,
649        )?;
650
651        let alice_sent = hopr_api::types::crypto_random::random_bytes::<DATA_LEN>();
652        let bob_sent = hopr_api::types::crypto_random::random_bytes::<DATA_LEN>();
653
654        let mut bob_recv = [0u8; DATA_LEN];
655        let mut alice_recv = [0u8; DATA_LEN];
656
657        tokio::time::timeout(Duration::from_secs(1), alice_session.write_all(&alice_sent))
658            .await
659            .context("alice write failed")?
660            .context("alice write timed out")?;
661        alice_session.flush().await?;
662
663        tokio::time::timeout(Duration::from_secs(1), bob_session.write_all(&bob_sent))
664            .await
665            .context("bob write failed")?
666            .context("bob write timed out")?;
667        bob_session.flush().await?;
668
669        tokio::time::timeout(Duration::from_secs(1), bob_session.read_exact(&mut bob_recv))
670            .await
671            .context("bob read failed")?
672            .context("bob read timed out")?;
673
674        tokio::time::timeout(Duration::from_secs(1), alice_session.read_exact(&mut alice_recv))
675            .await
676            .context("alice read failed")?
677            .context("alice read timed out")?;
678
679        assert_eq!(alice_sent, bob_recv);
680        assert_eq!(bob_sent, alice_recv);
681
682        Ok(())
683    }
684}
hopr_transport_session/types.rs

hopr_transport_session/
types.rs
