[lafere] enable basic server requests #1

Author: soerenmeier

lafere/Cargo.toml

@@ -13,6 +13,11 @@ name = "request_response"
 test = true
 required-features = ["connection"]
 
+[[example]]
+name = "server_requests"
+test = true
+required-features = ["connection"]
+
 [features]
 default = ["connection"]
 connection = ["tokio"]

lafere/examples/request_response.rs

@@ -168,6 +168,6 @@ async fn main() {
 }
 
 #[test]
-fn run_main() {
+fn request_response_main() {
 	main();
 }

lafere/examples/server_requests.rs

@@ -0,0 +1,201 @@
+use std::time::Duration;
+
+use tokio::io;
+
+use lafere::packet::{
+	BodyBytes, BodyBytesMut, Flags, Packet, PacketBytes, PacketError,
+	PacketHeader,
+};
+use lafere::server::Message;
+use lafere::{client, server};
+
+use bytes::{Bytes, BytesMut, BytesRead, BytesWrite};
+
+#[derive(Debug, Clone)]
+struct MyPacket<B> {
+	header: MyHeader,
+	bytes: B,
+}
+
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+struct MyHeader {
+	length: u32,
+	flags: Flags,
+	id: u32,
+}
+
+impl<B> MyPacket<B>
+where
+	B: PacketBytes,
+{
+	pub fn empty() -> Self {
+		Self {
+			header: MyHeader::default(),
+			bytes: B::new(MyHeader::LEN as usize),
+		}
+	}
+
+	pub fn body(&self) -> BodyBytes<'_> {
+		self.bytes.body()
+	}
+
+	pub fn body_mut(&mut self) -> BodyBytesMut<'_> {
+		self.bytes.body_mut()
+	}
+}
+
+impl<B> Packet<B> for MyPacket<B>
+where
+	B: PacketBytes,
+{
+	type Header = MyHeader;
+
+	fn header(&self) -> &Self::Header {
+		&self.header
+	}
+
+	fn header_mut(&mut self) -> &mut Self::Header {
+		&mut self.header
+	}
+
+	fn empty() -> Self {
+		Self {
+			header: MyHeader::default(),
+			bytes: B::new(MyHeader::LEN as usize),
+		}
+	}
+
+	fn from_bytes_and_header(
+		bytes: B,
+		header: Self::Header,
+	) -> Result<Self, PacketError> {
+		Ok(Self { header, bytes })
+	}
+
+	fn into_bytes(mut self) -> B {
+		self.header.length = self.bytes.body().len() as u32;
+		self.header.write_to(self.bytes.header_mut());
+		self.bytes
+	}
+}
+
+impl MyHeader {
+	fn write_to(&self, mut bytes: BytesMut) {
+		bytes.write_u32(self.length);
+		bytes.write_u8(self.flags.as_u8());
+		bytes.write_u32(self.id);
+	}
+}
+
+impl PacketHeader for MyHeader {
+	const LEN: u32 = 4 + 4 + 1;
+
+	fn from_bytes(mut bytes: Bytes) -> Result<Self, PacketError> {
+		Ok(Self {
+			length: bytes.read_u32(),
+			flags: Flags::from_u8(bytes.read_u8())?,
+			id: bytes.read_u32(),
+		})
+	}
+
+	fn body_len(&self) -> u32 {
+		self.length
+	}
+
+	fn flags(&self) -> &Flags {
+		&self.flags
+	}
+
+	fn set_flags(&mut self, flags: Flags) {
+		self.flags = flags;
+	}
+
+	fn id(&self) -> u32 {
+		self.id
+	}
+
+	fn set_id(&mut self, id: u32) {
+		self.id = id;
+	}
+}
+
+#[tokio::main]
+async fn main() {
+	let (client, server) = io::duplex(1024);
+
+	let mut alice = client::Connection::<MyPacket<_>>::new(
+		client,
+		client::Config {
+			timeout: Duration::from_secs(1),
+			body_limit: 1024,
+		},
+		None,
+	);
+
+	let mut bob = server::Connection::<MyPacket<_>>::new(
+		server,
+		server::Config {
+			timeout: Duration::from_secs(1),
+			body_limit: 1024,
+		},
+	);
+
+	let client_task = tokio::spawn(async move {
+		alice.enable_server_requests().await.unwrap();
+
+		// listen for a request from Bob
+		let msg = alice.receive().await.unwrap();
+		match msg {
+			Message::Request(req, resp_sender) => {
+				assert_eq!(req.body().as_slice(), b"Hello, Alice!");
+
+				let mut resp = MyPacket::empty();
+				resp.body_mut().write(b"Hello, Back!");
+				resp_sender.send(resp).unwrap();
+			}
+			_ => unreachable!(),
+		}
+
+		// send a request to Bob
+		let mut req = MyPacket::empty();
+		req.body_mut().write(b"Hello, Bob!");
+
+		let resp = alice.request(req).await.unwrap();
+		assert_eq!(resp.body().as_slice(), b"Hello, Back!");
+
+		assert!(alice.receive().await.is_none());
+	});
+
+	let server_task = tokio::spawn(async move {
+		// wait for Alice to enable server requests
+		let msg = bob.receive().await.unwrap();
+		assert!(matches!(msg, Message::EnableServerRequests));
+
+		// send a request to Alice
+		let mut req = MyPacket::empty();
+		req.body_mut().write(b"Hello, Alice!");
+
+		let resp = bob.request(req).await.unwrap();
+		assert_eq!(resp.body().as_slice(), b"Hello, Back!");
+
+		// listen for a request from Alice
+		let msg = bob.receive().await.unwrap();
+		match msg {
+			Message::Request(req, resp_sender) => {
+				assert_eq!(req.body().as_slice(), b"Hello, Bob!");
+
+				let mut resp = MyPacket::empty();
+				resp.body_mut().write(b"Hello, Back!");
+				resp_sender.send(resp).unwrap();
+			}
+			_ => unreachable!(),
+		}
+	});
+
+	tokio::try_join!(client_task, server_task).unwrap();
+}
+
+#[test]
+fn server_requests_main() {
+	main();
+}

lafere/src/client.rs

@@ -1,9 +1,11 @@
 use crate::error::{RequestError, TaskError};
+use crate::handler::server::Receiver;
 use crate::handler::{
 	Configurator, StreamReceiver, StreamSender, TaskHandle, client::Sender,
 };
 use crate::packet::{Packet, PlainBytes};
 use crate::plain;
+use crate::server::Request;
 use crate::util::{ByteStream, PinnedFuture};
 
 #[cfg(feature = "encrypted")]
@@ -45,7 +47,10 @@ impl<S> ReconStrat<S> {
 
 /// A connection to a server
 pub struct Connection<P> {
-	sender: Sender<P, Config>,
+	sender: Sender<P>,
+	receiver: Receiver<P>,
+	receiver_enabled: bool,
+	config: Configurator<Config>,
 	task: TaskHandle,
 }
 
@@ -82,19 +87,45 @@ impl<P> Connection<P> {
 	}
 
 	/// Creates a new Stream.
-	pub(crate) fn new_raw(sender: Sender<P, Config>, task: TaskHandle) -> Self {
-		Self { sender, task }
+	pub(crate) fn new_raw(
+		sender: Sender<P>,
+		receiver: Receiver<P>,
+		config: Configurator<Config>,
+		task: TaskHandle,
+	) -> Self {
+		Self {
+			sender,
+			receiver,
+			receiver_enabled: false,
+			config,
+			task,
+		}
 	}
 
 	/// Update the connection configuration
 	pub fn update_config(&self, cfg: Config) {
-		self.sender.update_config(cfg);
+		self.config.update(cfg);
 	}
 
 	/// Get's a `Configurator` which allows to configure this connection
 	/// without needing to have access to the connection
 	pub fn configurator(&self) -> Configurator<Config> {
-		self.sender.configurator()
+		self.config.clone()
+	}
+
+	pub async fn enable_server_requests(&mut self) -> Result<(), RequestError> {
+		self.sender.enable_server_requests().await?;
+		self.receiver_enabled = true;
+
+		Ok(())
+	}
+
+	/// ## Panics
+	/// - If server requests are not enabled
+	pub async fn receive(&mut self) -> Option<Request<P>> {
+		assert!(self.receiver_enabled);
+
+		self.receiver.receive().await
 	}
 
 	/// Send a request waiting until a response is available or the connection

lafere/src/encrypted/stream.rs

@@ -1,10 +1,10 @@
 use super::handshake::{client_handshake, server_handshake};
 use crate::client::{Config as ClientConfig, Connection as Client, ReconStrat};
 use crate::error::TaskError;
-use crate::handler::TaskHandle;
 use crate::handler::handler::{
 	Handler, PacketStream, bg_stream, bg_stream_reconnect,
 };
+use crate::handler::{Configurator, TaskHandle};
 use crate::packet::builder::{PacketReceiver, PacketReceiverError};
 use crate::packet::{EncryptedBytes, Packet};
 use crate::server::{Config as ServerConfig, Connection as Server};
@@ -30,7 +30,8 @@ where
 	P: Packet<EncryptedBytes> + Send + 'static,
 	P::Header: Send,
 {
-	let (sender, _, mut cfg_rx, mut bg_handler) = Handler::new(cfg, false);
+	let (sender, receiver, mut bg_handler) = Handler::new(false);
+	let (cfg_tx, mut cfg_rx) = Configurator::new(cfg);
 
 	let (tx_close, mut rx_close) = oneshot::channel();
 	let task = tokio::spawn(async move {
@@ -56,7 +57,7 @@ where
 		task,
 	};
 
-	Client::new_raw(sender, task)
+	Client::new_raw(sender, receiver, cfg_tx, task)
 }
 
 pub fn server<S, P>(
@@ -69,7 +70,8 @@ where
 	P: Packet<EncryptedBytes> + Send + 'static,
 	P::Header: Send,
 {
-	let (_, receiver, mut cfg_rx, mut bg_handler) = Handler::new(cfg, true);
+	let (sender, receiver, mut bg_handler) = Handler::new(true);
+	let (cfg_tx, mut cfg_rx) = Configurator::new(cfg);
 
 	let (tx_close, mut rx_close) = oneshot::channel();
 	let task = tokio::spawn(async move {
@@ -100,7 +102,7 @@ where
 		task,
 	};
 
-	Server::new_raw(receiver, task)
+	Server::new_raw(sender, receiver, cfg_tx, task)
 }
 
 /// inner manages a stream

lafere/src/handler/client.rs

@@ -1,17 +1,23 @@
-use super::{Configurator, StreamReceiver, StreamSender};
+use super::{StreamReceiver, StreamSender};
 use crate::error::RequestError;
 use crate::handler::handler::InternalRequest;
-use crate::util::watch;
 
 use tokio::sync::{mpsc, oneshot};
 
 /// A sender that sends messages to the handler.
-pub struct Sender<P, C> {
+pub struct Sender<P> {
 	pub(super) inner: mpsc::Sender<InternalRequest<P>>,
-	pub(super) cfg: watch::Sender<C>,
 }
 
-impl<P, C> Sender<P, C> {
+impl<P> Sender<P> {
+	// can only be sent from a client
+	pub async fn enable_server_requests(&self) -> Result<(), RequestError> {
+		self.inner
+			.send(InternalRequest::EnableServerRequests)
+			.await
+			.map_err(|_| RequestError::ConnectionAlreadyClosed)
+	}
+
 	/// Send a request waiting until a response is available.
 	pub async fn request(&self, packet: P) -> Result<P, RequestError> {
 		let (tx, rx) = oneshot::channel();
@@ -50,21 +56,12 @@ impl<P, C> Sender<P, C> {
 
 		Ok(StreamReceiver::new(rx))
 	}
-
-	pub fn update_config(&self, cfg: C) {
-		self.cfg.send(cfg);
-	}
-
-	pub fn configurator(&self) -> Configurator<C> {
-		Configurator::new(self.cfg.clone())
-	}
 }
 
-impl<P, C> Clone for Sender<P, C> {
+impl<P> Clone for Sender<P> {
 	fn clone(&self) -> Self {
 		Self {
 			inner: self.inner.clone(),
-			cfg: self.cfg.clone(),
 		}
 	}
 }