Merge pull request #4 from swimos/socket-closure

Socket closure

.github/workflows/ci.yml

@@ -11,7 +11,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: stable
+          toolchain: 1.56.1
           override: true
       - uses: actions-rs/cargo@v1
         with:
@@ -31,7 +31,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: stable
+          toolchain: 1.56.1
           override: true
       - uses: actions-rs/cargo@v1
         with:
@@ -46,7 +46,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: stable
+          toolchain: 1.56.1
           override: true
       - run: rustup component add rustfmt
       - uses: actions-rs/cargo@v1
@@ -62,7 +62,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: stable
+          toolchain: 1.56.1
           override: true
       - run: rustup component add clippy
       - uses: actions-rs/cargo@v1

ratchet_core/src/errors.rs

@@ -255,9 +255,18 @@ impl From<InvalidHeaderValue> for Error {
 }
 
 #[derive(Clone, Copy, Error, Debug, PartialEq)]
-/// The channel is already closed
+/// The channel is closed
 #[error("The channel is already closed")]
-pub struct CloseError;
+pub enum CloseCause {
+    /// The channel closed nominally. This is **not** an error and instead indicates a clean closure
+    /// of the channel by either ourselves or the peer.
+    #[error("The channel closed as expected")]
+    Stopped,
+    /// This is only produced when a user attempts to reuse a closed channel and instead indicates a
+    /// bug in your code.
+    #[error("Attempted to use a closed channel")]
+    Error,
+}
 
 /// WebSocket protocol errors.
 #[derive(Copy, Clone, Debug, PartialEq, Error)]

ratchet_core/src/framed/mod.rs

@@ -572,6 +572,10 @@ where
         )
         .await
     }
+
+    pub async fn close(&mut self) {
+        let _r = self.io.shutdown().await;
+    }
 }
 
 pub async fn read_next<I, E>(
@@ -624,7 +628,6 @@ where
             |_, _| Ok(()),
         )
         .await
-        .map_err(|e| Error::with_cause(ErrorKind::Close, e))
 }
 
 pub async fn write_fragmented<A, I, F>(

ratchet_core/src/lib.rs

@@ -26,6 +26,8 @@
     unused_import_braces
 )]
 
+extern crate core;
+
 #[cfg(test)]
 mod test_fixture;
 
@@ -40,14 +42,14 @@ mod ws;
 /// Split WebSocket implementation.
 #[cfg(feature = "split")]
 mod split;
+#[cfg(feature = "split")]
+pub use split::{Receiver, ReuniteError, Sender};
 
 #[allow(missing_docs)]
 #[cfg(feature = "fixture")]
 pub mod fixture {
     pub use super::protocol::write_text_frame_header;
 }
-#[cfg(feature = "split")]
-pub use split::{Receiver, ReuniteError, Sender};
 
 pub use builder::{WebSocketClientBuilder, WebSocketServerBuilder};
 pub use errors::*;
@@ -59,7 +61,7 @@ pub use handshake::{
 pub use protocol::{
     CloseCode, CloseReason, Message, MessageType, PayloadType, Role, WebSocketConfig,
 };
-pub use ws::WebSocket;
+pub use ws::{CloseState, WebSocket};
 
 use tokio::io::{AsyncRead, AsyncWrite};