diff --git a/Cargo.toml b/Cargo.toml index 03d481e2..3d8adff9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ readme = "README.md" keywords = ["network", "framework", "async", "futures"] homepage = "https://actix.rs" repository = "https://github.com/actix/actix-net.git" -documentation = "https://actix.rs/api/actix-net/stable/actix_web/" +documentation = "https://actix.rs/api/actix-net/stable/actix_net/" categories = ["network-programming", "asynchronous"] license = "MIT/Apache-2.0" exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"] @@ -17,7 +17,7 @@ features = ["ssl", "tls", "rust-tls"] [badges] travis-ci = { repository = "actix/actix-net", branch = "master" } -appveyor = { repository = "fafhrd91/actix-web-hdy9d" } +# appveyor = { repository = "fafhrd91/actix-web-hdy9d" } codecov = { repository = "actix/actix-net", branch = "master", service = "github" } [lib] @@ -58,7 +58,6 @@ tokio-timer = "0.2" tokio-reactor = "0.1" tokio-current-thread = "0.1" tower-service = "0.1" - trust-dns-resolver = "0.10.0-alpha.2" # native-tls diff --git a/src/codec/framed2.rs b/src/codec/framed2.rs new file mode 100644 index 00000000..08211335 --- /dev/null +++ b/src/codec/framed2.rs @@ -0,0 +1,312 @@ +#![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 Framed2 { + inner: FramedRead2>>, +} + +pub struct Fuse2(pub T, pub D, pub E); + +impl Framed2 +where + T: AsyncRead + AsyncWrite, + D: Decoder, + E: 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, decoder: D, encoder: E) -> Framed2 { + Framed2 { + inner: framed_read2(framed_write2(Fuse2(inner, decoder, encoder))), + } + } +} + +impl Framed2 { + /// 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: FramedParts2) -> Framed2 { + Framed2 { + inner: framed_read2_with_buffer( + framed_write2_with_buffer( + Fuse2(parts.io, parts.decoder, parts.encoder), + parts.write_buf, + ), + parts.read_buf, + ), + } + } + + /// 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 + } + + /// Returns a reference to the underlying decoder. + pub fn decocer(&self) -> &D { + &self.inner.get_ref().get_ref().1 + } + + /// Returns a mutable reference to the underlying decoder. + pub fn decoder_mut(&mut self) -> &mut D { + &mut self.inner.get_mut().get_mut().1 + } + + /// Returns a reference to the underlying encoder. + pub fn encoder(&self) -> &E { + &self.inner.get_ref().get_ref().2 + } + + /// Returns a mutable reference to the underlying codec. + pub fn encoder_mut(&mut self) -> &mut E { + &mut self.inner.get_mut().get_mut().2 + } + + /// 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 + } + + /// Consume the `Frame`, returning `Frame` with different codec. + pub fn switch_encoder(self, encoder: E2) -> Framed2 { + let (inner, read_buf) = self.inner.into_parts(); + let (inner, write_buf) = inner.into_parts(); + + Framed2 { + inner: framed_read2_with_buffer( + framed_write2_with_buffer(Fuse2(inner.0, inner.1, encoder), write_buf), + read_buf, + ), + } + } + + /// 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) -> FramedParts2 { + let (inner, read_buf) = self.inner.into_parts(); + let (inner, write_buf) = inner.into_parts(); + + FramedParts2 { + io: inner.0, + decoder: inner.1, + encoder: inner.2, + read_buf: read_buf, + write_buf: write_buf, + _priv: (), + } + } +} + +impl Stream for Framed2 +where + T: AsyncRead, + D: Decoder, +{ + type Item = D::Item; + type Error = D::Error; + + fn poll(&mut self) -> Poll, Self::Error> { + self.inner.poll() + } +} + +impl Sink for Framed2 +where + T: AsyncWrite, + E: Encoder, + E::Error: From, +{ + type SinkItem = E::Item; + type SinkError = E::Error; + + fn start_send( + &mut self, item: Self::SinkItem, + ) -> StartSend { + 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 fmt::Debug for Framed2 +where + T: fmt::Debug, + D: fmt::Debug, + E: fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + f.debug_struct("Framed2") + .field("io", &self.inner.get_ref().get_ref().0) + .field("decoder", &self.inner.get_ref().get_ref().1) + .field("encoder", &self.inner.get_ref().get_ref().2) + .finish() + } +} + +// ===== impl Fuse2 ===== + +impl Read for Fuse2 { + fn read(&mut self, dst: &mut [u8]) -> io::Result { + self.0.read(dst) + } +} + +impl AsyncRead for Fuse2 { + unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool { + self.0.prepare_uninitialized_buffer(buf) + } +} + +impl Write for Fuse2 { + fn write(&mut self, src: &[u8]) -> io::Result { + self.0.write(src) + } + + fn flush(&mut self) -> io::Result<()> { + self.0.flush() + } +} + +impl AsyncWrite for Fuse2 { + fn shutdown(&mut self) -> Poll<(), io::Error> { + self.0.shutdown() + } +} + +impl Decoder for Fuse2 { + type Item = D::Item; + type Error = D::Error; + + fn decode(&mut self, buffer: &mut BytesMut) -> Result, Self::Error> { + self.1.decode(buffer) + } + + fn decode_eof(&mut self, buffer: &mut BytesMut) -> Result, Self::Error> { + self.1.decode_eof(buffer) + } +} + +impl Encoder for Fuse2 { + type Item = E::Item; + type Error = E::Error; + + fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> { + self.2.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 FramedParts2 { + /// The inner transport used to read bytes to and write bytes to + pub io: T, + + /// The decoder + pub decoder: D, + + /// The encoder + pub encoder: E, + + /// 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 FramedParts2 { + /// Create a new, default, `FramedParts` + pub fn new(io: T, decoder: D, encoder: E) -> FramedParts2 { + FramedParts2 { + io, + decoder, + encoder, + read_buf: BytesMut::new(), + write_buf: BytesMut::new(), + _priv: (), + } + } +} diff --git a/src/codec/mod.rs b/src/codec/mod.rs index 423a6a90..54aa35ff 100644 --- a/src/codec/mod.rs +++ b/src/codec/mod.rs @@ -11,24 +11,13 @@ //! [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 framed2; mod framed_read; mod framed_write; pub use self::framed::{Framed, FramedParts}; +pub use self::framed2::{Framed2, FramedParts2}; pub use self::framed_read::FramedRead; pub use self::framed_write::FramedWrite; diff --git a/src/timer.rs b/src/timer.rs index e4c3d3c6..0acf671f 100644 --- a/src/timer.rs +++ b/src/timer.rs @@ -14,22 +14,22 @@ pub struct LowResTimer(Cell); #[derive(Debug)] struct Inner { - interval: Duration, + resolution: Duration, current: Option, } impl Inner { - fn new(interval: Duration) -> Self { + fn new(resolution: Duration) -> Self { Inner { - interval, + resolution, current: None, } } } impl LowResTimer { - pub fn with_interval(interval: Duration) -> LowResTimer { - LowResTimer(Cell::new(Inner::new(interval))) + pub fn with(resolution: Duration) -> LowResTimer { + LowResTimer(Cell::new(Inner::new(resolution))) } pub fn timer(&self) -> LowResTimerService { @@ -60,7 +60,7 @@ impl NewService for LowResTimer { pub struct LowResTimerService(Cell); impl LowResTimerService { - pub fn with_resolution(resolution: Duration) -> LowResTimerService { + pub fn with(resolution: Duration) -> LowResTimerService { LowResTimerService(Cell::new(Inner::new(resolution))) } @@ -76,7 +76,7 @@ impl LowResTimerService { let interval = { let mut b = inner.borrow_mut(); b.current = Some(now); - b.interval + b.resolution }; spawn(sleep(interval).map_err(|_| panic!()).and_then(move |_| {