Merge branch 'feature/arc-module-impl' into feature/non-mpsc-connection

Author: rakshith-ravi

src/connection/inet_socket_connection.rs

@@ -2,9 +2,11 @@ use crate::{
 	connection::{BaseConnection, Buffer},
 	utils::{Error, READ_BUFFER_SIZE},
 };
-use std::time::Duration;
 
-use async_std::{io, net::Shutdown, net::TcpStream, prelude::*};
+use async_std::{
+	net::{Shutdown, TcpStream},
+	prelude::*,
+};
 use async_trait::async_trait;
 
 pub struct InetSocketConnection {
@@ -33,8 +35,7 @@ impl BaseConnection for InetSocketConnection {
 		if let Err(err) = result {
 			return Err(Error::Internal(format!("{}", err)));
 		}
-		let client = result.unwrap();
-		self.client = Some(client);
+		self.client = Some(result.unwrap());
 
 		self.connection_setup = true;
 		Ok(())
@@ -69,15 +70,14 @@ impl BaseConnection for InetSocketConnection {
 			let client = self.client.as_mut().unwrap();
 			let mut buffer = Vec::new();
 			let mut read_size = READ_BUFFER_SIZE;
-
 			while read_size > 0 {
 				let mut buf = [0u8; READ_BUFFER_SIZE];
-				let result = io::timeout(Duration::from_millis(10), client.read(&mut buf)).await;
+				let result = client.read(&mut buf).await;
 				if result.is_err() {
-					return Some(buffer);
+					return None;
 				}
 				read_size = result.unwrap();
-				buffer.extend(buf[..read_size].into_iter());
+				buffer.extend(buf[..read_size].iter());
 			}
 			Some(buffer)
 		}

src/connection/unix_socket_connection.rs

@@ -77,7 +77,7 @@ impl BaseConnection for UnixSocketConnection {
 					return Some(buffer);
 				}
 				read_size = result.unwrap();
-				buffer.extend(buf[..read_size].into_iter());
+				buffer.extend(buf[..read_size].iter());
 			}
 			Some(buffer)
 		}

src/juno_module_impl.rs

@@ -9,14 +9,17 @@ use async_std::sync::{Mutex, RwLock};
 use futures::channel::oneshot::Sender;
 use std::collections::HashMap;
 
+type HookListeners = RwLock<HashMap<String, Vec<fn(Value)>>>;
+type Functions = RwLock<HashMap<String, fn(HashMap<String, Value>) -> Value>>;
+
 // Create separate rwlocks for each individual element
 // Such that each one of them can be individually read or written independent of the other
 pub(crate) struct JunoModuleImpl {
 	pub(crate) protocol: RwLock<BaseProtocol>,
 	pub(crate) connection: RwLock<Box<dyn BaseConnection + Send + Sync>>,
 	pub(crate) requests: RwLock<HashMap<String, Sender<Result<Value>>>>,
-	pub(crate) functions: RwLock<HashMap<String, fn(HashMap<String, Value>) -> Value>>,
-	pub(crate) hook_listeners: RwLock<HashMap<String, Vec<fn(Value)>>>,
+	pub(crate) functions: Functions,
+	pub(crate) hook_listeners: HookListeners,
 	pub(crate) message_buffer: Mutex<Buffer>,
 	pub(crate) registered: RwLock<bool>,
 }

src/protocol/base_protocol.rs

@@ -126,4 +126,11 @@ impl BaseProtocol {
 			_ => panic!("Currently, only JsonProtocol is supported"),
 		}
 	}
+
+	pub fn decode(&self, data: &[u8]) -> BaseMessage {
+		match self {
+			BaseProtocol::JsonProtocol { .. } => json_protocol::decode(&self, data),
+			_ => panic!("Currently, only JsonProtocol is supported"),
+		}
+	}
 }

src/protocol/json_protocol.rs

@@ -164,6 +164,18 @@ pub fn get_next_message(protocol: &mut BaseProtocol) -> Option<BaseMessage> {
 	}
 }
 
+pub fn decode(protocol: &BaseProtocol, data: &[u8]) -> BaseMessage {
+	match protocol {
+		BaseProtocol::JsonProtocol { .. } => match decode_internal(data) {
+			Some(message) => message,
+			None => BaseMessage::Unknown {
+				request_id: String::default(),
+			},
+		},
+		_ => panic!("BaseProtocol tried to decode a non-JsonProtocol as a JsonProtocol"),
+	}
+}
+
 fn decode_internal(data: &[u8]) -> Option<BaseMessage> {
 	let result: Result<Value> = from_slice(data);
 	if result.is_err() {