mirrors/actix-web
1
0
Fork 0
mirror of https://github.com/actix/actix-web.git synced 2024-11-26 11:31:09 +00:00
Code Issues Projects Releases Wiki Activity

add websocket transport and test

Browse source
This commit is contained in:
Nikolay Kim 2018-10-05 14:30:40 -07:00
parent 5c0a2066cc
commit 7e135b798b
9 changed files with 165 additions and 781 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

283
src/framed/framed.rs
View file

@ -1,283 +0,0 @@
#![allow(deprecated)]
use std::fmt;
use std::io::{self, Read, Write};
use bytes::BytesMut;
use futures::{Poll, Sink, StartSend, Stream};
use tokio_codec::{Decoder, Encoder};
use tokio_io::{AsyncRead, AsyncWrite};
use super::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2};
use super::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2};
/// A unified `Stream` and `Sink` interface to an underlying I/O object, using
/// the `Encoder` and `Decoder` traits to encode and decode frames.
///
/// You can create a `Framed` instance by using the `AsyncRead::framed` adapter.
pub struct Framed<T, U> {
inner: FramedRead2<FramedWrite2<Fuse<T, U>>>,
}
pub struct Fuse<T, U>(pub T, pub U);
impl<T, U> Framed<T, U>
where
T: AsyncRead + AsyncWrite,
U: Decoder + Encoder,
{
/// Provides a `Stream` and `Sink` interface for reading and writing to this
/// `Io` object, using `Decode` and `Encode` to read and write the raw data.
///
/// Raw I/O objects work with byte sequences, but higher-level code usually
/// wants to batch these into meaningful chunks, called "frames". This
/// method layers framing on top of an I/O object, by using the `Codec`
/// traits to handle encoding and decoding of messages frames. Note that
/// the incoming and outgoing frame types may be distinct.
///
/// This function returns a *single* object that is both `Stream` and
/// `Sink`; grouping this into a single object is often useful for layering
/// things like gzip or TLS, which require both read and write access to the
/// underlying object.
///
/// If you want to work more directly with the streams and sink, consider
/// calling `split` on the `Framed` returned by this method, which will
/// break them into separate objects, allowing them to interact more easily.
pub fn new(inner: T, codec: U) -> Framed<T, U> {
Framed {
inner: framed_read2(framed_write2(Fuse(inner, codec))),
}
}
}
impl<T, U> Framed<T, U> {
/// Provides a `Stream` and `Sink` interface for reading and writing to this
/// `Io` object, using `Decode` and `Encode` to read and write the raw data.
///
/// Raw I/O objects work with byte sequences, but higher-level code usually
/// wants to batch these into meaningful chunks, called "frames". This
/// method layers framing on top of an I/O object, by using the `Codec`
/// traits to handle encoding and decoding of messages frames. Note that
/// the incoming and outgoing frame types may be distinct.
///
/// This function returns a *single* object that is both `Stream` and
/// `Sink`; grouping this into a single object is often useful for layering
/// things like gzip or TLS, which require both read and write access to the
/// underlying object.
///
/// This objects takes a stream and a readbuffer and a writebuffer. These field
/// can be obtained from an existing `Framed` with the `into_parts` method.
///
/// If you want to work more directly with the streams and sink, consider
/// calling `split` on the `Framed` returned by this method, which will
/// break them into separate objects, allowing them to interact more easily.
pub fn from_parts(parts: FramedParts<T, U>) -> Framed<T, U> {
Framed {
inner: framed_read2_with_buffer(
framed_write2_with_buffer(Fuse(parts.io, parts.codec), parts.write_buf),
parts.read_buf,
),
}
}
/// Returns a reference to the underlying codec.
pub fn get_codec(&self) -> &U {
&self.inner.get_ref().get_ref().1
}
/// Returns a mutable reference to the underlying codec.
pub fn get_codec_mut(&mut self) -> &mut U {
&mut self.inner.get_mut().get_mut().1
}
/// Returns a reference to the underlying I/O stream wrapped by
/// `Frame`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_ref(&self) -> &T {
&self.inner.get_ref().get_ref().0
}
/// Returns a mutable reference to the underlying I/O stream wrapped by
/// `Frame`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner.get_mut().get_mut().0
}
/// Consumes the `Frame`, returning its underlying I/O stream.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn into_inner(self) -> T {
self.inner.into_inner().into_inner().0
}
/// Consumes the `Frame`, returning its underlying I/O stream, the buffer
/// with unprocessed data, and the codec.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn into_parts(self) -> FramedParts<T, U> {
let (inner, read_buf) = self.inner.into_parts();
let (inner, write_buf) = inner.into_parts();
FramedParts {
io: inner.0,
codec: inner.1,
read_buf: read_buf,
write_buf: write_buf,
_priv: (),
}
}
}
impl<T, U> Stream for Framed<T, U>
where
T: AsyncRead,
U: Decoder,
{
type Item = U::Item;
type Error = U::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.inner.poll()
}
}
impl<T, U> Sink for Framed<T, U>
where
T: AsyncWrite,
U: Encoder,
U::Error: From<io::Error>,
{
type SinkItem = U::Item;
type SinkError = U::Error;
fn start_send(
&mut self, item: Self::SinkItem,
) -> StartSend<Self::SinkItem, Self::SinkError> {
self.inner.get_mut().start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
self.inner.get_mut().poll_complete()
}
fn close(&mut self) -> Poll<(), Self::SinkError> {
self.inner.get_mut().close()
}
}
impl<T, U> fmt::Debug for Framed<T, U>
where
T: fmt::Debug,
U: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("Framed")
.field("io", &self.inner.get_ref().get_ref().0)
.field("codec", &self.inner.get_ref().get_ref().1)
.finish()
}
}
// ===== impl Fuse =====
impl<T: Read, U> Read for Fuse<T, U> {
fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
self.0.read(dst)
}
}
impl<T: AsyncRead, U> AsyncRead for Fuse<T, U> {
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
self.0.prepare_uninitialized_buffer(buf)
}
}
impl<T: Write, U> Write for Fuse<T, U> {
fn write(&mut self, src: &[u8]) -> io::Result<usize> {
self.0.write(src)
}
fn flush(&mut self) -> io::Result<()> {
self.0.flush()
}
}
impl<T: AsyncWrite, U> AsyncWrite for Fuse<T, U> {
fn shutdown(&mut self) -> Poll<(), io::Error> {
self.0.shutdown()
}
}
impl<T, U: Decoder> Decoder for Fuse<T, U> {
type Item = U::Item;
type Error = U::Error;
fn decode(
&mut self, buffer: &mut BytesMut,
) -> Result<Option<Self::Item>, Self::Error> {
self.1.decode(buffer)
}
fn decode_eof(
&mut self, buffer: &mut BytesMut,
) -> Result<Option<Self::Item>, Self::Error> {
self.1.decode_eof(buffer)
}
}
impl<T, U: Encoder> Encoder for Fuse<T, U> {
type Item = U::Item;
type Error = U::Error;
fn encode(
&mut self, item: Self::Item, dst: &mut BytesMut,
) -> Result<(), Self::Error> {
self.1.encode(item, dst)
}
}
/// `FramedParts` contains an export of the data of a Framed transport.
/// It can be used to construct a new `Framed` with a different codec.
/// It contains all current buffers and the inner transport.
#[derive(Debug)]
pub struct FramedParts<T, U> {
/// The inner transport used to read bytes to and write bytes to
pub io: T,
/// The codec
pub codec: U,
/// The buffer with read but unprocessed data.
pub read_buf: BytesMut,
/// A buffer with unprocessed data which are not written yet.
pub write_buf: BytesMut,
/// This private field allows us to add additional fields in the future in a
/// backwards compatible way.
_priv: (),
}
impl<T, U> FramedParts<T, U> {
/// Create a new, default, `FramedParts`
pub fn new(io: T, codec: U) -> FramedParts<T, U> {
FramedParts {
io,
codec,
read_buf: BytesMut::new(),
write_buf: BytesMut::new(),
_priv: (),
}
}
}

216
src/framed/framed_read.rs
View file

@ -1,216 +0,0 @@
use std::fmt;
use bytes::BytesMut;
use futures::{Async, Poll, Sink, StartSend, Stream};
use tokio_codec::Decoder;
use tokio_io::AsyncRead;
use super::framed::Fuse;
/// A `Stream` of messages decoded from an `AsyncRead`.
pub struct FramedRead<T, D> {
inner: FramedRead2<Fuse<T, D>>,
}
pub struct FramedRead2<T> {
inner: T,
eof: bool,
is_readable: bool,
buffer: BytesMut,
}
const INITIAL_CAPACITY: usize = 8 * 1024;
// ===== impl FramedRead =====
impl<T, D> FramedRead<T, D>
where
T: AsyncRead,
D: Decoder,
{
/// Creates a new `FramedRead` with the given `decoder`.
pub fn new(inner: T, decoder: D) -> FramedRead<T, D> {
FramedRead {
inner: framed_read2(Fuse(inner, decoder)),
}
}
}
impl<T, D> FramedRead<T, D> {
/// Returns a reference to the underlying I/O stream wrapped by
/// `FramedRead`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_ref(&self) -> &T {
&self.inner.inner.0
}
/// Returns a mutable reference to the underlying I/O stream wrapped by
/// `FramedRead`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner.inner.0
}
/// Consumes the `FramedRead`, returning its underlying I/O stream.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn into_inner(self) -> T {
self.inner.inner.0
}
/// Returns a reference to the underlying decoder.
pub fn decoder(&self) -> &D {
&self.inner.inner.1
}
/// Returns a mutable reference to the underlying decoder.
pub fn decoder_mut(&mut self) -> &mut D {
&mut self.inner.inner.1
}
}
impl<T, D> Stream for FramedRead<T, D>
where
T: AsyncRead,
D: Decoder,
{
type Item = D::Item;
type Error = D::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.inner.poll()
}
}
impl<T, D> Sink for FramedRead<T, D>
where
T: Sink,
{
type SinkItem = T::SinkItem;
type SinkError = T::SinkError;
fn start_send(
&mut self, item: Self::SinkItem,
) -> StartSend<Self::SinkItem, Self::SinkError> {
self.inner.inner.0.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
self.inner.inner.0.poll_complete()
}
fn close(&mut self) -> Poll<(), Self::SinkError> {
self.inner.inner.0.close()
}
}
impl<T, D> fmt::Debug for FramedRead<T, D>
where
T: fmt::Debug,
D: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FramedRead")
.field("inner", &self.inner.inner.0)
.field("decoder", &self.inner.inner.1)
.field("eof", &self.inner.eof)
.field("is_readable", &self.inner.is_readable)
.field("buffer", &self.inner.buffer)
.finish()
}
}
// ===== impl FramedRead2 =====
pub fn framed_read2<T>(inner: T) -> FramedRead2<T> {
FramedRead2 {
inner: inner,
eof: false,
is_readable: false,
buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
}
}
pub fn framed_read2_with_buffer<T>(inner: T, mut buf: BytesMut) -> FramedRead2<T> {
if buf.capacity() < INITIAL_CAPACITY {
let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity();
buf.reserve(bytes_to_reserve);
}
FramedRead2 {
inner: inner,
eof: false,
is_readable: buf.len() > 0,
buffer: buf,
}
}
impl<T> FramedRead2<T> {
pub fn get_ref(&self) -> &T {
&self.inner
}
pub fn into_inner(self) -> T {
self.inner
}
pub fn into_parts(self) -> (T, BytesMut) {
(self.inner, self.buffer)
}
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner
}
}
impl<T> Stream for FramedRead2<T>
where
T: AsyncRead + Decoder,
{
type Item = T::Item;
type Error = T::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
loop {
// Repeatedly call `decode` or `decode_eof` as long as it is
// "readable". Readable is defined as not having returned `None`. If
// the upstream has returned EOF, and the decoder is no longer
// readable, it can be assumed that the decoder will never become
// readable again, at which point the stream is terminated.
if self.is_readable {
if self.eof {
let frame = try!(self.inner.decode_eof(&mut self.buffer));
return Ok(Async::Ready(frame));
}
trace!("attempting to decode a frame");
if let Some(frame) = try!(self.inner.decode(&mut self.buffer)) {
trace!("frame decoded from buffer");
return Ok(Async::Ready(Some(frame)));
}
self.is_readable = false;
}
assert!(!self.eof);
// Otherwise, try to read more data and try again. Make sure we've
// got room for at least one byte to read to ensure that we don't
// get a spurious 0 that looks like EOF
self.buffer.reserve(1);
if 0 == try_ready!(self.inner.read_buf(&mut self.buffer)) {
self.eof = true;
}
self.is_readable = true;
}
}
}

243
src/framed/framed_write.rs
View file

@ -1,243 +0,0 @@
use std::fmt;
use std::io::{self, Read};
use bytes::BytesMut;
use futures::{Async, AsyncSink, Poll, Sink, StartSend, Stream};
use tokio_codec::{Decoder, Encoder};
use tokio_io::{AsyncRead, AsyncWrite};
use super::framed::Fuse;
/// A `Sink` of frames encoded to an `AsyncWrite`.
pub struct FramedWrite<T, E> {
inner: FramedWrite2<Fuse<T, E>>,
}
pub struct FramedWrite2<T> {
inner: T,
buffer: BytesMut,
}
const INITIAL_CAPACITY: usize = 8 * 1024;
const BACKPRESSURE_BOUNDARY: usize = INITIAL_CAPACITY;
impl<T, E> FramedWrite<T, E>
where
T: AsyncWrite,
E: Encoder,
{
/// Creates a new `FramedWrite` with the given `encoder`.
pub fn new(inner: T, encoder: E) -> FramedWrite<T, E> {
FramedWrite {
inner: framed_write2(Fuse(inner, encoder)),
}
}
}
impl<T, E> FramedWrite<T, E> {
/// Returns a reference to the underlying I/O stream wrapped by
/// `FramedWrite`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_ref(&self) -> &T {
&self.inner.inner.0
}
/// Returns a mutable reference to the underlying I/O stream wrapped by
/// `FramedWrite`.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner.inner.0
}
/// Consumes the `FramedWrite`, returning its underlying I/O stream.
///
/// Note that care should be taken to not tamper with the underlying stream
/// of data coming in as it may corrupt the stream of frames otherwise
/// being worked with.
pub fn into_inner(self) -> T {
self.inner.inner.0
}
/// Returns a reference to the underlying decoder.
pub fn encoder(&self) -> &E {
&self.inner.inner.1
}
/// Returns a mutable reference to the underlying decoder.
pub fn encoder_mut(&mut self) -> &mut E {
&mut self.inner.inner.1
}
}
impl<T, E> Sink for FramedWrite<T, E>
where
T: AsyncWrite,
E: Encoder,
{
type SinkItem = E::Item;
type SinkError = E::Error;
fn start_send(&mut self, item: E::Item) -> StartSend<E::Item, E::Error> {
self.inner.start_send(item)
}
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
self.inner.poll_complete()
}
fn close(&mut self) -> Poll<(), Self::SinkError> {
Ok(try!(self.inner.close()))
}
}
impl<T, D> Stream for FramedWrite<T, D>
where
T: Stream,
{
type Item = T::Item;
type Error = T::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
self.inner.inner.0.poll()
}
}
impl<T, U> fmt::Debug for FramedWrite<T, U>
where
T: fmt::Debug,
U: fmt::Debug,
{
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("FramedWrite")
.field("inner", &self.inner.get_ref().0)
.field("encoder", &self.inner.get_ref().1)
.field("buffer", &self.inner.buffer)
.finish()
}
}
// ===== impl FramedWrite2 =====
pub fn framed_write2<T>(inner: T) -> FramedWrite2<T> {
FramedWrite2 {
inner: inner,
buffer: BytesMut::with_capacity(INITIAL_CAPACITY),
}
}
pub fn framed_write2_with_buffer<T>(inner: T, mut buf: BytesMut) -> FramedWrite2<T> {
if buf.capacity() < INITIAL_CAPACITY {
let bytes_to_reserve = INITIAL_CAPACITY - buf.capacity();
buf.reserve(bytes_to_reserve);
}
FramedWrite2 {
inner: inner,
buffer: buf,
}
}
impl<T> FramedWrite2<T> {
pub fn get_ref(&self) -> &T {
&self.inner
}
pub fn into_inner(self) -> T {
self.inner
}
pub fn into_parts(self) -> (T, BytesMut) {
(self.inner, self.buffer)
}
pub fn get_mut(&mut self) -> &mut T {
&mut self.inner
}
}
impl<T> Sink for FramedWrite2<T>
where
T: AsyncWrite + Encoder,
{
type SinkItem = T::Item;
type SinkError = T::Error;
fn start_send(&mut self, item: T::Item) -> StartSend<T::Item, T::Error> {
// If the buffer is already over 8KiB, then attempt to flush it. If after flushing it's
// *still* over 8KiB, then apply backpressure (reject the send).
if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
try!(self.poll_complete());
if self.buffer.len() >= BACKPRESSURE_BOUNDARY {
return Ok(AsyncSink::NotReady(item));
}
}
try!(self.inner.encode(item, &mut self.buffer));
Ok(AsyncSink::Ready)
}
fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
trace!("flushing framed transport");
while !self.buffer.is_empty() {
trace!("writing; remaining={}", self.buffer.len());
let n = try_ready!(self.inner.poll_write(&self.buffer));
if n == 0 {
return Err(io::Error::new(
io::ErrorKind::WriteZero,
"failed to \
write frame to transport",
).into());
}
// TODO: Add a way to `bytes` to do this w/o returning the drained
// data.
let _ = self.buffer.split_to(n);
}
// Try flushing the underlying IO
try_ready!(self.inner.poll_flush());
trace!("framed transport flushed");
return Ok(Async::Ready(()));
}
fn close(&mut self) -> Poll<(), Self::SinkError> {
try_ready!(self.poll_complete());
Ok(try!(self.inner.shutdown()))
}
}
impl<T: Decoder> Decoder for FramedWrite2<T> {
type Item = T::Item;
type Error = T::Error;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<T::Item>, T::Error> {
self.inner.decode(src)
}
fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<T::Item>, T::Error> {
self.inner.decode_eof(src)
}
}
impl<T: Read> Read for FramedWrite2<T> {
fn read(&mut self, dst: &mut [u8]) -> io::Result<usize> {
self.inner.read(dst)
}
}
impl<T: AsyncRead> AsyncRead for FramedWrite2<T> {
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
self.inner.prepare_uninitialized_buffer(buf)
}
}

32
src/framed/mod.rs
View file

@ -1,32 +0,0 @@
//! Utilities for encoding and decoding frames.
//!
//! Contains adapters to go from streams of bytes, [`AsyncRead`] and
//! [`AsyncWrite`], to framed streams implementing [`Sink`] and [`Stream`].
//! Framed streams are also known as [transports].
//!
//! [`AsyncRead`]: #
//! [`AsyncWrite`]: #
//! [`Sink`]: #
//! [`Stream`]: #
//! [transports]: #
#![deny(missing_docs, missing_debug_implementations, warnings)]
#![doc(hidden, html_root_url = "https://docs.rs/tokio-codec/0.1.0")]
// _tokio_codec are the items that belong in the `tokio_codec` crate. However, because we need to
// maintain backward compatibility until the next major breaking change, they are defined here.
// When the next breaking change comes, they should be moved to the `tokio_codec` crate and become
// independent.
//
// The primary reason we can't move these to `tokio-codec` now is because, again for backward
// compatibility reasons, we need to keep `Decoder` and `Encoder` in tokio_io::codec. And `Decoder`
// and `Encoder` needs to reference `Framed`. So they all still need to still be in the same
// module.
mod framed;
mod framed_read;
mod framed_write;
pub use self::framed::{Framed, FramedParts};
pub use self::framed_read::FramedRead;
pub use self::framed_write::FramedWrite;

4
src/h1/dispatcher.rs
View file

@ -1,12 +1,11 @@
// #![allow(unused_imports, unused_variables, dead_code)]
use std::collections::VecDeque;
use std::fmt::{Debug, Display};
use std::time::Instant;
use actix_net::codec::Framed;
use actix_net::service::Service;
use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
// use tokio_current_thread::spawn;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_timer::Delay;
@ -16,7 +15,6 @@ use payload::{Payload, PayloadSender, PayloadStatus, PayloadWriter};
use body::Body;
use config::ServiceConfig;
use error::DispatchError;
use framed::Framed;
use request::Request;
use response::Response;

3
src/lib.rs
View file

@ -135,9 +135,6 @@ mod request;
mod response;
mod uri;
#[doc(hidden)]
pub mod framed;
pub mod error;
pub mod h1;
pub(crate) mod helpers;

4
src/ws/mod.rs
View file

@ -15,10 +15,12 @@ mod codec;
mod frame;
mod mask;
mod proto;
mod transport;
pub use self::codec::Message;
pub use self::codec::{Codec, Message};
pub use self::frame::Frame;
pub use self::proto::{CloseCode, CloseReason, OpCode};
pub use self::transport::Transport;
/// Websocket protocol errors
#[derive(Fail, Debug)]

50
src/ws/transport.rs Normal file
View file

@ -0,0 +1,50 @@
use actix_net::codec::Framed;
use actix_net::framed::{FramedTransport, FramedTransportError};
use actix_net::service::{IntoService, Service};
use futures::{Future, Poll};
use tokio_io::{AsyncRead, AsyncWrite};
use super::{Codec, Message};
pub struct Transport<S, T>
where
S: Service,
T: AsyncRead + AsyncWrite,
{
inner: FramedTransport<S, T, Codec>,
}
impl<S, T> Transport<S, T>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Message, Response = Message>,
S::Future: 'static,
S::Error: 'static,
{
pub fn new<F: IntoService<S>>(io: T, service: F) -> Self {
Transport {
inner: FramedTransport::new(Framed::new(io, Codec::new()), service),
}
}
pub fn with<F: IntoService<S>>(framed: Framed<T, Codec>, service: F) -> Self {
Transport {
inner: FramedTransport::new(framed, service),
}
}
}
impl<S, T> Future for Transport<S, T>
where
T: AsyncRead + AsyncWrite,
S: Service<Request = Message, Response = Message>,
S::Future: 'static,
S::Error: 'static,
{
type Item = ();
type Error = FramedTransportError<S::Error, Codec>;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
self.inner.poll()
}
}

111
tests/test_ws.rs Normal file
View file

@ -0,0 +1,111 @@
extern crate actix;
extern crate actix_http;
extern crate actix_net;
extern crate actix_web;
extern crate bytes;
extern crate futures;
use std::{io, thread};
use actix::System;
use actix_net::codec::Framed;
use actix_net::server::Server;
use actix_net::service::IntoNewService;
use actix_web::{test, ws as web_ws};
use bytes::Bytes;
use futures::future::{ok, Either};
use futures::{Future, Sink, Stream};
use actix_http::{h1, ws, ResponseError};
fn ws_handler(req: ws::Message) -> impl Future<Item = ws::Message, Error = io::Error> {
match req {
ws::Message::Ping(msg) => ok(ws::Message::Pong(msg)),
ws::Message::Text(text) => ok(ws::Message::Text(text)),
ws::Message::Binary(bin) => ok(ws::Message::Binary(bin)),
ws::Message::Close(reason) => ok(ws::Message::Close(reason)),
_ => ok(ws::Message::Close(None)),
}
}
#[test]
fn test_simple() {
let addr = test::TestServer::unused_addr();
thread::spawn(move || {
Server::new()
.bind("test", addr, move || {
(|io| {
// read http request
let framed = Framed::new(io, h1::Codec::new(false));
framed
.into_future()
.map_err(|_| ())
.and_then(|(req, framed)| {
// validate request
if let Some(h1::InMessage::MessageWithPayload(req)) = req {
match ws::handshake(&req) {
Err(e) => {
// validation failed
let resp = e.error_response();
Either::A(
framed
.send(h1::OutMessage::Response(resp))
.map_err(|_| ())
.map(|_| ()),
)
}
Ok(mut resp) => Either::B(
// send response
framed
.send(h1::OutMessage::Response(
resp.finish(),
)).map_err(|_| ())
.and_then(|framed| {
// start websocket service
let framed =
framed.into_framed(ws::Codec::new());
ws::Transport::with(framed, ws_handler)
.map_err(|_| ())
}),
),
}
} else {
panic!()
}
})
}).into_new_service()
}).unwrap()
.run();
});
let mut sys = System::new("test");
{
let (reader, mut writer) = sys
.block_on(web_ws::Client::new(format!("http://{}/", addr)).connect())
.unwrap();
writer.text("text");
let (item, reader) = sys.block_on(reader.into_future()).unwrap();
assert_eq!(item, Some(web_ws::Message::Text("text".to_owned())));
writer.binary(b"text".as_ref());
let (item, reader) = sys.block_on(reader.into_future()).unwrap();
assert_eq!(
item,
Some(web_ws::Message::Binary(Bytes::from_static(b"text").into()))
);
writer.ping("ping");
let (item, reader) = sys.block_on(reader.into_future()).unwrap();
assert_eq!(item, Some(web_ws::Message::Pong("ping".to_owned())));
writer.close(Some(web_ws::CloseCode::Normal.into()));
let (item, _) = sys.block_on(reader.into_future()).unwrap();
assert_eq!(
item,
Some(web_ws::Message::Close(Some(
web_ws::CloseCode::Normal.into()
)))
);
}
}