fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-28 17:21:34 +00:00
Code Issues Projects Releases Wiki Activity

More concerns

Browse Source
This commit is contained in:
Pierre Krieger
2018-01-11 10:51:13 +01:00
parent 2783c5713e
commit b8829d7cb1
5 changed files with 59 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
libp2p-websocket/src/browser.rs
View File

@ -38,21 +38,21 @@ use tokio_io::{AsyncRead, AsyncWrite};
///
/// > **Note**: The `/wss` protocol isn't supported.
#[derive(Debug, Clone)]
pub struct WsConfig;
pub struct BrowserWsConfig;
impl WsConfig {
impl BrowserWsConfig {
/// Creates a new configuration object for websocket.
#[inline]
pub fn new() -> WsConfig {
WsConfig
pub fn new() -> BrowserWsConfig {
BrowserWsConfig
}
}
impl Transport for WsConfig {
type RawConn = WsConn;
impl Transport for BrowserWsConfig {
type RawConn = BrowserWsConn;
type Listener = Box<Stream<Item = (Self::ListenerUpgrade, Multiaddr), Error = IoError>>; // TODO: use `!`
type ListenerUpgrade = Box<Future<Item = Self::RawConn, Error = IoError>>; // TODO: use `!`
type Dial = FutureThen<oneshot::Receiver<Result<WsConn, IoError>>, Result<WsConn, IoError>, fn(Result<Result<WsConn, IoError>, oneshot::Canceled>) -> Result<WsConn, IoError>>;
type Dial = FutureThen<oneshot::Receiver<Result<BrowserWsConn, IoError>>, Result<BrowserWsConn, IoError>, fn(Result<Result<BrowserWsConn, IoError>, oneshot::Canceled>) -> Result<BrowserWsConn, IoError>>;
#[inline]
fn listen_on(self, a: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
@ -107,10 +107,10 @@ impl Transport for WsConfig {
}
};
// Create a `open` channel that will be used to communicate the `WsConn` that represents
// Create a `open` channel that will be used to communicate the `BrowserWsConn` that represents
// the open dialing websocket. Also create a `open_cb` callback that will be used for the
// `open` message of the websocket.
let (open_tx, open_rx) = oneshot::channel::<Result<WsConn, IoError>>();
let (open_tx, open_rx) = oneshot::channel::<Result<BrowserWsConn, IoError>>();
let open_tx = Arc::new(Mutex::new(Some(open_tx)));
let websocket_clone = websocket.clone();
let open_cb = {
@ -124,18 +124,18 @@ impl Transport for WsConfig {
// is not supposed to happen.
let message_rx = message_rx.take().expect("the websocket can only open once");
// Send a `WsConn` to the future that was returned by `dial`. Ignoring errors that
// Send a `BrowserWsConn` to the future that was returned by `dial`. Ignoring errors that
// would happen the future has been dropped by the user.
let _ = tx
.send(Ok(WsConn {
.send(Ok(BrowserWsConn {
websocket: websocket_clone.clone(),
incoming_data: RwStreamSink::new(message_rx.then(|result| {
// An `Err` happens here if `message_tx` has been dropped. However
// `message_tx` is grabbed by the websocket, which stays alive for as
// long as the `WsConn` is alive.
// long as the `BrowserWsConn` is alive.
match result {
Ok(r) => r,
Err(_) => unreachable!("the message channel outlives the WsConn")
Err(_) => unreachable!("the message channel outlives the BrowserWsConn")
}
})),
}));
@ -191,7 +191,7 @@ impl Transport for WsConfig {
}
}
pub struct WsConn {
pub struct BrowserWsConn {
websocket: Reference,
// Stream of messages that goes through a `RwStreamSink` in order to become a `AsyncRead`.
incoming_data: RwStreamSink<StreamThen<mpsc::UnboundedReceiver<Result<Vec<u8>, IoError>>,
@ -200,7 +200,7 @@ pub struct WsConn {
>>,
}
impl Drop for WsConn {
impl Drop for BrowserWsConn {
#[inline]
fn drop(&mut self) {
// TODO: apparently there's a memory leak related to callbacks?
@ -208,24 +208,24 @@ impl Drop for WsConn {
}
}
impl AsyncRead for WsConn {
impl AsyncRead for BrowserWsConn {
}
impl Read for WsConn {
impl Read for BrowserWsConn {
#[inline]
fn read(&mut self, buf: &mut [u8]) -> Result<usize, IoError> {
self.incoming_data.read(buf)
}
}
impl AsyncWrite for WsConn {
impl AsyncWrite for BrowserWsConn {
#[inline]
fn shutdown(&mut self) -> Poll<(), IoError> {
Ok(Async::Ready(()))
}
}
impl Write for WsConn {
impl Write for BrowserWsConn {
fn write(&mut self, buf: &[u8]) -> Result<usize, IoError> {
let typed_array = TypedArray::from(buf);

22
libp2p-websocket/src/desktop.rs
View File

@ -74,11 +74,9 @@ where
};
let (inner_listen, new_addr) = match self.transport.listen_on(inner_addr) {
Ok((listen, inner_new_addr)) => {
Ok((listen, mut new_addr)) => {
// Need to suffix `/ws` to the listening address.
let new_addr = inner_new_addr
.encapsulate("/ws")
.expect("the /ws suffix is always valid");
new_addr.append(AddrComponent::WS);
(listen, new_addr)
}
Err((transport, _)) => {
@ -91,11 +89,9 @@ where
}
};
let listen = inner_listen.map::<_, fn(_) -> _>(|(stream, client_addr)| {
let listen = inner_listen.map::<_, fn(_) -> _>(|(stream, mut client_addr)| {
// Need to suffix `/ws` to each client address.
let client_addr = client_addr
.encapsulate("/ws")
.expect("the /ws suffix is always valid");
client_addr.append(AddrComponent::WS);
// Upgrade the listener to websockets like the websockets library requires us to do.
let upgraded = stream.and_then(|stream| {
@ -118,8 +114,9 @@ where
OwnedMessage::Binary(data) => Ok(Some(data)),
OwnedMessage::Text(data) => Ok(Some(data.into_bytes())),
OwnedMessage::Close(_) => Ok(None),
// TODO: pings and pongs ; freaking hard
_ => unimplemented!()
// TODO: handle pings and pongs, which is freaking hard
// for now we close the socket when that happens
_ => Ok(None)
}
})
// TODO: is there a way to merge both lines into one?
@ -181,8 +178,9 @@ where
OwnedMessage::Binary(data) => Ok(data),
OwnedMessage::Text(data) => Ok(data.into_bytes()),
// TODO: pings and pongs and close messages need to be
// answered ; and this is really hard
_ => unimplemented!(),
// answered ; and this is really hard ; for now we produce
// an error when that happens
_ => Err(IoError::new(IoErrorKind::Other, "unimplemented")),
}
});
let read_write = RwStreamSink::new(framed_data);

13
libp2p-websocket/src/lib.rs
View File

@ -33,16 +33,16 @@
//!
//! # Emscripten
//!
//! On emscripten, you can create a `WsConfig` object with `WsConfig::new()`. It can then be used
//! as a transport.
//! On emscripten, you can create a `BrowserWsConfig` object with `BrowserWsConfig::new()`. It can
//! then be used as a transport.
//!
//! Listening on a websockets multiaddress isn't supported on emscripten. Dialing a multiaddress
//! which uses `ws` on top of TCP/IP will automatically use the `XMLHttpRequest` Javascript object.
//!
//! ```ignore
//! use libp2p_websocket::WsConfig;
//! use libp2p_websocket::BrowserWsConfig;
//!
//! let ws_config = WsConfig::new();
//! let ws_config = BrowserWsConfig::new();
//! // let _ = ws_config.dial("/ip4/40.41.42.43/tcp/12345/ws".parse().unwrap());
//! ```
//!
@ -51,7 +51,8 @@
//! On other operating systems, this library doesn't open any socket by itself. Instead it must be
//! plugged on top of another implementation of `Transport` such as TCP/IP.
//!
//! This underlying transport must be passed to the `WsConfig::new()` function.
//! This underlying transport must be put inside a `WsConfig` object through the
//! `WsConfig::new()` function.
//!
//! ```
//! extern crate libp2p_swarm;
@ -91,6 +92,6 @@ mod desktop;
mod browser;
#[cfg(target_os = "emscripten")]
pub use self::browser::WsConfig;
pub use self::browser::{BrowserWsConfig, BrowserWsConn};
#[cfg(not(target_os = "emscripten"))]
pub use self::desktop::WsConfig;