hopr_transport/
pipeline.rs

1use bytes::Bytes;
2use hopr_api::{
3    chain::{ChainKeyOperations, ChainReadChannelOperations, ChainReadTicketOperations, ChainValues},
4    tickets::TicketFactory,
5    types::{
6        crypto::prelude::*,
7        internal::{prelude::*, routing::ResolvedTransportRouting},
8    },
9};
10use hopr_crypto_packet::HoprSurb;
11use hopr_protocol_app::prelude::*;
12use hopr_protocol_hopr::prelude::*;
13use hopr_utils::runtime::AbortableList;
14
15use crate::{
16    HoprTransportProcess, PeerProtocolCounterRegistry,
17    config::HoprPacketPipelineConfig,
18    protocol::{
19        PacketPipelineBuilder, Unset, surb_telemetry,
20        surb_telemetry::{PathSlotResolver, SurbRoundTripRegistry, SurbTelemetryCodec},
21    },
22};
23
24/// Ceiling on SURBs awaiting a reply before the oldest are forgotten.
25///
26/// Only bounds memory: a SURB evicted early simply stops being creditable, which understates
27/// delivery slightly rather than misattributing it.
28const MAX_PENDING_SURBS: u64 = 100_000;
29
30/// Builder for the HOPR packet pipeline.
31///
32/// Creates the encoder/decoder, the unacknowledged-ticket processor, optionally hooks up the
33/// packet capture (when the `capture` feature is enabled) and finally delegates to the lower-level
34/// [`PacketPipelineBuilder`] to spawn the per-stage tasks. The shape of the spawned pipeline is
35/// selected by which terminal `build_for_*` method is called:
36///
37/// - [`HoprPacketPipelineBuilder::build_for_relay`] — full pipeline. Requires
38///   [`HoprPacketPipelineBuilder::with_ticket_events`] to be called beforehand.
39/// - [`HoprPacketPipelineBuilder::build_for_entry`] — Entry nodes. Ticket events are not needed (and any value
40///   previously set is ignored).
41/// - [`HoprPacketPipelineBuilder::build_for_exit`] — Exit nodes. Ticket events are not needed (and any value previously
42///   set is ignored).
43///
44/// The builder is constructed via [`HoprPacketPipelineBuilder::new`] which takes no arguments.
45/// The required components must then be supplied via the corresponding builder methods:
46/// [`identity`](HoprPacketPipelineBuilder::identity), [`transport`](HoprPacketPipelineBuilder::transport),
47/// [`api`](HoprPacketPipelineBuilder::api), [`surb_store`](HoprPacketPipelineBuilder::surb_store),
48/// [`chain_api`](HoprPacketPipelineBuilder::chain_api),
49/// [`ticket_factory`](HoprPacketPipelineBuilder::ticket_factory) and
50/// [`channels_dst`](HoprPacketPipelineBuilder::channels_dst).
51///
52/// The per-peer counter registry defaults to an empty one; override it via
53/// [`HoprPacketPipelineBuilder::with_counters`].
54///
55/// The configuration ([`HoprPacketPipelineConfig`]) is optional and defaults to
56/// `HoprPacketPipelineConfig::default()`; override it via [`HoprPacketPipelineBuilder::with_config`].
57pub struct HoprPacketPipelineBuilder<
58    WIn,
59    WOut,
60    Chain,
61    S,
62    TFact,
63    AppOut,
64    AppIn,
65    TEvt = futures::sink::Drain<hopr_api::node::TicketEvent>,
66> {
67    packet_key: Option<OffchainKeypair>,
68    chain_key: Option<ChainKeypair>,
69    wire_msg: (WOut, WIn),
70    api: (AppOut, AppIn),
71    surb_store: S,
72    chain_api: Chain,
73    ticket_factory: TFact,
74    counters: PeerProtocolCounterRegistry,
75    surb_telemetry: (SurbRoundTripRegistry, PathSlotResolver),
76    channels_dst: Option<Hash>,
77    cfg: HoprPacketPipelineConfig,
78    ticket_events: Option<TEvt>,
79}
80
81impl Default
82    for HoprPacketPipelineBuilder<
83        Unset,
84        Unset,
85        Unset,
86        Unset,
87        Unset,
88        Unset,
89        Unset,
90        futures::sink::Drain<hopr_api::node::TicketEvent>,
91    >
92{
93    fn default() -> Self {
94        Self::new()
95    }
96}
97
98impl
99    HoprPacketPipelineBuilder<
100        Unset,
101        Unset,
102        Unset,
103        Unset,
104        Unset,
105        Unset,
106        Unset,
107        futures::sink::Drain<hopr_api::node::TicketEvent>,
108    >
109{
110    /// Creates a new empty builder. All required components must then be supplied via the
111    /// corresponding builder methods before calling any of the terminal `build_for_*` methods.
112    pub fn new() -> Self {
113        Self {
114            packet_key: None,
115            chain_key: None,
116            wire_msg: (Unset, Unset),
117            api: (Unset, Unset),
118            surb_store: Unset,
119            chain_api: Unset,
120            ticket_factory: Unset,
121            counters: PeerProtocolCounterRegistry::default(),
122            surb_telemetry: (SurbRoundTripRegistry::default(), surb_telemetry::no_path_slots()),
123            channels_dst: None,
124            cfg: HoprPacketPipelineConfig::default(),
125            ticket_events: None,
126        }
127    }
128}
129
130impl<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
131    HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
132{
133    /// Overrides the default [`HoprPacketPipelineConfig`].
134    #[must_use]
135    pub fn with_config(mut self, cfg: HoprPacketPipelineConfig) -> Self {
136        self.cfg = cfg;
137        self
138    }
139
140    /// Overrides the default (empty) per-peer protocol counter registry.
141    #[must_use]
142    pub fn with_counters(mut self, counters: PeerProtocolCounterRegistry) -> Self {
143        self.counters = counters;
144        self
145    }
146
147    /// Accumulates SURB round-trips into `registry`, naming paths via `slots`.
148    ///
149    /// Without this the codec is still wrapped, but nothing resolves and nothing is recorded.
150    #[must_use]
151    pub fn with_surb_telemetry(mut self, registry: SurbRoundTripRegistry, slots: PathSlotResolver) -> Self {
152        self.surb_telemetry = (registry, slots);
153        self
154    }
155
156    /// Sets the node identity (chain and offchain keypairs).
157    #[must_use]
158    pub fn identity<'a, I>(mut self, identity: I) -> Self
159    where
160        I: Into<(&'a ChainKeypair, &'a OffchainKeypair)>,
161    {
162        let (chain_key, packet_key) = identity.into();
163        self.chain_key = Some(chain_key.clone());
164        self.packet_key = Some(packet_key.clone());
165        self
166    }
167
168    /// Sets the channel-set domain separator used by the codec and ticket processor.
169    #[must_use]
170    pub fn channels_dst(mut self, channels_dst: Hash) -> Self {
171        self.channels_dst = Some(channels_dst);
172        self
173    }
174
175    /// Sets the underlying wire-message transport (outgoing sink, incoming stream).
176    #[must_use]
177    pub fn transport<WIn2, WOut2>(
178        self,
179        wire_msg: (WOut2, WIn2),
180    ) -> HoprPacketPipelineBuilder<WIn2, WOut2, Chain, S, TFact, AppOut, AppIn, TEvt> {
181        HoprPacketPipelineBuilder {
182            packet_key: self.packet_key,
183            chain_key: self.chain_key,
184            wire_msg,
185            api: self.api,
186            surb_store: self.surb_store,
187            chain_api: self.chain_api,
188            ticket_factory: self.ticket_factory,
189            counters: self.counters,
190            surb_telemetry: self.surb_telemetry,
191            channels_dst: self.channels_dst,
192            cfg: self.cfg,
193            ticket_events: self.ticket_events,
194        }
195    }
196
197    /// Sets the application API (incoming sink, outgoing stream).
198    #[must_use]
199    pub fn api<AppOut2, AppIn2>(
200        self,
201        api: (AppOut2, AppIn2),
202    ) -> HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut2, AppIn2, TEvt> {
203        HoprPacketPipelineBuilder {
204            packet_key: self.packet_key,
205            chain_key: self.chain_key,
206            wire_msg: self.wire_msg,
207            api,
208            surb_store: self.surb_store,
209            chain_api: self.chain_api,
210            ticket_factory: self.ticket_factory,
211            counters: self.counters,
212            surb_telemetry: self.surb_telemetry,
213            channels_dst: self.channels_dst,
214            cfg: self.cfg,
215            ticket_events: self.ticket_events,
216        }
217    }
218
219    /// Sets the SURB store used by the encoder/decoder.
220    #[must_use]
221    pub fn surb_store<S2>(
222        self,
223        surb_store: S2,
224    ) -> HoprPacketPipelineBuilder<WIn, WOut, Chain, S2, TFact, AppOut, AppIn, TEvt> {
225        HoprPacketPipelineBuilder {
226            packet_key: self.packet_key,
227            chain_key: self.chain_key,
228            wire_msg: self.wire_msg,
229            api: self.api,
230            surb_store,
231            chain_api: self.chain_api,
232            ticket_factory: self.ticket_factory,
233            counters: self.counters,
234            surb_telemetry: self.surb_telemetry,
235            channels_dst: self.channels_dst,
236            cfg: self.cfg,
237            ticket_events: self.ticket_events,
238        }
239    }
240
241    /// Sets the chain API used by the encoder/decoder and the unacknowledged ticket processor.
242    #[must_use]
243    pub fn chain_api<Chain2>(
244        self,
245        chain_api: Chain2,
246    ) -> HoprPacketPipelineBuilder<WIn, WOut, Chain2, S, TFact, AppOut, AppIn, TEvt> {
247        HoprPacketPipelineBuilder {
248            packet_key: self.packet_key,
249            chain_key: self.chain_key,
250            wire_msg: self.wire_msg,
251            api: self.api,
252            surb_store: self.surb_store,
253            chain_api,
254            ticket_factory: self.ticket_factory,
255            counters: self.counters,
256            surb_telemetry: self.surb_telemetry,
257            channels_dst: self.channels_dst,
258            cfg: self.cfg,
259            ticket_events: self.ticket_events,
260        }
261    }
262
263    /// Sets the ticket factory used by the encoder/decoder.
264    #[must_use]
265    pub fn ticket_factory<TFact2>(
266        self,
267        ticket_factory: TFact2,
268    ) -> HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact2, AppOut, AppIn, TEvt> {
269        HoprPacketPipelineBuilder {
270            packet_key: self.packet_key,
271            chain_key: self.chain_key,
272            wire_msg: self.wire_msg,
273            api: self.api,
274            surb_store: self.surb_store,
275            chain_api: self.chain_api,
276            ticket_factory,
277            counters: self.counters,
278            surb_telemetry: self.surb_telemetry,
279            channels_dst: self.channels_dst,
280            cfg: self.cfg,
281            ticket_events: self.ticket_events,
282        }
283    }
284
285    /// Attaches the ticket events sink. Required for Relay nodes (see
286    /// [`HoprPacketPipelineBuilder::build_for_relay`]); ignored by Entry and Exit nodes.
287    #[must_use]
288    pub fn with_ticket_events<TEvt2>(
289        self,
290        ticket_events: TEvt2,
291    ) -> HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt2> {
292        HoprPacketPipelineBuilder {
293            packet_key: self.packet_key,
294            chain_key: self.chain_key,
295            wire_msg: self.wire_msg,
296            api: self.api,
297            surb_store: self.surb_store,
298            chain_api: self.chain_api,
299            ticket_factory: self.ticket_factory,
300            counters: self.counters,
301            surb_telemetry: self.surb_telemetry,
302            channels_dst: self.channels_dst,
303            cfg: self.cfg,
304            ticket_events: Some(ticket_events),
305        }
306    }
307}
308
309// Implementation detail: codec, decoder and optional capture wiring shared by the three terminals.
310impl<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
311    HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
312where
313    WOut: futures::Sink<(PeerId, Bytes)> + Clone + Unpin + Send + 'static,
314    WOut::Error: std::error::Error,
315    WIn: futures::Stream<Item = (PeerId, Bytes)> + Send + 'static,
316    Chain: ChainKeyOperations
317        + ChainReadChannelOperations
318        + ChainReadTicketOperations
319        + ChainValues
320        + Clone
321        + Send
322        + Sync
323        + 'static,
324    S: SurbStore + Clone + Send + Sync + 'static,
325    TFact: TicketFactory + Clone + Send + Sync + 'static,
326    AppOut: futures::Sink<(HoprPseudonym, ApplicationDataIn)> + Send + 'static,
327    AppOut::Error: std::error::Error,
328    AppIn: futures::Stream<Item = (ResolvedTransportRouting<HoprSurb>, ApplicationDataOut)> + Send + 'static,
329{
330    /// Builds the codec pair (and capture wiring when enabled) and the unacknowledged ticket
331    /// processor, returning them together with an [`AbortableList`] already containing the
332    /// capture task if any was started.
333    #[allow(clippy::type_complexity)]
334    fn prepare(
335        self,
336    ) -> (
337        OffchainKeypair,
338        (WOut, WIn),
339        (AppOut, AppIn),
340        PeerProtocolCounterRegistry,
341        HoprUnacknowledgedTicketProcessor<Chain>,
342        Option<TEvt>,
343        HoprPacketPipelineConfig,
344        // Codec parts in their final shape (possibly wrapped by capture)
345        AbortableList<HoprTransportProcess>,
346        BuiltCodec<Chain, S, TFact>,
347    ) {
348        let HoprPacketPipelineBuilder {
349            packet_key,
350            chain_key,
351            wire_msg,
352            api,
353            surb_store,
354            chain_api,
355            ticket_factory,
356            counters,
357            surb_telemetry,
358            channels_dst,
359            cfg,
360            ticket_events,
361        } = self;
362
363        let packet_key = packet_key.expect("identity() must be called before building the pipeline");
364        let chain_key = chain_key.expect("identity() must be called before building the pipeline");
365        let channels_dst = channels_dst.expect("channels_dst() must be called before building the pipeline");
366
367        let unack_ticket_proc = HoprUnacknowledgedTicketProcessor::new(
368            chain_api.clone(),
369            chain_key.clone(),
370            channels_dst,
371            cfg.ack_processor,
372        );
373
374        let encoder = HoprEncoder::new(
375            chain_key.clone(),
376            chain_api.clone(),
377            surb_store.clone(),
378            ticket_factory.clone(),
379            channels_dst,
380            cfg.codec,
381        );
382
383        let decoder = HoprDecoder::new(
384            (packet_key.clone(), chain_key.clone()),
385            chain_api.clone(),
386            surb_store,
387            ticket_factory.clone(),
388            channels_dst,
389            cfg.codec,
390        );
391
392        // Wrapped unconditionally: with no resolver configured nothing places a node, so no leg is
393        // ever identified and the decorator costs one closure call per minted SURB.
394        let (surb_registry, path_slots) = surb_telemetry;
395        let me = *packet_key.public();
396        // One map for both halves: the encoder mints and the decoder observes the reply.
397        let pending_legs = surb_telemetry::pending_legs(MAX_PENDING_SURBS);
398        let encoder = SurbTelemetryCodec::new(
399            encoder,
400            me,
401            path_slots.clone(),
402            surb_registry.clone(),
403            pending_legs.clone(),
404        );
405        let decoder = SurbTelemetryCodec::new(decoder, me, path_slots, surb_registry, pending_legs);
406
407        #[allow(unused_mut)]
408        let mut processes = AbortableList::default();
409
410        #[cfg(feature = "capture")]
411        let codec = {
412            use crate::capture;
413
414            let writer: Box<dyn capture::PacketWriter + Send + 'static> =
415                if let Ok(desc) = std::env::var("HOPR_CAPTURE_PACKETS") {
416                    if let Ok(pcap_writer) = std::fs::File::create(&desc).and_then(capture::PcapPacketWriter::new) {
417                        tracing::warn!("pcap file packet capture initialized to {desc}");
418                        Box::new(pcap_writer)
419                    } else {
420                        tracing::error!(desc, "failed to create packet capture: invalid socket address or file");
421                        Box::new(capture::NullWriter)
422                    }
423                } else {
424                    tracing::warn!("no packet capture specified");
425                    Box::new(capture::NullWriter)
426                };
427
428            let (sender, ah) = capture::packet_capture_channel(writer);
429            processes.insert(HoprTransportProcess::Capture, ah);
430            BuiltCodec::Captured(
431                capture::CapturePacketCodec::new(encoder, *packet_key.public(), sender.clone()),
432                capture::CapturePacketCodec::new(decoder, *packet_key.public(), sender),
433            )
434        };
435
436        #[cfg(not(feature = "capture"))]
437        let codec = BuiltCodec::Plain(encoder, decoder);
438
439        (
440            packet_key,
441            wire_msg,
442            api,
443            counters,
444            unack_ticket_proc,
445            ticket_events,
446            cfg,
447            processes,
448            codec,
449        )
450    }
451}
452
453/// Internal helper to keep the codec types abstracted between the capture/no-capture builds.
454enum BuiltCodec<Chain, S, TFact>
455where
456    Chain: ChainKeyOperations
457        + ChainReadChannelOperations
458        + ChainReadTicketOperations
459        + ChainValues
460        + Clone
461        + Send
462        + Sync
463        + 'static,
464    S: SurbStore + Clone + Send + Sync + 'static,
465    TFact: TicketFactory + Clone + Send + Sync + 'static,
466{
467    #[cfg(not(feature = "capture"))]
468    Plain(
469        SurbTelemetryCodec<HoprEncoder<Chain, S, TFact>>,
470        SurbTelemetryCodec<HoprDecoder<Chain, S, TFact>>,
471    ),
472    #[cfg(feature = "capture")]
473    Captured(
474        crate::capture::CapturePacketCodec<SurbTelemetryCodec<HoprEncoder<Chain, S, TFact>>>,
475        crate::capture::CapturePacketCodec<SurbTelemetryCodec<HoprDecoder<Chain, S, TFact>>>,
476    ),
477}
478
479// Terminal: Relay
480impl<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
481    HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
482where
483    WOut: futures::Sink<(PeerId, Bytes)> + Clone + Unpin + Send + 'static,
484    WOut::Error: std::error::Error,
485    WIn: futures::Stream<Item = (PeerId, Bytes)> + Send + 'static,
486    Chain: ChainKeyOperations
487        + ChainReadChannelOperations
488        + ChainReadTicketOperations
489        + ChainValues
490        + Clone
491        + Send
492        + Sync
493        + 'static,
494    S: SurbStore + Clone + Send + Sync + 'static,
495    TEvt: futures::Sink<hopr_api::node::TicketEvent> + Clone + Unpin + Send + 'static,
496    TEvt::Error: std::error::Error,
497    TFact: TicketFactory + Clone + Send + Sync + 'static,
498    AppOut: futures::Sink<(HoprPseudonym, ApplicationDataIn)> + Send + 'static,
499    AppOut::Error: std::error::Error,
500    AppIn: futures::Stream<Item = (ResolvedTransportRouting<HoprSurb>, ApplicationDataOut)> + Send + 'static,
501{
502    /// Builds the pipeline configured for a Relay node.
503    ///
504    /// # Panics
505    /// Panics if [`HoprPacketPipelineBuilder::with_ticket_events`] was not called.
506    pub fn build_for_relay(self) -> AbortableList<HoprTransportProcess> {
507        let (packet_key, wire_msg, api, counters, unack_ticket_proc, ticket_events, _cfg, mut processes, codec) =
508            self.prepare();
509
510        let ticket_events = ticket_events.expect("Relay node requires ticket events; call with_ticket_events() first");
511
512        let inner = match codec {
513            #[cfg(not(feature = "capture"))]
514            BuiltCodec::Plain(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
515                .transport(wire_msg)
516                .codec((encoder, decoder))
517                .api(api)
518                .with_counters(counters)
519                .with_config(_cfg.pipeline)
520                .with_ticket_processing(unack_ticket_proc, ticket_events)
521                .build_for_relay(),
522            #[cfg(feature = "capture")]
523            BuiltCodec::Captured(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
524                .transport(wire_msg)
525                .codec((encoder, decoder))
526                .api(api)
527                .with_counters(counters)
528                .with_config(_cfg.pipeline)
529                .with_ticket_processing(unack_ticket_proc, ticket_events)
530                .build_for_relay(),
531        };
532
533        processes.flat_map_extend_from(inner, HoprTransportProcess::Pipeline);
534        processes
535    }
536}
537
538// Terminal: Entry / Exit (no ticket events required)
539impl<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
540    HoprPacketPipelineBuilder<WIn, WOut, Chain, S, TFact, AppOut, AppIn, TEvt>
541where
542    WOut: futures::Sink<(PeerId, Bytes)> + Clone + Unpin + Send + 'static,
543    WOut::Error: std::error::Error,
544    WIn: futures::Stream<Item = (PeerId, Bytes)> + Send + 'static,
545    Chain: ChainKeyOperations
546        + ChainReadChannelOperations
547        + ChainReadTicketOperations
548        + ChainValues
549        + Clone
550        + Send
551        + Sync
552        + 'static,
553    S: SurbStore + Clone + Send + Sync + 'static,
554    TFact: TicketFactory + Clone + Send + Sync + 'static,
555    AppOut: futures::Sink<(HoprPseudonym, ApplicationDataIn)> + Send + 'static,
556    AppOut::Error: std::error::Error,
557    AppIn: futures::Stream<Item = (ResolvedTransportRouting<HoprSurb>, ApplicationDataOut)> + Send + 'static,
558{
559    /// Builds the pipeline configured for an Entry node.
560    ///
561    /// The incoming acknowledgement pipeline is not started; ticket events (if any) are ignored.
562    pub fn build_for_entry(self) -> AbortableList<HoprTransportProcess> {
563        let (packet_key, wire_msg, api, counters, _unack, _ticket_events, _cfg, mut processes, codec) = self.prepare();
564
565        let inner = match codec {
566            #[cfg(not(feature = "capture"))]
567            BuiltCodec::Plain(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
568                .transport(wire_msg)
569                .codec((encoder, decoder))
570                .api(api)
571                .with_counters(counters)
572                .with_config(_cfg.pipeline)
573                .build_for_entry(),
574            #[cfg(feature = "capture")]
575            BuiltCodec::Captured(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
576                .transport(wire_msg)
577                .codec((encoder, decoder))
578                .api(api)
579                .with_counters(counters)
580                .with_config(_cfg.pipeline)
581                .build_for_entry(),
582        };
583
584        processes.flat_map_extend_from(inner, HoprTransportProcess::Pipeline);
585        processes
586    }
587
588    /// Builds the pipeline configured for an Exit node.
589    ///
590    /// The incoming acknowledgement pipeline is started but its acknowledgements are drained
591    /// (never forwarded to a ticket processor); ticket events (if any) are ignored.
592    pub fn build_for_exit(self) -> AbortableList<HoprTransportProcess> {
593        let (packet_key, wire_msg, api, counters, _unack, _ticket_events, _cfg, mut processes, codec) = self.prepare();
594
595        let inner = match codec {
596            #[cfg(not(feature = "capture"))]
597            BuiltCodec::Plain(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
598                .transport(wire_msg)
599                .codec((encoder, decoder))
600                .api(api)
601                .with_counters(counters)
602                .with_config(_cfg.pipeline)
603                .build_for_exit(),
604            #[cfg(feature = "capture")]
605            BuiltCodec::Captured(encoder, decoder) => PacketPipelineBuilder::new(packet_key.clone())
606                .transport(wire_msg)
607                .codec((encoder, decoder))
608                .api(api)
609                .with_counters(counters)
610                .with_config(_cfg.pipeline)
611                .build_for_exit(),
612        };
613
614        processes.flat_map_extend_from(inner, HoprTransportProcess::Pipeline);
615        processes
616    }
617}
hopr_transport/pipeline.rs

hopr_transport/
pipeline.rs
