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 mod reader;
  28 pub(crate) mod schema;
  29 pub mod tcp;
  30 pub mod tls;
  31 mod writer;
  32
  33 pub use crate::reader::ConnectionReader;
  34 pub use crate::writer::ConnectionWriter;
  35 use async_std::{net::SocketAddr, pin::Pin};
  36 use futures::{AsyncRead, AsyncWrite};
  37 pub use futures::{SinkExt, StreamExt};
  38
  39 /// Wrapper around a [`ConnectionReader`] and [`ConnectionWriter`] to read and write on a network
  40 /// connection
  41 pub struct Connection {
  42     local_addr: SocketAddr,
  43     peer_addr: SocketAddr,
  44     reader: ConnectionReader,
  45     writer: ConnectionWriter,
  46 }
  47
  48 #[allow(dead_code)]
  49 impl Connection {
  50     pub(crate) fn new(
  51         local_addr: SocketAddr,
  52         peer_addr: SocketAddr,
  53         read_stream: Pin<Box<dyn AsyncRead + Send + Sync>>,
  54         write_stream: Pin<Box<dyn AsyncWrite + Send + Sync>>,
  55     ) -> Self {
  56         Self {
  57             local_addr,
  58             peer_addr,
  59             reader: ConnectionReader::new(local_addr, peer_addr, read_stream),
  60             writer: ConnectionWriter::new(local_addr, peer_addr, write_stream),
  61         }
  62     }
  63
  64     /// Get the local IP address and port
  65     pub fn local_addr(&self) -> SocketAddr {
  66         self.local_addr.clone()
  67     }
  68
  69     /// Get the peer IP address and port
  70     pub fn peer_addr(&self) -> SocketAddr {
  71         self.peer_addr.clone()
  72     }
  73
  74     /// Consume the [`Connection`] to split into separate [`ConnectionReader`] and
  75     /// [`ConnectionWriter`] halves
  76     ///
  77     /// [`Connection`]s are split when reading and writing must be concurrent operations.
  78     pub fn split(self) -> (ConnectionReader, ConnectionWriter) {
  79         (self.reader, self.writer)
  80     }
  81
  82     /// Re-wrap the [`ConnectionReader`] and [`ConnectionWriter`] halves into a [`Connection`]
  83     pub fn join(reader: ConnectionReader, writer: ConnectionWriter) -> Self {
  84         Self {
  85             local_addr: reader.local_addr(),
  86             peer_addr: reader.peer_addr(),
  87             reader,
  88             writer,
  89         }
  90     }
  91
  92     /// Get mutable access to the underlying [`ConnectionReader`]
  93     pub fn reader(&mut self) -> &mut ConnectionReader {
  94         &mut self.reader
  95     }
  96
  97     /// Get mutable access to the underlying [`ConnectionWriter`]
  98     pub fn writer(&mut self) -> &mut ConnectionWriter {
  99         &mut self.writer
 100     }
 101
 102     /// Close the connection by closing both the reading and writing halves
 103     pub async fn close(self) -> SocketAddr {
 104         let peer_addr = self.peer_addr();
 105         let (reader, writer) = self.split();
 106
 107         drop(reader);
 108
 109         // writer.close().await;
 110         drop(writer);
 111
 112         return peer_addr;
 113     }
 114 }