src/lib.rs

   1 //! This crate provides a simple brokerless message-queue abstraction over asynchronous network
   2 //! streams.
   3 //!
   4 //! # Why?
   5 //! When building networked applications, developers shouldn't have to focus on repeatedly solving
   6 //! the problem of reliable, fault-tolerant message delivery over byte-streams. By using a message
   7 //! queue abstraction, crate users can focus on core application logic and leave the low-level
   8 //! networking and message-queue guarantees to the abstraction.
   9 //!
  10 //! # Examples
  11 //! Please use the [examples](https://github.com/sachanganesh/connect-rs/tree/main/examples)
  12 //! provided to help understand crate usage.
  13 //!
  14 //! # Protobuf
  15 //! This crate relies on the use of [Protocol Buffers](https://developers.google.com/protocol-buffers)
  16 //! due to it being widely adopted and industry-proven. All messages are Protobuf messages that
  17 //! are packed into a Protobuf `Any` type and then sent over the wire. Message recipients must
  18 //! decide what Protobuf message type it is, and correspondingly unpack the `Any` into a particular
  19 //! message type.
  20 //!
  21 //! Protobuf was chosen when the library hit a roadblock with Rust's `TypeId` potentially not being
  22 //! consistent between Rust compiler versions. The crate requires a consistent way to determine what
  23 //! type of message is received, so it can appropriately deserialize the message from network bytes.
  24 //! Until the Rust ecosystem around reflection improves, the crate will use Protobuf to fill the
  25 //! void.
  26
  27 pub mod protocol;
  28 mod reader;
  29 pub mod tcp;
  30 pub mod tls;
  31 mod writer;
  32
  33 pub use crate::protocol::{ConnectDatagram, DatagramEmptyError};
  34 pub use crate::reader::ConnectionReader;
  35 pub use crate::writer::ConnectionWriter;
  36 use async_std::{net::SocketAddr, pin::Pin};
  37 use futures::{AsyncRead, AsyncWrite};
  38 pub use futures::{SinkExt, StreamExt};
  39
  40 /// Wrapper around a [`ConnectionReader`] and [`ConnectionWriter`] to read and write on a network
  41 /// connection
  42 pub struct Connection {
  43     local_addr: SocketAddr,
  44     peer_addr: SocketAddr,
  45     reader: ConnectionReader,
  46     writer: ConnectionWriter,
  47 }
  48
  49 #[allow(dead_code)]
  50 impl Connection {
  51     pub(crate) fn new(
  52         local_addr: SocketAddr,
  53         peer_addr: SocketAddr,
  54         read_stream: Pin<Box<dyn AsyncRead + Send + Sync>>,
  55         write_stream: Pin<Box<dyn AsyncWrite + Send + Sync>>,
  56     ) -> Self {
  57         Self {
  58             local_addr,
  59             peer_addr,
  60             reader: ConnectionReader::new(local_addr, peer_addr, read_stream),
  61             writer: ConnectionWriter::new(local_addr, peer_addr, write_stream),
  62         }
  63     }
  64
  65     /// Get the local IP address and port
  66     pub fn local_addr(&self) -> SocketAddr {
  67         self.local_addr.clone()
  68     }
  69
  70     /// Get the peer IP address and port
  71     pub fn peer_addr(&self) -> SocketAddr {
  72         self.peer_addr.clone()
  73     }
  74
  75     /// Consume the [`Connection`] to split into separate [`ConnectionReader`] and
  76     /// [`ConnectionWriter`] halves
  77     ///
  78     /// [`Connection`]s are split when reading and writing must be concurrent operations.
  79     pub fn split(self) -> (ConnectionReader, ConnectionWriter) {
  80         (self.reader, self.writer)
  81     }
  82
  83     /// Re-wrap the [`ConnectionReader`] and [`ConnectionWriter`] halves into a [`Connection`]
  84     pub fn join(reader: ConnectionReader, writer: ConnectionWriter) -> Self {
  85         Self {
  86             local_addr: reader.local_addr(),
  87             peer_addr: reader.peer_addr(),
  88             reader,
  89             writer,
  90         }
  91     }
  92
  93     /// Get mutable access to the underlying [`ConnectionReader`]
  94     pub fn reader(&mut self) -> &mut ConnectionReader {
  95         &mut self.reader
  96     }
  97
  98     /// Get mutable access to the underlying [`ConnectionWriter`]
  99     pub fn writer(&mut self) -> &mut ConnectionWriter {
 100         &mut self.writer
 101     }
 102
 103     /// Close the connection by closing both the reading and writing halves
 104     pub async fn close(self) -> SocketAddr {
 105         let peer_addr = self.peer_addr();
 106         let (reader, writer) = self.split();
 107
 108         drop(reader);
 109
 110         // writer.close().await;
 111         drop(writer);
 112
 113         return peer_addr;
 114     }
 115 }