Start work on handling TLS for the RPC endpoint

zed-rpc/src/peer.rs

@@ -1,7 +1,11 @@
 use crate::proto::{self, EnvelopedMessage, MessageStream, RequestMessage};
 use anyhow::{anyhow, Context, Result};
 use async_lock::{Mutex, RwLock};
-use futures::{future::BoxFuture, AsyncRead, AsyncWrite, FutureExt};
+use futures::{
+    future::BoxFuture,
+    io::{ReadHalf, WriteHalf},
+    AsyncRead, AsyncReadExt, AsyncWrite, FutureExt,
+};
 use postage::{
     mpsc,
     prelude::{Sink, Stream},
@@ -72,13 +76,13 @@ struct Connection {
     response_channels: ResponseChannels,
 }
 
-pub struct ConnectionHandler<Conn> {
+pub struct ConnectionHandler<ReadConn, WriteConn> {
     peer: Arc<Peer>,
     connection_id: ConnectionId,
     response_channels: ResponseChannels,
     outgoing_rx: mpsc::Receiver<proto::Envelope>,
-    reader: MessageStream<Conn>,
-    writer: MessageStream<Conn>,
+    reader: MessageStream<ReadConn>,
+    writer: MessageStream<WriteConn>,
 }
 
 type ResponseChannels = Arc<Mutex<HashMap<u32, mpsc::Sender<proto::Envelope>>>>;
@@ -131,9 +135,12 @@ impl Peer {
     pub async fn add_connection<Conn>(
         self: &Arc<Self>,
         conn: Conn,
-    ) -> (ConnectionId, ConnectionHandler<Conn>)
+    ) -> (
+        ConnectionId,
+        ConnectionHandler<ReadHalf<Conn>, WriteHalf<Conn>>,
+    )
     where
-        Conn: Clone + AsyncRead + AsyncWrite + Unpin + Send + 'static,
+        Conn: AsyncRead + AsyncWrite + Unpin + Send + 'static,
     {
         let connection_id = ConnectionId(
             self.next_connection_id
@@ -145,13 +152,15 @@ impl Peer {
             next_message_id: Default::default(),
             response_channels: Default::default(),
         };
+
+        let (read_conn, write_conn) = conn.split();
         let handler = ConnectionHandler {
             peer: self.clone(),
             connection_id,
             response_channels: connection.response_channels.clone(),
             outgoing_rx,
-            reader: MessageStream::new(conn.clone()),
-            writer: MessageStream::new(conn),
+            reader: MessageStream::new(read_conn),
+            writer: MessageStream::new(write_conn),
         };
         self.connections
             .write()
@@ -291,9 +300,10 @@ impl Peer {
     }
 }
 
-impl<Conn> ConnectionHandler<Conn>
+impl<ReadConn, WriteConn> ConnectionHandler<ReadConn, WriteConn>
 where
-    Conn: Clone + AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    ReadConn: AsyncRead + Unpin + Send + 'static,
+    WriteConn: AsyncWrite + Unpin + Send + 'static,
 {
     pub async fn run(mut self) -> Result<()> {
         loop {