diff --git a/Cargo.lock b/Cargo.lock
index 02a1cd8c91de7ae84a3727ec90623a0116d94fea..ebf809b7302dfd5b2b19090f253b72828c95baf1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -67,6 +67,95 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
+[[package]]
+name = "futures"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-channel"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+]
+
+[[package]]
+name = "futures-core"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
+
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-io"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
+
+[[package]]
+name = "futures-macro"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "futures-sink"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7"
+
+[[package]]
+name = "futures-task"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988"
+
+[[package]]
+name = "futures-util"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-macro",
+ "futures-sink",
+ "futures-task",
+ "memchr",
+ "pin-project-lite",
+ "pin-utils",
+ "slab",
+]
+
[[package]]
name = "gimli"
version = "0.31.1"
@@ -162,6 +251,12 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b"
+[[package]]
+name = "pin-utils"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+
[[package]]
name = "proc-macro2"
version = "1.0.94"
@@ -218,6 +313,15 @@ dependencies = [
"libc",
]
+[[package]]
+name = "slab"
+version = "0.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67"
+dependencies = [
+ "autocfg",
+]
+
[[package]]
name = "smallvec"
version = "1.14.0"
@@ -245,6 +349,17 @@ dependencies = [
"unicode-ident",
]
+[[package]]
+name = "tcp"
+version = "0.1.0"
+dependencies = [
+ "async-trait",
+ "futures",
+ "rust-networking-examples",
+ "tokio",
+ "tokio-util",
+]
+
[[package]]
name = "tokio"
version = "1.44.1"
@@ -274,6 +389,19 @@ dependencies = [
"syn",
]
+[[package]]
+name = "tokio-util"
+version = "0.7.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "66a539a9ad6d5d281510d5bd368c973d636c02dbf8a67300bfb6b950696ad7df"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
[[package]]
name = "unicode-ident"
version = "1.0.18"
diff --git a/Cargo.toml b/Cargo.toml
index 0198f9d88c4d126e1a7978a6947ebf21c36246a4..df86bbfc06e4e1e5a55589ca227a4cf21f44f1fb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,7 +6,7 @@ edition = "2021"
[workspace]
members = [
"examples/local"
-]
+, "examples/tcp"]
[dependencies]
tokio = { version = "1.44.1", features = ["full"] }
diff --git a/examples/tcp/Cargo.toml b/examples/tcp/Cargo.toml
new file mode 100644
index 0000000000000000000000000000000000000000..beb69655470bd55716354ef8db4eb15f59f09063
--- /dev/null
+++ b/examples/tcp/Cargo.toml
@@ -0,0 +1,14 @@
+[package]
+name = "tcp"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+# For the interface
+rust-networking-examples = { path = "../../" }
+
+# Async related
+tokio = { version = "1.44.1", features = ["full"] }
+tokio-util = { version = "0.7.15", features = ["codec"] }
+async-trait = "0.1.87"
+futures = "0.3.31"
diff --git a/examples/tcp/src/main.rs b/examples/tcp/src/main.rs
new file mode 100644
index 0000000000000000000000000000000000000000..c7d87096bb4f63f38568ceb374a11f587a64ad2f
--- /dev/null
+++ b/examples/tcp/src/main.rs
@@ -0,0 +1,30 @@
+use std::time::Duration;
+
+use tcp::TcpConnector;
+use rust_networking_examples::Network;
+use tokio::time::sleep;
+
+mod tcp;
+
+#[tokio::main]
+async fn main() {
+ let mut c_a = TcpConnector::new("127.0.0.1:3000");
+ let mut c_b = TcpConnector::new("127.0.0.1:3001");
+
+ let mut r_a = c_a.recv().await;
+
+ tokio::spawn(async move {
+ while let Some((from, msg)) = r_a.recv().await {
+ println!("Got {}, {:?}", from, msg);
+ }
+ });
+
+ sleep(Duration::from_secs(2)).await;
+
+ c_b.send("127.0.0.1:3000", "Hello world!".as_bytes().to_vec());
+ c_b.send("127.0.0.1:3000", "Hello world!".as_bytes().to_vec());
+ c_b.send("127.0.0.1:3000", "Hello world!".as_bytes().to_vec());
+
+ sleep(Duration::from_secs(10)).await;
+ println!("Closing");
+}
diff --git a/examples/tcp/src/tcp/actor.rs b/examples/tcp/src/tcp/actor.rs
new file mode 100644
index 0000000000000000000000000000000000000000..7d105401ff50eb09f5b2a3df653a338a5d5d2914
--- /dev/null
+++ b/examples/tcp/src/tcp/actor.rs
@@ -0,0 +1,129 @@
+use tokio::net::TcpStream;
+use tokio::select;
+use std::collections::HashMap;
+use tokio::sync::{mpsc, oneshot};
+
+use super::connection::{ConnectionHandle, ReceivedEvent};
+use super::listen::listen_tcp;
+use super::DEFAULT_CHANNEL_SIZE;
+
+/// Messages that the Net actor knows how to process
+pub enum NetMessage {
+ NewConnection {
+ from: String,
+ socket: TcpStream
+ },
+ SendMessage {
+ to: String,
+ message: Vec<u8>,
+ },
+ ReceiveMessage {
+ from: String,
+ message: Vec<u8>
+ },
+ RegisterMessageHandler {
+ respond_to: oneshot::Sender<mpsc::Receiver<(String, Vec<u8>)>>
+ }
+}
+
+// Map address:port to connection handle(s)
+type Connections = HashMap<String, Vec<ConnectionHandle>>;
+
+/// Net actor, orchestrates the RPS, Listen and Connection Actors
+struct Actor {
+ /// Wallet message handlers
+ emit: Vec<mpsc::Sender<(String, Vec<u8>)>>,
+
+ /// Map of addr -> communication channel for currently active connection agents
+ connections: Connections,
+
+ connection_sender: mpsc::Sender<ReceivedEvent>,
+}
+
+#[derive(Clone)]
+pub struct ActorHandle {
+ pub sender: mpsc::Sender<NetMessage>
+}
+
+impl ActorHandle {
+ pub fn new(bind_address: String) -> Self {
+ // Channel to communicate with parent
+ let (sender, mut receiver) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ // Channel to receive incoming messages from connections
+ let (sender_connection, mut receiver_connection) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ tokio::spawn(async move {
+ // Channel to accept new connections
+ let mut listening = listen_tcp(&bind_address).await;
+
+ // Event loop
+ let mut actor = Actor::new(sender_connection);
+
+ loop {
+ select! {
+ event = receiver.recv() => {
+ match event {
+ Some(msg) => actor.handle_message(msg),
+ None => break, // Exit
+ };
+ },
+ Some((from, socket)) = listening.recv() => {
+ actor.handle_message(NetMessage::NewConnection { from, socket });
+ },
+ Some(ReceivedEvent { from, message }) = receiver_connection.recv() => {
+ actor.handle_message(NetMessage::ReceiveMessage { from, message });
+ }
+ }
+ }
+ });
+
+ ActorHandle { sender }
+ }
+}
+
+
+impl Actor {
+ pub fn new(connection_sender: mpsc::Sender<ReceivedEvent>) -> Self {
+ Actor { emit: vec![], connections: Connections::new(), connection_sender }
+ }
+
+ pub fn handle_message(&mut self, msg: NetMessage) {
+ match msg {
+ NetMessage::NewConnection { from, socket } => {
+ let handle = ConnectionHandle::from_socket(from.clone(), socket, self.connection_sender.clone());
+
+ self.connections
+ .entry(from)
+ .or_default()
+ .push(handle);
+ },
+ NetMessage::SendMessage { to, message } => {
+ // Purge stale connection handles that have already closed.
+ self.connections.entry(to.clone()).and_modify(|handles| {
+ handles.retain(|h| h.open());
+ });
+
+ // Send message to the first connection handle we find, or open a new one.
+ self.connections
+ .entry(to.clone())
+ .or_insert(vec![ConnectionHandle::new(to, self.connection_sender.clone())])
+ .first_mut()
+ .unwrap()
+ .send(message);
+ },
+ NetMessage::ReceiveMessage { from, message } => {
+ self.emit
+ .iter_mut()
+ .for_each(|sender| { let _ = sender.try_send((from.clone(), message.clone())); });
+ },
+ NetMessage::RegisterMessageHandler { respond_to } => {
+ let (sender, receiver) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ self.emit.push(sender);
+
+ respond_to.send(receiver).expect("Channel should be alive");
+ },
+ }
+ }
+}
\ No newline at end of file
diff --git a/examples/tcp/src/tcp/connection.rs b/examples/tcp/src/tcp/connection.rs
new file mode 100644
index 0000000000000000000000000000000000000000..56c1a62d5322adf682f5c60f77111f73ab2ccb6a
--- /dev/null
+++ b/examples/tcp/src/tcp/connection.rs
@@ -0,0 +1,134 @@
+use futures::{sink::SinkExt, StreamExt};
+use tokio::{io::{self, split}, net::TcpStream, select, sync::mpsc};
+use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
+
+use super::DEFAULT_CHANNEL_SIZE;
+
+// The event emitted by this actor, denoting a message and who it was received from.
+pub struct ReceivedEvent {
+ pub from: String,
+ pub message: Vec<u8>
+}
+
+pub enum ConnectionHandleMessage {
+ Send(Vec<u8>),
+ Receive(Vec<u8>)
+}
+
+#[derive(Clone)]
+pub struct ConnectionHandle {
+ pub sender: mpsc::Sender<ConnectionHandleMessage>
+}
+
+impl ConnectionHandle {
+ pub fn from_socket(address: String, socket: TcpStream, emit: mpsc::Sender<ReceivedEvent>) -> Self {
+ let (sender, receiver) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ event_loop(receiver, address, Some(socket), emit);
+
+ ConnectionHandle { sender }
+ }
+
+ pub fn new(address: String, emit: mpsc::Sender<ReceivedEvent>) -> Self {
+ let (sender, receiver) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ event_loop(receiver, address, None, emit);
+
+ ConnectionHandle { sender }
+ }
+
+ pub fn open(&self) -> bool {
+ !self.sender.is_closed()
+ }
+
+ pub fn send(&mut self, message: Vec<u8>) {
+ self.sender.try_send(ConnectionHandleMessage::Send(message)).expect("Actor should be alive");
+ }
+}
+
+/// Function for creating the actor and then running the event loop
+fn event_loop(mut receiver: mpsc::Receiver<ConnectionHandleMessage>, address: String, socket: Option<TcpStream>, emit: mpsc::Sender<ReceivedEvent>) {
+ tokio::spawn(async move {
+ let socket = match socket {
+ Some(s) => s,
+ None => {
+ TcpStream::connect(address.clone())
+ .await
+ .expect("Failed to connect to address")
+ },
+ };
+
+ // Start connection actor
+ let (writer, mut reader) = frame_and_split_socket(socket);
+
+ // Writer becomes and channel that we can send messsages down
+ let writer = start_writer_task(writer);
+
+ let mut actor = Actor { address: address.clone(), emit: emit.clone(), writer };
+
+ loop {
+ select!{
+ received = reader.next() => {
+ if let Some(message) = received {
+ let bytes = message.expect("Message should be readble");
+ actor.handle_message(ConnectionHandleMessage::Receive(bytes.into()));
+ } else {
+ break; // Connection finished
+ }
+ },
+ Some(msg) = receiver.recv() => {
+ actor.handle_message(msg);
+ },
+ else => break // Actor handle dropped
+ }
+ }
+ });
+}
+
+struct Actor {
+ address: String,
+ emit: mpsc::Sender<ReceivedEvent>,
+ writer: mpsc::Sender<Vec<u8>>,
+}
+
+impl Actor {
+ pub fn handle_message(&mut self, msg: ConnectionHandleMessage) {
+ match msg {
+ ConnectionHandleMessage::Send(message) => {
+ self.writer.try_send(message).expect("Writer task should be running");
+ },
+ ConnectionHandleMessage::Receive(message) => {
+ self.emit.try_send(ReceivedEvent { from: self.address.clone(), message }).expect("Should be able to emit events");
+ },
+ }
+ }
+}
+
+
+type ConnectionReader = FramedRead<io::ReadHalf<TcpStream>, LengthDelimitedCodec>;
+type ConnectionWriter = FramedWrite<io::WriteHalf<TcpStream>, LengthDelimitedCodec>;
+
+/// Split TcpStream into two seperate ownership objects; a writer (for sending) and a reader (for receiving).
+fn frame_and_split_socket(socket: TcpStream) -> (ConnectionWriter, ConnectionReader) {
+ // Split the TcpStream into read and write halves
+ let (read_half, write_half) = split(socket);
+
+ // Create FramedRead for receiving and FramedWrite for sending
+ let reader = FramedRead::new(read_half, LengthDelimitedCodec::new());
+ let writer = FramedWrite::new(write_half, LengthDelimitedCodec::new());
+
+ (writer, reader)
+}
+
+/// To write to the underlying TCP socket, we need an async context. This is setup by this function.
+fn start_writer_task(mut writer: ConnectionWriter) -> mpsc::Sender<Vec<u8>> {
+ let (sender, mut receiver) = mpsc::channel::<Vec<u8>>(DEFAULT_CHANNEL_SIZE);
+
+ tokio::spawn(async move {
+ while let Some(bytes) = receiver.recv().await {
+ writer.send(bytes.into()).await.expect("Should have sent message");
+ }
+ });
+
+ sender
+}
\ No newline at end of file
diff --git a/examples/tcp/src/tcp/listen.rs b/examples/tcp/src/tcp/listen.rs
new file mode 100644
index 0000000000000000000000000000000000000000..595de1a017382ecdd5c6220c5768886f0750c48a
--- /dev/null
+++ b/examples/tcp/src/tcp/listen.rs
@@ -0,0 +1,33 @@
+use tokio::select;
+use tokio::sync::mpsc;
+use tokio::net::{TcpListener, TcpStream};
+
+use super::DEFAULT_CHANNEL_SIZE;
+
+
+/// `listen_tcp` spawns an Actor that waits for inbound requests to open TCP connections, accepts them and then starts a connection actor to exchange messages with them.
+pub async fn listen_tcp(listen_addr: &str) -> mpsc::Receiver<(String, TcpStream)> {
+ let listener = TcpListener::bind(listen_addr).await.expect("Couldn't bind to listen address");
+
+ let (sender, receiver) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
+
+ tokio::spawn(async move {
+ loop {
+ select! {
+ // Accept all inbound TCP connections
+ incoming = listener.accept() => {
+ match incoming {
+ Ok((socket, socket_addr)) => {
+ sender.send((socket_addr.to_string(), socket)).await.expect("Parent actor should be alive and accepting new connections");
+ },
+ Err(error) => {
+ println!("TCP : error accepting connection {}", error);
+ }
+ }
+ }
+ }
+ }
+ });
+
+ receiver
+}
\ No newline at end of file
diff --git a/examples/tcp/src/tcp/mod.rs b/examples/tcp/src/tcp/mod.rs
new file mode 100644
index 0000000000000000000000000000000000000000..1ba975ccbda2872a0d453e46c1173c53768ff056
--- /dev/null
+++ b/examples/tcp/src/tcp/mod.rs
@@ -0,0 +1,37 @@
+// Actors
+mod actor;
+mod listen;
+mod connection;
+
+pub use actor::ActorHandle;
+use rust_networking_examples::Network;
+use tokio::sync::{mpsc, oneshot};
+
+const DEFAULT_CHANNEL_SIZE: usize = 1024;
+
+pub struct TcpConnector {
+ handle: ActorHandle
+}
+
+impl TcpConnector {
+ pub fn new(bind_address: &str) -> TcpConnector {
+ TcpConnector { handle: ActorHandle::new(String::from(bind_address)) }
+ }
+}
+
+#[async_trait::async_trait]
+impl Network for TcpConnector {
+ /// Send a message to the identifier specified by the 'to' field.
+ fn send(&mut self, to: &str, message: Vec<u8>) {
+ self.handle.sender.try_send(actor::NetMessage::SendMessage { to: String::from(to), message }).expect("Actor should be alive");
+ }
+
+ /// Generate one or more receivers to catch inbound messages.
+ async fn recv(&mut self) -> mpsc::Receiver<(String, Vec<u8>)> {
+ let (respond_to, receiver) = oneshot::channel();
+
+ self.handle.sender.try_send(actor::NetMessage::RegisterMessageHandler { respond_to }).expect("Actor should be alive");
+
+ receiver.await.expect("Should have receievd channel back")
+ }
+}