|
1 | | -use crate::client::mix_traffic::MixMessage; |
2 | | -use crate::client::topology_control::TopologyInnerRef; |
3 | | -use futures::channel::mpsc; |
4 | | -use log::{error, info, trace, warn}; |
| 1 | +use crate::client::mix_traffic::{MixMessage, MixMessageSender}; |
| 2 | +use crate::client::topology_control::TopologyAccessor; |
| 3 | +use futures::task::{Context, Poll}; |
| 4 | +use futures::{Future, Stream, StreamExt}; |
| 5 | +use log::*; |
5 | 6 | use sphinx::route::Destination; |
6 | | -use std::time; |
| 7 | +use std::pin::Pin; |
| 8 | +use std::time::Duration; |
| 9 | +use tokio::runtime::Handle; |
| 10 | +use tokio::task::JoinHandle; |
| 11 | +use tokio::time; |
7 | 12 | use topology::NymTopology; |
8 | 13 |
|
9 | | -pub(crate) async fn start_loop_cover_traffic_stream<T: NymTopology>( |
10 | | - tx: mpsc::UnboundedSender<MixMessage>, |
| 14 | +pub(crate) struct LoopCoverTrafficStream<T: NymTopology> { |
| 15 | + average_packet_delay: Duration, |
| 16 | + average_cover_message_sending_delay: Duration, |
| 17 | + next_delay: time::Delay, |
| 18 | + mix_tx: MixMessageSender, |
11 | 19 | our_info: Destination, |
12 | | - topology_ctrl_ref: TopologyInnerRef<T>, |
13 | | - average_cover_message_delay_duration: time::Duration, |
14 | | - average_packet_delay_duration: time::Duration, |
15 | | -) { |
16 | | - info!("Starting loop cover traffic stream"); |
17 | | - loop { |
18 | | - trace!("next cover message!"); |
19 | | - let delay_duration = mix_client::poisson::sample(average_cover_message_delay_duration); |
20 | | - tokio::time::delay_for(delay_duration).await; |
| 20 | + topology_access: TopologyAccessor<T>, |
| 21 | +} |
| 22 | + |
| 23 | +impl<T: NymTopology> Stream for LoopCoverTrafficStream<T> { |
| 24 | + // Item is only used to indicate we should create a new message rather than actual cover message |
| 25 | + // reason being to not introduce unnecessary complexity by having to keep state of topology |
| 26 | + // mutex when trying to acquire it. So right now the Stream trait serves as a glorified timer. |
| 27 | + // Perhaps this should be changed in the future. |
| 28 | + type Item = (); |
| 29 | + |
| 30 | + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 31 | + // it is not yet time to return a message |
| 32 | + if Pin::new(&mut self.next_delay).poll(cx).is_pending() { |
| 33 | + return Poll::Pending; |
| 34 | + }; |
| 35 | + |
| 36 | + // we know it's time to send a message, so let's prepare delay for the next one |
| 37 | + // Get the `now` by looking at the current `delay` deadline |
| 38 | + let now = self.next_delay.deadline(); |
| 39 | + let next_poisson_delay = |
| 40 | + mix_client::poisson::sample(self.average_cover_message_sending_delay); |
| 41 | + |
| 42 | + // The next interval value is `next_poisson_delay` after the one that just |
| 43 | + // yielded. |
| 44 | + let next = now + next_poisson_delay; |
| 45 | + self.next_delay.reset(next); |
| 46 | + |
| 47 | + Poll::Ready(Some(())) |
| 48 | + } |
| 49 | +} |
| 50 | + |
| 51 | +impl<T: 'static + NymTopology> LoopCoverTrafficStream<T> { |
| 52 | + pub(crate) fn new( |
| 53 | + mix_tx: MixMessageSender, |
| 54 | + our_info: Destination, |
| 55 | + topology_access: TopologyAccessor<T>, |
| 56 | + average_cover_message_sending_delay: time::Duration, |
| 57 | + average_packet_delay: time::Duration, |
| 58 | + ) -> Self { |
| 59 | + LoopCoverTrafficStream { |
| 60 | + average_packet_delay, |
| 61 | + average_cover_message_sending_delay, |
| 62 | + next_delay: time::delay_for(Default::default()), |
| 63 | + mix_tx, |
| 64 | + our_info, |
| 65 | + topology_access, |
| 66 | + } |
| 67 | + } |
21 | 68 |
|
22 | | - let read_lock = topology_ctrl_ref.read().await; |
23 | | - let topology = match read_lock.topology.as_ref() { |
| 69 | + async fn on_new_message(&mut self) { |
| 70 | + trace!("next cover message!"); |
| 71 | + let route = match self.topology_access.random_route().await { |
24 | 72 | None => { |
25 | 73 | warn!("No valid topology detected - won't send any loop cover message this time"); |
26 | | - continue; |
| 74 | + return; |
27 | 75 | } |
28 | | - Some(topology) => topology, |
| 76 | + Some(route) => route, |
29 | 77 | }; |
30 | 78 |
|
31 | | - let cover_message = match mix_client::packet::loop_cover_message( |
32 | | - our_info.address.clone(), |
33 | | - our_info.identifier, |
34 | | - topology, |
35 | | - average_packet_delay_duration, |
| 79 | + let cover_message = match mix_client::packet::loop_cover_message_route( |
| 80 | + self.our_info.address.clone(), |
| 81 | + self.our_info.identifier, |
| 82 | + route, |
| 83 | + self.average_packet_delay, |
36 | 84 | ) { |
37 | 85 | Ok(message) => message, |
38 | 86 | Err(err) => { |
39 | 87 | error!( |
40 | 88 | "Somehow we managed to create an invalid cover message - {:?}", |
41 | 89 | err |
42 | 90 | ); |
43 | | - continue; |
| 91 | + return; |
44 | 92 | } |
45 | 93 | }; |
46 | 94 |
|
47 | 95 | // if this one fails, there's no retrying because it means that either: |
48 | 96 | // - we run out of memory |
49 | 97 | // - the receiver channel is closed |
50 | 98 | // in either case there's no recovery and we can only panic |
51 | | - tx.unbounded_send(MixMessage::new(cover_message.0, cover_message.1)) |
| 99 | + self.mix_tx |
| 100 | + .unbounded_send(MixMessage::new(cover_message.0, cover_message.1)) |
52 | 101 | .unwrap(); |
53 | 102 | } |
| 103 | + |
| 104 | + async fn run(&mut self) { |
| 105 | + // we should set initial delay only when we actually start the stream |
| 106 | + self.next_delay = time::delay_for(mix_client::poisson::sample( |
| 107 | + self.average_cover_message_sending_delay, |
| 108 | + )); |
| 109 | + |
| 110 | + while let Some(_) = self.next().await { |
| 111 | + self.on_new_message().await; |
| 112 | + } |
| 113 | + } |
| 114 | + |
| 115 | + pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { |
| 116 | + handle.spawn(async move { |
| 117 | + self.run().await; |
| 118 | + }) |
| 119 | + } |
54 | 120 | } |
0 commit comments