Skip to content

Commit

Permalink
Merge pull request #15 from halzy/halzy/7-docs
Browse files Browse the repository at this point in the history
docs: Updating docs from feedback.
  • Loading branch information
halzy authored Mar 1, 2020
2 parents e344fbf + bd1a65a commit 16bad0e
Show file tree
Hide file tree
Showing 2 changed files with 17 additions and 14 deletions.
2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "stream_multiplexer"
version = "0.5.0"
version = "0.5.1"
authors = ["Benjamin Halsted <[email protected]>"]
edition = "2018"
license = "MIT OR Apache-2.0"
Expand Down
29 changes: 16 additions & 13 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -22,17 +22,19 @@ Channels have their own backpressure that does not affect other channels.
Incoming streams are by default set to channel 0 and can be moved to other channels via `ControlMessage`s.
```rust
# use std::error::Error;
use bytes::Bytes;
use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::mpsc;
use stream_multiplexer::{Multiplexer, HalvesStream, ControlMessage, IncomingPacket, OutgoingPacket};
use futures::stream::StreamExt;
# fn main() -> Result<(), Box<dyn Error>> {
# tokio::runtime::Builder::new().basic_scheduler().enable_all().build().unwrap().block_on(async move {
// 3 channels of incoming streams, 0 is the channel that new streams join.
// Backpressure is per channel. Streams can be moved between channels by
// sending an OutgoingPackt::ChangeChannel message.
// sending an OutgoingPacket::ChangeChannel message.
let (channel0_tx, mut channel0_rx) = mpsc::channel(32);
let (channel1_tx, mut channel1_rx) = mpsc::channel(32);
let (channel2_tx, mut channel2_rx) = mpsc::channel(32);
Expand All @@ -50,9 +52,9 @@ let multiplexer = Multiplexer::new(
);
// Bind to a random port on localhost
let socket = TcpListener::bind("127.0.0.1:0").await.unwrap();
let socket = TcpListener::bind("127.0.0.1:0").await?;
let local_addr = socket.local_addr().unwrap();
let local_addr = socket.local_addr()?;
// Use the HalvesStream utility struct to map the stream of new sockets.
// It will use LengthDelimitedCodec with 2 bytes as the packet size.
Expand All @@ -63,11 +65,11 @@ let (control_write, control_read) = mpsc::unbounded_channel();
let mp_joinhandle = tokio::task::spawn(multiplexer.run(halves, control_read));
// Make a test connection:
let mut client = tokio::net::TcpStream::connect(local_addr).await.unwrap();
let mut client = tokio::net::TcpStream::connect(local_addr).await?;
// Send 'a message'
let mut data = Bytes::from("\x00\x09a message");
client.write_buf(&mut data).await.unwrap();
client.write_buf(&mut data).await?;
client.flush();
// Receive 'a message' on channel 0
Expand All @@ -84,12 +86,11 @@ assert_eq!(
// Move the client to channel 1
outgoing_tx
.send(OutgoingPacket::ChangeChannel(vec![incoming_packet.id()], 1))
.await
.unwrap();
.await?;
// Send 'a message' again, on channel 1 this time.
let mut data = Bytes::from("\x00\x09a message");
client.write_buf(&mut data).await.unwrap();
client.write_buf(&mut data).await?;
client.flush();
// Receive 'a message' on channel 1
Expand All @@ -106,12 +107,11 @@ assert_eq!(
// Move the client to channel 2
outgoing_tx
.send(OutgoingPacket::ChangeChannel(vec![incoming_packet.id()], 2))
.await
.unwrap();
.await?;
// Send 'a message' again, on channel 2 this time.
let mut data = Bytes::from("\x00\x09a message");
client.write_buf(&mut data).await.unwrap();
client.write_buf(&mut data).await?;
client.flush();
// Receive 'a message' on channel 2
Expand All @@ -125,11 +125,14 @@ assert_eq!(
&Bytes::from("a message")
);
// Tell multiplexer te shut down
control_write.send(ControlMessage::Shutdown).unwrap();
// Tell multiplexer to shut down
control_write.send(ControlMessage::Shutdown)?;
mp_joinhandle.await.unwrap();
# Ok::<_, Box<dyn Error>>(())
# });
# Ok(())
# }
```
*/
mod error;
Expand Down

0 comments on commit 16bad0e

Please sign in to comment.