mirror of
https://github.com/fluencelabs/rust-libp2p
synced 2025-06-24 15:21:33 +00:00
Jef
Pierre Krieger
350
websocket/src/browser.rs
Normal file
350
websocket/src/browser.rs
Normal file
@ -0,0 +1,350 @@
|
||||
// Copyright 2017 Parity Technologies (UK) Ltd.
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a
|
||||
// copy of this software and associated documentation files (the "Software"),
|
||||
// to deal in the Software without restriction, including without limitation
|
||||
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
||||
// and/or sell copies of the Software, and to permit persons to whom the
|
||||
// Software is furnished to do so, subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
|
||||
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
// DEALINGS IN THE SOFTWARE.
|
||||
|
||||
use futures::{Async, Future, Poll, Stream};
|
||||
use futures::stream::Then as StreamThen;
|
||||
use futures::sync::{mpsc, oneshot};
|
||||
use multiaddr::{AddrComponent, Multiaddr};
|
||||
use rw_stream_sink::RwStreamSink;
|
||||
use std::io::{Error as IoError, ErrorKind as IoErrorKind};
|
||||
use std::io::{Read, Write};
|
||||
use std::iter;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use stdweb::{self, Reference};
|
||||
use stdweb::web::TypedArray;
|
||||
use swarm::Transport;
|
||||
use tokio_io::{AsyncRead, AsyncWrite};
|
||||
|
||||
/// Represents the configuration for a websocket transport capability for libp2p.
|
||||
///
|
||||
/// This implementation of `Transport` accepts any address that looks like
|
||||
/// `/ip4/.../tcp/.../ws` or `/ip6/.../tcp/.../ws`, and connect to the corresponding IP and port.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct BrowserWsConfig;
|
||||
|
||||
impl BrowserWsConfig {
|
||||
/// Creates a new configuration object for websocket.
|
||||
#[inline]
|
||||
pub fn new() -> BrowserWsConfig {
|
||||
BrowserWsConfig
|
||||
}
|
||||
}
|
||||
|
||||
impl Transport for BrowserWsConfig {
|
||||
type RawConn = BrowserWsConn;
|
||||
type Listener = Box<Stream<Item = Self::ListenerUpgrade, Error = IoError>>; // TODO: use `!`
|
||||
type ListenerUpgrade = Box<Future<Item = (Self::RawConn, Multiaddr), Error = IoError>>; // TODO: use `!`
|
||||
type Dial = Box<Future<Item = (Self::RawConn, Multiaddr), Error = IoError>>;
|
||||
|
||||
#[inline]
|
||||
fn listen_on(self, a: Multiaddr) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
|
||||
// Listening is never supported.
|
||||
Err((self, a))
|
||||
}
|
||||
|
||||
fn dial(self, original_addr: Multiaddr) -> Result<Self::Dial, (Self, Multiaddr)> {
|
||||
// Making sure we are initialized before we dial. Initialization is protected by a simple
|
||||
// boolean static variable, so it's not a problem to call it multiple times and the cost
|
||||
// is negligible.
|
||||
stdweb::initialize();
|
||||
|
||||
// Tries to interpret the multiaddr, and returns a corresponding `ws://x.x.x.x/` URL (as
|
||||
// a string) on success.
|
||||
let inner_addr = match multiaddr_to_target(&original_addr) {
|
||||
Ok(a) => a,
|
||||
Err(_) => return Err((self, original_addr)),
|
||||
};
|
||||
|
||||
debug!(target: "libp2p-websocket", "Dialing {}", original_addr);
|
||||
|
||||
// Create the JS `WebSocket` object.
|
||||
let websocket = {
|
||||
let val = js! {
|
||||
try {
|
||||
return new WebSocket(@{inner_addr});
|
||||
} catch(e) {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
match val.into_reference() {
|
||||
Some(ws) => ws,
|
||||
None => return Err((self, original_addr)), // `false` was returned by `js!`
|
||||
}
|
||||
};
|
||||
|
||||
// Create a `message` channel that will be used for both bytes messages and errors, and a
|
||||
// `message_cb` used for the `message` event on the WebSocket.
|
||||
// `message_tx` is grabbed by `message_cb` and `close_cb`, and `message_rx` is grabbed
|
||||
// by `open_cb`.
|
||||
let (message_tx, message_rx) = mpsc::unbounded::<Result<Vec<u8>, IoError>>();
|
||||
let message_tx = Arc::new(message_tx);
|
||||
let mut message_rx = Some(message_rx);
|
||||
let message_cb = {
|
||||
let message_tx = message_tx.clone();
|
||||
move |message_data: Reference| {
|
||||
if let Some(buffer) = message_data.downcast::<TypedArray<u8>>() {
|
||||
let _ = message_tx.unbounded_send(Ok(buffer.to_vec()));
|
||||
} else {
|
||||
let _ = message_tx.unbounded_send(Err(IoError::new(
|
||||
IoErrorKind::InvalidData,
|
||||
"received ws message of unknown type",
|
||||
)));
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// 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<BrowserWsConn, IoError>>();
|
||||
let open_tx = Arc::new(Mutex::new(Some(open_tx)));
|
||||
let websocket_clone = websocket.clone();
|
||||
let open_cb = {
|
||||
let open_tx = open_tx.clone();
|
||||
move || {
|
||||
// Note that `open_tx` can be empty (and a panic happens) if the `open` event
|
||||
// is triggered twice, or is triggered after the `close` event. We never reuse the
|
||||
// same websocket twice, so this is not supposed to happen.
|
||||
let tx = open_tx
|
||||
.lock()
|
||||
.unwrap()
|
||||
.take()
|
||||
.expect("the websocket can only open once");
|
||||
// `message_rx` can be empty if the `open` event is triggered twice, which again
|
||||
// is not supposed to happen.
|
||||
let message_rx = message_rx.take().expect("the websocket can only open once");
|
||||
|
||||
// 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(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 `BrowserWsConn` is alive.
|
||||
match result {
|
||||
Ok(r) => r,
|
||||
Err(_) => {
|
||||
unreachable!("the message channel outlives the BrowserWsConn")
|
||||
}
|
||||
}
|
||||
})),
|
||||
}));
|
||||
}
|
||||
};
|
||||
|
||||
// Used for the `close` message of the websocket.
|
||||
// The websocket can be closed either before or after being opened, so we send an error
|
||||
// to both the `open` and `message` channels if that happens.
|
||||
let close_cb = move || {
|
||||
if let Some(tx) = open_tx.lock().unwrap().take() {
|
||||
let _ = tx.send(Err(IoError::new(
|
||||
IoErrorKind::ConnectionRefused,
|
||||
"close event on the websocket",
|
||||
)));
|
||||
}
|
||||
|
||||
let _ = message_tx.unbounded_send(Err(IoError::new(
|
||||
IoErrorKind::ConnectionRefused,
|
||||
"close event on the websocket",
|
||||
)));
|
||||
};
|
||||
|
||||
js! {
|
||||
var socket = @{websocket};
|
||||
var open_cb = @{open_cb};
|
||||
var message_cb = @{message_cb};
|
||||
var close_cb = @{close_cb};
|
||||
socket.addEventListener("open", function(event) {
|
||||
open_cb();
|
||||
});
|
||||
socket.addEventListener("message", function(event) {
|
||||
var reader = new FileReader();
|
||||
reader.addEventListener("loadend", function() {
|
||||
var typed = new Uint8Array(reader.result);
|
||||
message_cb(typed);
|
||||
});
|
||||
reader.readAsArrayBuffer(event.data);
|
||||
});
|
||||
socket.addEventListener("close", function(event) {
|
||||
close_cb();
|
||||
});
|
||||
};
|
||||
|
||||
Ok(Box::new(open_rx.then(|result| {
|
||||
match result {
|
||||
Ok(Ok(r)) => Ok((r, original_addr)),
|
||||
Ok(Err(e)) => Err(e),
|
||||
// `Err` would happen here if `open_tx` is destroyed. `open_tx` is captured by
|
||||
// the `WebSocket`, and the `WebSocket` is captured by `open_cb`, which is itself
|
||||
// captured by the `WebSocket`. Due to this cyclic dependency, `open_tx` should
|
||||
// never be destroyed.
|
||||
// TODO: how do we break this cyclic dependency? difficult question
|
||||
Err(_) => unreachable!("the sending side will only close when we drop the future"),
|
||||
}
|
||||
})) as Box<_>)
|
||||
}
|
||||
|
||||
fn nat_traversal(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
|
||||
let mut server_protocols = server.iter();
|
||||
let server_proto0 = server_protocols.next()?;
|
||||
let server_proto1 = server_protocols.next()?;
|
||||
let server_proto2 = server_protocols.next()?;
|
||||
if server_protocols.next().is_some() {
|
||||
return None;
|
||||
}
|
||||
|
||||
let mut observed_protocols = observed.iter();
|
||||
let obs_proto0 = observed_protocols.next()?;
|
||||
let obs_proto1 = observed_protocols.next()?;
|
||||
let obs_proto2 = observed_protocols.next()?;
|
||||
if observed_protocols.next().is_some() {
|
||||
return None;
|
||||
}
|
||||
|
||||
// Check that `server` is a valid TCP/IP address.
|
||||
match (&server_proto0, &server_proto1, &server_proto2) {
|
||||
(&AddrComponent::IP4(_), &AddrComponent::TCP(_), &AddrComponent::WS)
|
||||
| (&AddrComponent::IP6(_), &AddrComponent::TCP(_), &AddrComponent::WS)
|
||||
| (&AddrComponent::IP4(_), &AddrComponent::TCP(_), &AddrComponent::WSS)
|
||||
| (&AddrComponent::IP6(_), &AddrComponent::TCP(_), &AddrComponent::WSS) => {}
|
||||
_ => return None,
|
||||
}
|
||||
|
||||
// Check that `observed` is a valid TCP/IP address.
|
||||
match (&obs_proto0, &obs_proto1, &obs_proto2) {
|
||||
(&AddrComponent::IP4(_), &AddrComponent::TCP(_), &AddrComponent::WS)
|
||||
| (&AddrComponent::IP6(_), &AddrComponent::TCP(_), &AddrComponent::WS)
|
||||
| (&AddrComponent::IP4(_), &AddrComponent::TCP(_), &AddrComponent::WSS)
|
||||
| (&AddrComponent::IP6(_), &AddrComponent::TCP(_), &AddrComponent::WSS) => {}
|
||||
_ => return None,
|
||||
}
|
||||
|
||||
// Note that it will still work if the server uses WSS while the client uses WS,
|
||||
// or vice-versa.
|
||||
|
||||
let result = iter::once(obs_proto0)
|
||||
.chain(iter::once(server_proto1))
|
||||
.chain(iter::once(server_proto2))
|
||||
.collect();
|
||||
|
||||
Some(result)
|
||||
}
|
||||
}
|
||||
|
||||
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>>,
|
||||
fn(Result<Result<Vec<u8>, IoError>, ()>) -> Result<Vec<u8>, IoError>,
|
||||
Result<Vec<u8>, IoError>,
|
||||
>,
|
||||
>,
|
||||
}
|
||||
|
||||
impl Drop for BrowserWsConn {
|
||||
#[inline]
|
||||
fn drop(&mut self) {
|
||||
// TODO: apparently there's a memory leak related to callbacks?
|
||||
js! { @{&self.websocket}.close(); }
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncRead for BrowserWsConn {}
|
||||
|
||||
impl Read for BrowserWsConn {
|
||||
#[inline]
|
||||
fn read(&mut self, buf: &mut [u8]) -> Result<usize, IoError> {
|
||||
self.incoming_data.read(buf)
|
||||
}
|
||||
}
|
||||
|
||||
impl AsyncWrite for BrowserWsConn {
|
||||
#[inline]
|
||||
fn shutdown(&mut self) -> Poll<(), IoError> {
|
||||
Ok(Async::Ready(()))
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for BrowserWsConn {
|
||||
fn write(&mut self, buf: &[u8]) -> Result<usize, IoError> {
|
||||
let typed_array = TypedArray::from(buf);
|
||||
|
||||
// `send` can throw if the websocket isn't open (which can happen if it was closed by the
|
||||
// remote).
|
||||
let returned = js! {
|
||||
try {
|
||||
@{&self.websocket}.send(@{typed_array.buffer()});
|
||||
return true;
|
||||
} catch(e) {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
match returned {
|
||||
stdweb::Value::Bool(true) => Ok(buf.len()),
|
||||
stdweb::Value::Bool(false) => Err(IoError::new(
|
||||
IoErrorKind::BrokenPipe,
|
||||
"websocket has been closed by the remote",
|
||||
)),
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn flush(&mut self) -> Result<(), IoError> {
|
||||
// Everything is always considered flushed.
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// Tries to interpret the `Multiaddr` as a `/ipN/.../tcp/.../ws` multiaddress, and if so returns
|
||||
// the corresponding `ws://.../` URL.
|
||||
fn multiaddr_to_target(addr: &Multiaddr) -> Result<String, ()> {
|
||||
let protocols: Vec<_> = addr.iter().collect();
|
||||
|
||||
if protocols.len() != 3 {
|
||||
return Err(());
|
||||
}
|
||||
|
||||
match (&protocols[0], &protocols[1], &protocols[2]) {
|
||||
(&AddrComponent::IP4(ref ip), &AddrComponent::TCP(port), &AddrComponent::WS) => {
|
||||
Ok(format!("ws://{}:{}/", ip, port))
|
||||
}
|
||||
(&AddrComponent::IP6(ref ip), &AddrComponent::TCP(port), &AddrComponent::WS) => {
|
||||
Ok(format!("ws://[{}]:{}/", ip, port))
|
||||
}
|
||||
(&AddrComponent::IP4(ref ip), &AddrComponent::TCP(port), &AddrComponent::WSS) => {
|
||||
Ok(format!("wss://{}:{}/", ip, port))
|
||||
}
|
||||
(&AddrComponent::IP6(ref ip), &AddrComponent::TCP(port), &AddrComponent::WSS) => {
|
||||
Ok(format!("wss://[{}]:{}/", ip, port))
|
||||
}
|
||||
_ => Err(()),
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: write tests (tests are very difficult to write with emscripten)
|
||||
// - remote refuses connection
|
||||
// - remote closes connection before we receive
|
||||
// - remote closes connection before we send
|
||||
// - remote sends text data instead of binary
|
||||
348
websocket/src/desktop.rs
Normal file
348
websocket/src/desktop.rs
Normal file
@ -0,0 +1,348 @@
|
||||
// Copyright 2017 Parity Technologies (UK) Ltd.
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a
|
||||
// copy of this software and associated documentation files (the "Software"),
|
||||
// to deal in the Software without restriction, including without limitation
|
||||
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
||||
// and/or sell copies of the Software, and to permit persons to whom the
|
||||
// Software is furnished to do so, subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
|
||||
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
// DEALINGS IN THE SOFTWARE.
|
||||
|
||||
use futures::{stream, Future, IntoFuture, Sink, Stream};
|
||||
use multiaddr::{AddrComponent, Multiaddr};
|
||||
use rw_stream_sink::RwStreamSink;
|
||||
use std::io::{Error as IoError, ErrorKind as IoErrorKind};
|
||||
use swarm::Transport;
|
||||
use websocket::client::builder::ClientBuilder;
|
||||
use websocket::message::OwnedMessage;
|
||||
use websocket::server::upgrade::async::IntoWs;
|
||||
use websocket::stream::async::Stream as AsyncStream;
|
||||
|
||||
/// Represents the configuration for a websocket transport capability for libp2p. Must be put on
|
||||
/// top of another `Transport`.
|
||||
///
|
||||
/// This implementation of `Transport` accepts any address that ends with `/ws` or `/wss`, and will
|
||||
/// try to pass the underlying multiaddress to the underlying `Transport`.
|
||||
///
|
||||
/// > **Note**: `/wss` is only supported for dialing and not listening.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WsConfig<T> {
|
||||
transport: T,
|
||||
}
|
||||
|
||||
impl<T> WsConfig<T> {
|
||||
/// Creates a new configuration object for websocket.
|
||||
///
|
||||
/// The websockets will run on top of the `Transport` you pass as parameter.
|
||||
#[inline]
|
||||
pub fn new(inner: T) -> WsConfig<T> {
|
||||
WsConfig { transport: inner }
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Transport for WsConfig<T>
|
||||
where
|
||||
// TODO: this 'static is pretty arbitrary and is necessary because of the websocket library
|
||||
T: Transport + 'static,
|
||||
// TODO: this Send is pretty arbitrary and is necessary because of the websocket library
|
||||
T::RawConn: Send,
|
||||
{
|
||||
type RawConn = Box<AsyncStream>;
|
||||
type Listener =
|
||||
stream::Map<T::Listener, fn(<T as Transport>::ListenerUpgrade) -> Self::ListenerUpgrade>;
|
||||
type ListenerUpgrade = Box<Future<Item = (Self::RawConn, Multiaddr), Error = IoError>>;
|
||||
type Dial = Box<Future<Item = (Self::RawConn, Multiaddr), Error = IoError>>;
|
||||
|
||||
fn listen_on(
|
||||
self,
|
||||
original_addr: Multiaddr,
|
||||
) -> Result<(Self::Listener, Multiaddr), (Self, Multiaddr)> {
|
||||
let mut inner_addr = original_addr.clone();
|
||||
match inner_addr.pop() {
|
||||
Some(AddrComponent::WS) => {}
|
||||
_ => return Err((self, original_addr)),
|
||||
};
|
||||
|
||||
let (inner_listen, new_addr) = match self.transport.listen_on(inner_addr) {
|
||||
Ok((listen, mut new_addr)) => {
|
||||
// Need to suffix `/ws` to the listening address.
|
||||
new_addr.append(AddrComponent::WS);
|
||||
(listen, new_addr)
|
||||
}
|
||||
Err((transport, _)) => {
|
||||
return Err((
|
||||
WsConfig {
|
||||
transport: transport,
|
||||
},
|
||||
original_addr,
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
debug!(target: "libp2p-websocket", "Listening on {}", new_addr);
|
||||
|
||||
let listen = inner_listen.map::<_, fn(_) -> _>(|stream| {
|
||||
// Upgrade the listener to websockets like the websockets library requires us to do.
|
||||
let upgraded = stream.and_then(|(stream, mut client_addr)| {
|
||||
// Need to suffix `/ws` to each client address.
|
||||
client_addr.append(AddrComponent::WS);
|
||||
debug!(target: "libp2p-websocket", "Incoming connection from {}", client_addr);
|
||||
|
||||
stream
|
||||
.into_ws()
|
||||
.map_err(|e| IoError::new(IoErrorKind::Other, e.3))
|
||||
.and_then(|stream| {
|
||||
// Accept the next incoming connection.
|
||||
stream
|
||||
.accept()
|
||||
.map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.map(|(client, _http_headers)| {
|
||||
debug!(target: "libp2p-websocket", "Upgraded incoming connection \
|
||||
to websockets");
|
||||
|
||||
// Plug our own API on top of the `websockets` API.
|
||||
let framed_data = client
|
||||
.map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.sink_map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.with(|data| Ok(OwnedMessage::Binary(data)))
|
||||
.and_then(|recv| {
|
||||
match recv {
|
||||
OwnedMessage::Binary(data) => Ok(Some(data)),
|
||||
OwnedMessage::Text(data) => Ok(Some(data.into_bytes())),
|
||||
OwnedMessage::Close(_) => Ok(None),
|
||||
// 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?
|
||||
.take_while(|v| Ok(v.is_some()))
|
||||
.map(|v| v.expect("we only take while this is Some"));
|
||||
|
||||
let read_write = RwStreamSink::new(framed_data);
|
||||
Box::new(read_write) as Box<AsyncStream>
|
||||
})
|
||||
})
|
||||
.map(|s| Box::new(Ok(s).into_future()) as Box<Future<Item = _, Error = _>>)
|
||||
.into_future()
|
||||
.flatten()
|
||||
.map(move |v| (v, client_addr))
|
||||
});
|
||||
|
||||
Box::new(upgraded) as Box<Future<Item = _, Error = _>>
|
||||
});
|
||||
|
||||
Ok((listen, new_addr))
|
||||
}
|
||||
|
||||
fn dial(self, original_addr: Multiaddr) -> Result<Self::Dial, (Self, Multiaddr)> {
|
||||
let mut inner_addr = original_addr.clone();
|
||||
let is_wss = match inner_addr.pop() {
|
||||
Some(AddrComponent::WS) => false,
|
||||
Some(AddrComponent::WSS) => true,
|
||||
_ => return Err((self, original_addr)),
|
||||
};
|
||||
|
||||
debug!(target: "libp2p-websocket", "Dialing {} through inner transport", inner_addr);
|
||||
|
||||
let inner_dial = match self.transport.dial(inner_addr) {
|
||||
Ok(d) => d,
|
||||
Err((transport, _)) => {
|
||||
return Err((
|
||||
WsConfig {
|
||||
transport: transport,
|
||||
},
|
||||
original_addr,
|
||||
));
|
||||
}
|
||||
};
|
||||
|
||||
let dial = inner_dial
|
||||
.into_future()
|
||||
.and_then(move |(connec, client_addr)| {
|
||||
// We pass a dummy address to `ClientBuilder` because it is never used anywhere
|
||||
// in the negotiation anyway, and we use `async_connect_on` to pass a stream.
|
||||
ClientBuilder::new(if is_wss {
|
||||
"wss://127.0.0.1"
|
||||
} else {
|
||||
"ws://127.0.0.1"
|
||||
}).expect("hard-coded ws address is always valid")
|
||||
.async_connect_on(connec)
|
||||
.map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.map(|(client, _)| {
|
||||
debug!(target: "libp2p-websocket", "Upgraded outgoing connection to \
|
||||
websockets");
|
||||
|
||||
// Plug our own API on top of the API of the websockets library.
|
||||
let framed_data = client
|
||||
.map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.sink_map_err(|err| IoError::new(IoErrorKind::Other, err))
|
||||
.with(|data| Ok(OwnedMessage::Binary(data)))
|
||||
.and_then(|recv| {
|
||||
match recv {
|
||||
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 ; for now we produce
|
||||
// an error when that happens
|
||||
_ => Err(IoError::new(IoErrorKind::Other, "unimplemented")),
|
||||
}
|
||||
});
|
||||
let read_write = RwStreamSink::new(framed_data);
|
||||
Box::new(read_write) as Box<AsyncStream>
|
||||
})
|
||||
.map(|c| (c, client_addr))
|
||||
});
|
||||
|
||||
Ok(Box::new(dial) as Box<_>)
|
||||
}
|
||||
|
||||
fn nat_traversal(&self, server: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
|
||||
let mut server = server.clone();
|
||||
let last_proto = match server.pop() {
|
||||
Some(v @ AddrComponent::WS) | Some(v @ AddrComponent::WSS) => v,
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
let mut observed = observed.clone();
|
||||
match observed.pop() {
|
||||
Some(AddrComponent::WS) => false,
|
||||
Some(AddrComponent::WSS) => true,
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
self.transport
|
||||
.nat_traversal(&server, &observed)
|
||||
.map(move |mut result| {
|
||||
result.append(last_proto);
|
||||
result
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
extern crate libp2p_tcp_transport as tcp;
|
||||
extern crate tokio_core;
|
||||
use self::tokio_core::reactor::Core;
|
||||
use WsConfig;
|
||||
use futures::{Future, Stream};
|
||||
use multiaddr::Multiaddr;
|
||||
use swarm::Transport;
|
||||
|
||||
#[test]
|
||||
fn dialer_connects_to_listener_ipv4() {
|
||||
let mut core = Core::new().unwrap();
|
||||
let ws_config = WsConfig::new(tcp::TcpConfig::new(core.handle()));
|
||||
|
||||
let (listener, addr) = ws_config
|
||||
.clone()
|
||||
.listen_on("/ip4/0.0.0.0/tcp/0/ws".parse().unwrap())
|
||||
.unwrap();
|
||||
assert!(addr.to_string().ends_with("/ws"));
|
||||
assert!(!addr.to_string().ends_with("/0/ws"));
|
||||
let listener = listener
|
||||
.into_future()
|
||||
.map_err(|(e, _)| e)
|
||||
.and_then(|(c, _)| c.unwrap().map(|v| v.0));
|
||||
let dialer = ws_config.clone().dial(addr).unwrap().map(|v| v.0);
|
||||
|
||||
let future = listener
|
||||
.select(dialer)
|
||||
.map_err(|(e, _)| e)
|
||||
.and_then(|(_, n)| n);
|
||||
core.run(future).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dialer_connects_to_listener_ipv6() {
|
||||
let mut core = Core::new().unwrap();
|
||||
let ws_config = WsConfig::new(tcp::TcpConfig::new(core.handle()));
|
||||
|
||||
let (listener, addr) = ws_config
|
||||
.clone()
|
||||
.listen_on("/ip6/::1/tcp/0/ws".parse().unwrap())
|
||||
.unwrap();
|
||||
assert!(addr.to_string().ends_with("/ws"));
|
||||
assert!(!addr.to_string().ends_with("/0/ws"));
|
||||
let listener = listener
|
||||
.into_future()
|
||||
.map_err(|(e, _)| e)
|
||||
.and_then(|(c, _)| c.unwrap().map(|v| v.0));
|
||||
let dialer = ws_config.clone().dial(addr).unwrap().map(|v| v.0);
|
||||
|
||||
let future = listener
|
||||
.select(dialer)
|
||||
.map_err(|(e, _)| e)
|
||||
.and_then(|(_, n)| n);
|
||||
core.run(future).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nat_traversal() {
|
||||
let core = Core::new().unwrap();
|
||||
let ws_config = WsConfig::new(tcp::TcpConfig::new(core.handle()));
|
||||
|
||||
{
|
||||
let server = "/ip4/127.0.0.1/tcp/10000/ws".parse::<Multiaddr>().unwrap();
|
||||
let observed = "/ip4/80.81.82.83/tcp/25000/ws"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
ws_config.nat_traversal(&server, &observed).unwrap(),
|
||||
"/ip4/80.81.82.83/tcp/10000/ws"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
{
|
||||
let server = "/ip4/127.0.0.1/tcp/10000/wss".parse::<Multiaddr>().unwrap();
|
||||
let observed = "/ip4/80.81.82.83/tcp/25000/wss"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
ws_config.nat_traversal(&server, &observed).unwrap(),
|
||||
"/ip4/80.81.82.83/tcp/10000/wss"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
{
|
||||
let server = "/ip4/127.0.0.1/tcp/10000/ws".parse::<Multiaddr>().unwrap();
|
||||
let observed = "/ip4/80.81.82.83/tcp/25000/wss"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
ws_config.nat_traversal(&server, &observed).unwrap(),
|
||||
"/ip4/80.81.82.83/tcp/10000/ws"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
{
|
||||
let server = "/ip4/127.0.0.1/tcp/10000/wss".parse::<Multiaddr>().unwrap();
|
||||
let observed = "/ip4/80.81.82.83/tcp/25000/ws"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
ws_config.nat_traversal(&server, &observed).unwrap(),
|
||||
"/ip4/80.81.82.83/tcp/10000/wss"
|
||||
.parse::<Multiaddr>()
|
||||
.unwrap()
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
99
websocket/src/lib.rs
Normal file
99
websocket/src/lib.rs
Normal file
@ -0,0 +1,99 @@
|
||||
// Copyright 2017 Parity Technologies (UK) Ltd.
|
||||
//
|
||||
// Permission is hereby granted, free of charge, to any person obtaining a
|
||||
// copy of this software and associated documentation files (the "Software"),
|
||||
// to deal in the Software without restriction, including without limitation
|
||||
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
|
||||
// and/or sell copies of the Software, and to permit persons to whom the
|
||||
// Software is furnished to do so, subject to the following conditions:
|
||||
//
|
||||
// The above copyright notice and this permission notice shall be included in
|
||||
// all copies or substantial portions of the Software.
|
||||
//
|
||||
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
|
||||
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
|
||||
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
|
||||
// DEALINGS IN THE SOFTWARE.
|
||||
|
||||
#![recursion_limit = "512"]
|
||||
|
||||
// TODO: use this once stable ; for now we just copy-paste the content of the README.md
|
||||
//#![doc(include = "../README.md")]
|
||||
|
||||
//! Implementation of the libp2p `Transport` trait for Websockets.
|
||||
//!
|
||||
//! See the documentation of `swarm` and of libp2p in general to learn how to use the `Transport`
|
||||
//! trait.
|
||||
//!
|
||||
//! This library is used in a different way depending on whether you are compiling for emscripten
|
||||
//! or for a different operating system.
|
||||
//!
|
||||
//! # Emscripten
|
||||
//!
|
||||
//! 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::BrowserWsConfig;
|
||||
//!
|
||||
//! let ws_config = BrowserWsConfig::new();
|
||||
//! // let _ = ws_config.dial("/ip4/40.41.42.43/tcp/12345/ws".parse().unwrap());
|
||||
//! ```
|
||||
//!
|
||||
//! # Other operating systems
|
||||
//!
|
||||
//! 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 put inside a `WsConfig` object through the
|
||||
//! `WsConfig::new()` function.
|
||||
//!
|
||||
//! ```
|
||||
//! extern crate libp2p_swarm;
|
||||
//! extern crate libp2p_tcp_transport;
|
||||
//! extern crate libp2p_websocket;
|
||||
//! extern crate tokio_core;
|
||||
//!
|
||||
//! use libp2p_swarm::{Multiaddr, Transport};
|
||||
//! use libp2p_tcp_transport::TcpConfig;
|
||||
//! use libp2p_websocket::WsConfig;
|
||||
//! use tokio_core::reactor::Core;
|
||||
//!
|
||||
//! # fn main() {
|
||||
//! let core = Core::new().unwrap();
|
||||
//! let ws_config = WsConfig::new(TcpConfig::new(core.handle()));
|
||||
//! # return;
|
||||
//! let _ = ws_config.dial("/ip4/40.41.42.43/tcp/12345/ws".parse().unwrap());
|
||||
//! # }
|
||||
//! ```
|
||||
//!
|
||||
|
||||
extern crate futures;
|
||||
extern crate libp2p_swarm as swarm;
|
||||
#[macro_use]
|
||||
extern crate log;
|
||||
extern crate multiaddr;
|
||||
extern crate rw_stream_sink;
|
||||
extern crate tokio_io;
|
||||
|
||||
#[cfg(target_os = "emscripten")]
|
||||
#[macro_use]
|
||||
extern crate stdweb;
|
||||
#[cfg(not(target_os = "emscripten"))]
|
||||
extern crate websocket;
|
||||
|
||||
#[cfg(not(target_os = "emscripten"))]
|
||||
mod desktop;
|
||||
#[cfg(target_os = "emscripten")]
|
||||
mod browser;
|
||||
|
||||
#[cfg(target_os = "emscripten")]
|
||||
pub use self::browser::{BrowserWsConfig, BrowserWsConn};
|
||||
#[cfg(not(target_os = "emscripten"))]
|
||||
pub use self::desktop::WsConfig;
|
||||
Reference in New Issue
Block a user