src/lib.rs

   1 //! This crate provides a simple brokerless message-queue abstraction over asynchronous network
   2 //! streams.
   3 //!
   4 //! # Examples
   5 //! Please use the [example programs](https://github.com/sachanganesh/connect-rs/tree/main/examples)
   6 //! provided to help understand crate usage.
   7 //!
   8 //! # Why?
   9 //! When building networked applications, developers shouldn't have to focus on repeatedly solving
  10 //! the problem of reliable, fault-tolerant message delivery over byte-streams. By using a message
  11 //! queue abstraction, crate users can focus on core application logic and leave the low-level
  12 //! networking and message-queue guarantees to the abstraction.
  13 //!
  14
  15 #![feature(doc_cfg)]
  16
  17 mod protocol;
  18 mod reader;
  19 pub mod tcp;
  20 mod writer;
  21
  22 #[cfg(feature = "tls")]
  23 #[doc(cfg(feature = "tls"))]
  24 pub mod tls;
  25
  26 use async_std::{net::SocketAddr, pin::Pin};
  27 use futures::{AsyncRead, AsyncWrite};
  28
  29 pub use crate::protocol::{ConnectDatagram, DatagramError};
  30 pub use crate::reader::ConnectionReader;
  31 pub use crate::writer::{ConnectionWriteError, ConnectionWriter};
  32 pub use futures::{SinkExt, StreamExt};
  33
  34 /// Wrapper around a [`ConnectionReader`] and [`ConnectionWriter`] to read and write on a network
  35 /// connection.
  36 pub struct Connection {
  37     reader: ConnectionReader,
  38     writer: ConnectionWriter,
  39 }
  40
  41 #[allow(dead_code)]
  42 impl Connection {
  43     pub(crate) fn new(
  44         local_addr: SocketAddr,
  45         peer_addr: SocketAddr,
  46         read_stream: Pin<Box<dyn AsyncRead + Send + Sync>>,
  47         write_stream: Pin<Box<dyn AsyncWrite + Send + Sync>>,
  48     ) -> Self {
  49         Self {
  50             reader: ConnectionReader::new(local_addr, peer_addr, read_stream),
  51             writer: ConnectionWriter::new(local_addr, peer_addr, write_stream),
  52         }
  53     }
  54
  55     /// Get the local IP address and port.
  56     pub fn local_addr(&self) -> SocketAddr {
  57         self.reader.local_addr()
  58     }
  59
  60     /// Get the peer IP address and port.
  61     pub fn peer_addr(&self) -> SocketAddr {
  62         self.reader.peer_addr()
  63     }
  64
  65     /// Consume the [`Connection`] to split into separate [`ConnectionReader`] and
  66     /// [`ConnectionWriter`] halves.
  67     ///
  68     /// [`Connection`]s are split when reading and writing must be concurrent operations.
  69     pub fn split(self) -> (ConnectionReader, ConnectionWriter) {
  70         (self.reader, self.writer)
  71     }
  72
  73     /// Re-wrap the [`ConnectionReader`] and [`ConnectionWriter`] halves into a [`Connection`].
  74     pub fn join(reader: ConnectionReader, writer: ConnectionWriter) -> Self {
  75         Self { reader, writer }
  76     }
  77
  78     /// Get mutable access to the underlying [`ConnectionReader`].
  79     pub fn reader(&mut self) -> &mut ConnectionReader {
  80         &mut self.reader
  81     }
  82
  83     /// Get mutable access to the underlying [`ConnectionWriter`].
  84     pub fn writer(&mut self) -> &mut ConnectionWriter {
  85         &mut self.writer
  86     }
  87
  88     /// Close the connection by closing both the reading and writing halves.
  89     pub async fn close(self) -> SocketAddr {
  90         let peer_addr = self.peer_addr();
  91         let (reader, writer) = self.split();
  92
  93         drop(reader);
  94
  95         // writer.close().await;
  96         drop(writer);
  97
  98         return peer_addr;
  99     }
 100 }
 101
 102 #[cfg(test)]
 103 mod tests {}