Merge pull request #20 from bytesonus/feature/non-mpsc-connection

Refactor to allow independent handling of data events

Author: rakshith-ravi

src/connection/base_connection.rs

@@ -1,13 +1,13 @@
-use crate::{connection::Buffer, utils::Error};
+use crate::{connection::Buffer, juno_module_impl::JunoModuleImpl, utils::Error};
 use async_trait::async_trait;
-use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use std::sync::Arc;
 
 #[async_trait]
 pub trait BaseConnection {
 	async fn setup_connection(&mut self) -> Result<(), Error>;
-	async fn close_connection(&mut self);
-	async fn send(&mut self, buffer: Buffer);
+	async fn close_connection(&mut self) -> Result<(), Error>;
+	async fn send(&mut self, buffer: Buffer) -> Result<(), Error>;
 
-	fn get_data_receiver(&mut self) -> UnboundedReceiver<Buffer>;
-	fn clone_write_sender(&self) -> UnboundedSender<Buffer>;
+	fn set_data_listener(&mut self, listener: Arc<JunoModuleImpl>);
+	fn get_data_listener(&self) -> &Option<Arc<JunoModuleImpl>>;
 }

src/connection/inet_socket_connection.rs

@@ -1,48 +1,109 @@
 use crate::{
 	connection::{BaseConnection, Buffer},
+	juno_module_impl::JunoModuleImpl,
 	utils::Error,
 };
+use std::sync::Arc;
 
 use async_std::{io::BufReader, net::TcpStream, prelude::*, task};
 use async_trait::async_trait;
-
 use futures::{
 	channel::{
 		mpsc::{unbounded, UnboundedReceiver, UnboundedSender},
 		oneshot::{channel, Sender},
 	},
 	future::{self, Either},
+	SinkExt,
 };
-use futures_util::SinkExt;
 
 pub struct InetSocketConnection {
 	connection_setup: bool,
-	read_data_receiver: Option<UnboundedReceiver<Vec<u8>>>,
-	write_data_sender: Option<UnboundedSender<Vec<u8>>>,
-	close_sender: Option<UnboundedSender<()>>,
 	socket_path: String,
+	on_data_handler: Option<Arc<JunoModuleImpl>>,
+	write_data_sender: Option<UnboundedSender<Buffer>>,
+	close_sender: Option<UnboundedSender<()>>,
 }
 
 impl InetSocketConnection {
 	pub fn new(socket_path: String) -> Self {
 		InetSocketConnection {
 			connection_setup: false,
-			read_data_receiver: None,
+			socket_path,
+			on_data_handler: None,
 			write_data_sender: None,
 			close_sender: None,
-			socket_path,
 		}
 	}
 }
 
+#[async_trait]
+impl BaseConnection for InetSocketConnection {
+	async fn setup_connection(&mut self) -> Result<(), Error> {
+		if self.connection_setup {
+			panic!("Cannot call setup_connection() more than once!");
+		}
+		let (write_data_sender, write_data_receiver) = unbounded::<Vec<u8>>();
+		let (close_sender, close_receiver) = unbounded::<()>();
+		let (init_sender, init_receiver) = channel::<Result<(), Error>>();
+
+		self.write_data_sender = Some(write_data_sender);
+		self.close_sender = Some(close_sender);
+		let socket_path = self.socket_path.clone();
+		let juno_module_impl = self.on_data_handler.as_ref().unwrap().clone();
+
+		task::spawn(async {
+			read_data_from_socket(
+				socket_path,
+				init_sender,
+				juno_module_impl,
+				write_data_receiver,
+				close_receiver,
+			)
+			.await;
+		});
+
+		self.connection_setup = true;
+		init_receiver.await.unwrap()
+	}
+
+	async fn close_connection(&mut self) -> Result<(), Error> {
+		if !self.connection_setup || self.close_sender.is_none() {
+			panic!("Cannot close a connection that hasn't been established yet. Did you forget to call setup_connection()?");
+		}
+		self.close_sender.as_ref().unwrap().send(()).await.unwrap();
+		Ok(())
+	}
+
+	async fn send(&mut self, buffer: Buffer) -> Result<(), Error> {
+		if !self.connection_setup || self.write_data_sender.is_none() {
+			panic!("Cannot send data to a connection that hasn't been established yet. Did you forget to await the call to setup_connection()?");
+		}
+		self.write_data_sender
+			.as_ref()
+			.unwrap()
+			.send(buffer)
+			.await
+			.unwrap();
+		Ok(())
+	}
+
+	fn set_data_listener(&mut self, listener: Arc<JunoModuleImpl>) {
+		self.on_data_handler = Some(listener);
+	}
+
+	fn get_data_listener(&self) -> &Option<Arc<JunoModuleImpl>> {
+		&self.on_data_handler
+	}
+}
+
 async fn read_data_from_socket(
-	port: String,
+	connect_addr: String,
 	init_sender: Sender<Result<(), Error>>,
-	mut read_sender: UnboundedSender<Vec<u8>>,
+	juno_impl: Arc<JunoModuleImpl>,
 	mut write_receiver: UnboundedReceiver<Vec<u8>>,
 	mut close_receiver: UnboundedReceiver<()>,
 ) {
-	let result = TcpStream::connect(port).await;
+	let result = TcpStream::connect(connect_addr).await;
 	if let Err(err) = result {
 		init_sender
 			.send(Err(Error::Internal(format!("{}", err))))
@@ -70,10 +131,10 @@ async fn read_data_from_socket(
 				read_or_write_future = future::select(read_future, write_future);
 				// Send the read data to the MPSC sender
 				if let Some(Ok(line)) = read_future_result {
-					let result = read_sender.send(line.as_bytes().to_vec()).await;
-					if let Err(err) = result {
-						println!("Error queing data from the socket to the module: {}", err);
-					}
+					let juno_impl = juno_impl.clone();
+					task::spawn(async move {
+						juno_impl.on_data(line.as_bytes().to_vec()).await;
+					});
 				}
 			}
 			Either::Right((write_future_result, next_read_future)) => {
@@ -94,76 +155,6 @@ async fn read_data_from_socket(
 	// Either a read, nor a write event has happened.
 	// This means the socket close event happened. Shutdown the socket and close any mpsc channels
 	drop(lines);
-	let result = read_sender.close().await;
-	if let Err(err) = result {
-		println!("Error closing the MPSC sender to queue data: {}", err);
-	}
 	write_receiver.close();
 	close_receiver.close();
 }
-
-#[async_trait]
-impl BaseConnection for InetSocketConnection {
-	async fn setup_connection(&mut self) -> Result<(), Error> {
-		if self.connection_setup {
-			panic!("Cannot call setup_connection() more than once!");
-		}
-		let (read_data_sender, read_data_receiver) = unbounded::<Vec<u8>>();
-		let (write_data_sender, write_data_receiver) = unbounded::<Vec<u8>>();
-		let (close_sender, close_receiver) = unbounded::<()>();
-		let (init_sender, init_receiver) = channel::<Result<(), Error>>();
-
-		self.read_data_receiver = Some(read_data_receiver);
-		self.write_data_sender = Some(write_data_sender);
-		self.close_sender = Some(close_sender);
-		let socket_path = self.socket_path.clone();
-
-		task::spawn(async {
-			read_data_from_socket(
-				socket_path,
-				init_sender,
-				read_data_sender,
-				write_data_receiver,
-				close_receiver,
-			)
-			.await;
-		});
-
-		self.connection_setup = true;
-		init_receiver.await.unwrap()
-	}
-
-	async fn close_connection(&mut self) {
-		if !self.connection_setup || self.close_sender.is_none() {
-			panic!("Cannot close a connection that hasn't been established yet. Did you forget to call setup_connection()?");
-		}
-		let mut sender = &self.close_sender.as_ref().unwrap().clone();
-		if let Err(err) = sender.send(()).await {
-			println!("Error attempting to close connection: {}", err);
-		}
-	}
-
-	async fn send(&mut self, buffer: Buffer) {
-		if !self.connection_setup || self.write_data_sender.is_none() {
-			panic!("Cannot send data to a connection that hasn't been established yet. Did you forget to await the call to setup_connection()?");
-		}
-		let mut sender = &self.write_data_sender.as_ref().unwrap().clone();
-		if let Err(err) = sender.send(buffer).await {
-			println!("Error attempting to send data to connection: {}", err);
-		}
-	}
-
-	fn get_data_receiver(&mut self) -> UnboundedReceiver<Buffer> {
-		if !self.connection_setup || self.read_data_receiver.is_none() {
-			panic!("Cannot get read sender to a connection that hasn't been established yet. Did you forget to await the call to setup_connection()?");
-		}
-		self.read_data_receiver.take().unwrap()
-	}
-
-	fn clone_write_sender(&self) -> UnboundedSender<Buffer> {
-		if !self.connection_setup || self.write_data_sender.is_none() {
-			panic!("Cannot get write sender of a connection that hasn't been established yet. Did you forget to await the call to setup_connection()?");
-		}
-		self.write_data_sender.as_ref().unwrap().clone()
-	}
-}