GH-800: Drain the Pipeline

multinode_integration_tests/tests/data_routing_test.rs

@@ -316,7 +316,7 @@ fn multiple_stream_zero_hop_test() {
     let mut another_client = zero_hop_node.make_client(8080, STANDARD_CLIENT_TIMEOUT_MILLIS);
 
     one_client.send_chunk(b"GET / HTTP/1.1\r\nHost: www.example.com\r\n\r\n");
-    another_client.send_chunk(b"GET /online/ HTTP/1.1\r\nHost: whatever.neverssl.com\r\n\r\n");
+    another_client.send_chunk(b"GET /online/ HTTP/1.1\r\nAccept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7\r\nAccept-Language: cs-CZ,cs;q=0.9,en;q=0.8,sk;q=0.7\r\nCache-Control: max-age=0\r\nConnection: keep-alive\r\nHost: whatever.neverssl.com\r\nUpgrade-Insecure-Requests: 1\r\nUser-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36\r\n\r\n");
 
     let one_response = one_client.wait_for_chunk();
     let another_response = another_client.wait_for_chunk();

node/src/proxy_client/stream_establisher.rs

@@ -1,5 +1,6 @@
 // Copyright (c) 2019, MASQ (https://masq.ai) and/or its affiliates. All rights reserved.
 
+use crate::proxy_client::stream_handler_pool::StreamSenders;
 use crate::proxy_client::stream_reader::StreamReader;
 use crate::proxy_client::stream_writer::StreamWriter;
 use crate::sub_lib::channel_wrappers::FuturesChannelFactory;
@@ -14,15 +15,15 @@ use crate::sub_lib::stream_connector::StreamConnectorReal;
 use crate::sub_lib::stream_key::StreamKey;
 use crate::sub_lib::tokio_wrappers::ReadHalfWrapper;
 use actix::Recipient;
-use crossbeam_channel::Sender;
+use crossbeam_channel::{unbounded, Receiver, Sender};
 use masq_lib::logger::Logger;
 use std::io;
 use std::net::IpAddr;
 use std::net::SocketAddr;
 
 pub struct StreamEstablisher {
     pub cryptde: &'static dyn CryptDE,
-    pub stream_adder_tx: Sender<(StreamKey, Box<dyn SenderWrapper<SequencedPacket>>)>,
+    pub stream_adder_tx: Sender<(StreamKey, StreamSenders)>,
     pub stream_killer_tx: Sender<(StreamKey, u64)>,
     pub stream_connector: Box<dyn StreamConnector>,
     pub proxy_client_sub: Recipient<InboundServerData>,
@@ -57,11 +58,13 @@ impl StreamEstablisher {
             payload.target_port,
             &self.logger,
         )?;
+        let (shutdown_signal_tx, shutdown_signal_rx) = unbounded();
 
         self.spawn_stream_reader(
             &payload.clone(),
             connection_info.reader,
             connection_info.peer_addr,
+            shutdown_signal_rx,
         );
 
         let (tx_to_write, rx_to_write) = self.channel_factory.make(connection_info.peer_addr);
@@ -73,8 +76,13 @@ impl StreamEstablisher {
         );
         tokio::spawn(stream_writer);
 
+        let stream_senders = StreamSenders {
+            writer_data: tx_to_write.clone(),
+            reader_shutdown_tx: shutdown_signal_tx,
+        };
+
         self.stream_adder_tx
-            .send((payload.stream_key, tx_to_write.clone()))
+            .send((payload.stream_key, stream_senders))
             .expect("StreamHandlerPool died");
         Ok(tx_to_write)
     }
@@ -84,12 +92,14 @@ impl StreamEstablisher {
         payload: &ClientRequestPayload_0v1,
         read_stream: Box<dyn ReadHalfWrapper>,
         peer_addr: SocketAddr,
+        shutdown_signal: Receiver<()>,
     ) {
         let stream_reader = StreamReader::new(
             payload.stream_key,
             self.proxy_client_sub.clone(),
             read_stream,
             self.stream_killer_tx.clone(),
+            shutdown_signal,
             peer_addr,
         );
         debug!(self.logger, "Spawning StreamReader for {}", peer_addr);
@@ -103,7 +113,7 @@ pub trait StreamEstablisherFactory: Send {
 
 pub struct StreamEstablisherFactoryReal {
     pub cryptde: &'static dyn CryptDE,
-    pub stream_adder_tx: Sender<(StreamKey, Box<dyn SenderWrapper<SequencedPacket>>)>,
+    pub stream_adder_tx: Sender<(StreamKey, StreamSenders)>,
     pub stream_killer_tx: Sender<(StreamKey, u64)>,
     pub proxy_client_subs: ProxyClientSubs,
     pub logger: Logger,
@@ -191,6 +201,7 @@ mod tests {
                 },
                 read_stream,
                 SocketAddr::from_str("1.2.3.4:5678").unwrap(),
+                unbounded().1,
             );
 
             proxy_client_awaiter.await_message_count(1);