mirror of
https://github.com/fafhrd91/actix-net
synced 2025-08-16 10:18:59 +02:00
Compare commits
18 Commits
server-v1.
...
bytestring
Author | SHA1 | Date | |
---|---|---|---|
|
a80e1f8370 | ||
|
5fe759cc02 | ||
|
05549f0b42 | ||
|
b1430eaded | ||
|
0d3f9e74c5 | ||
|
cab73791ed | ||
|
a7ac1a76ed | ||
|
37bedff6fb | ||
|
33fd6adc11 | ||
|
4305cdba2c | ||
|
52ecb4bcc5 | ||
|
b28f32e82c | ||
|
081205a02f | ||
|
8bb81c0768 | ||
|
c7a8743bf9 | ||
|
f26fcc703b | ||
|
ce4587df82 | ||
|
9957f28137 |
20
.github/workflows/main.yml
vendored
20
.github/workflows/main.yml
vendored
@@ -2,6 +2,9 @@ name: CI
|
|||||||
|
|
||||||
on: [push, pull_request]
|
on: [push, pull_request]
|
||||||
|
|
||||||
|
env:
|
||||||
|
VCPKGRS_DYNAMIC: 1
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
build_and_test:
|
build_and_test:
|
||||||
strategy:
|
strategy:
|
||||||
@@ -18,10 +21,12 @@ jobs:
|
|||||||
include:
|
include:
|
||||||
- toolchain: x86_64-pc-windows-msvc
|
- toolchain: x86_64-pc-windows-msvc
|
||||||
os: windows-latest
|
os: windows-latest
|
||||||
|
arch: x64
|
||||||
- toolchain: x86_64-pc-windows-gnu
|
- toolchain: x86_64-pc-windows-gnu
|
||||||
os: windows-latest
|
os: windows-latest
|
||||||
- toolchain: i686-pc-windows-msvc
|
- toolchain: i686-pc-windows-msvc
|
||||||
os: windows-latest
|
os: windows-latest
|
||||||
|
arch: x86
|
||||||
- toolchain: x86_64-apple-darwin
|
- toolchain: x86_64-apple-darwin
|
||||||
os: macOS-latest
|
os: macOS-latest
|
||||||
|
|
||||||
@@ -37,6 +42,13 @@ jobs:
|
|||||||
toolchain: ${{ matrix.version }}-${{ matrix.toolchain }}
|
toolchain: ${{ matrix.version }}-${{ matrix.toolchain }}
|
||||||
default: true
|
default: true
|
||||||
|
|
||||||
|
|
||||||
|
- name: Install OpenSSL
|
||||||
|
if: matrix.toolchain == 'x86_64-pc-windows-msvc' || matrix.toolchain == 'i686-pc-windows-msvc'
|
||||||
|
run: |
|
||||||
|
vcpkg integrate install
|
||||||
|
vcpkg install openssl:${{ matrix.arch }}-windows
|
||||||
|
|
||||||
- name: check nightly
|
- name: check nightly
|
||||||
if: matrix.version == 'nightly'
|
if: matrix.version == 'nightly'
|
||||||
uses: actions-rs/cargo@v1
|
uses: actions-rs/cargo@v1
|
||||||
@@ -52,6 +64,14 @@ jobs:
|
|||||||
args: --all --bins --examples --tests
|
args: --all --bins --examples --tests
|
||||||
|
|
||||||
- name: tests
|
- name: tests
|
||||||
|
if: matrix.toolchain != 'x86_64-pc-windows-gnu'
|
||||||
|
uses: actions-rs/cargo@v1
|
||||||
|
with:
|
||||||
|
command: test
|
||||||
|
args: --all --all-features -- --nocapture
|
||||||
|
|
||||||
|
- name: tests on x86_64-pc-windows-gnu
|
||||||
|
if: matrix.toolchain == 'x86_64-pc-windows-gnu'
|
||||||
uses: actions-rs/cargo@v1
|
uses: actions-rs/cargo@v1
|
||||||
with:
|
with:
|
||||||
command: test
|
command: test
|
||||||
|
@@ -1,5 +1,7 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
* Use `.advance()` intead of `.split_to()`
|
||||||
|
|
||||||
## [0.2.0] - 2019-12-10
|
## [0.2.0] - 2019-12-10
|
||||||
|
|
||||||
* Use specific futures dependencies
|
* Use specific futures dependencies
|
||||||
|
@@ -2,7 +2,7 @@ use std::pin::Pin;
|
|||||||
use std::task::{Context, Poll};
|
use std::task::{Context, Poll};
|
||||||
use std::{fmt, io};
|
use std::{fmt, io};
|
||||||
|
|
||||||
use bytes::BytesMut;
|
use bytes::{Buf, BytesMut};
|
||||||
use futures_core::{ready, Stream};
|
use futures_core::{ready, Stream};
|
||||||
use futures_sink::Sink;
|
use futures_sink::Sink;
|
||||||
|
|
||||||
@@ -288,7 +288,7 @@ impl<T, U> Framed<T, U> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// remove written data
|
// remove written data
|
||||||
let _ = self.write_buf.split_to(n);
|
self.write_buf.advance(n);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Try flushing the underlying IO
|
// Try flushing the underlying IO
|
||||||
|
@@ -1,5 +1,13 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [1.0.1] - 2019-12-15
|
||||||
|
|
||||||
|
* Fix trust-dns-resolver compilation
|
||||||
|
|
||||||
|
## [1.0.0] - 2019-12-11
|
||||||
|
|
||||||
|
* Release
|
||||||
|
|
||||||
## [1.0.0-alpha.3] - 2019-12-07
|
## [1.0.0-alpha.3] - 2019-12-07
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-connect"
|
name = "actix-connect"
|
||||||
version = "1.0.0-alpha.3"
|
version = "1.0.1"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix connect - tcp connector service"
|
description = "Actix connect - tcp connector service"
|
||||||
keywords = ["network", "framework", "async", "futures"]
|
keywords = ["network", "framework", "async", "futures"]
|
||||||
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
|
|||||||
documentation = "https://docs.rs/actix-connect/"
|
documentation = "https://docs.rs/actix-connect/"
|
||||||
categories = ["network-programming", "asynchronous"]
|
categories = ["network-programming", "asynchronous"]
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
|
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
workspace = ".."
|
workspace = ".."
|
||||||
|
|
||||||
@@ -35,14 +34,15 @@ uri = ["http"]
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
actix-service = "1.0.0"
|
actix-service = "1.0.0"
|
||||||
actix-codec = "0.2.0"
|
actix-codec = "0.2.0"
|
||||||
actix-utils = "1.0.0"
|
actix-utils = "1.0.3"
|
||||||
actix-rt = "1.0.0"
|
actix-rt = "1.0.0"
|
||||||
derive_more = "0.99.2"
|
derive_more = "0.99.2"
|
||||||
either = "1.5.2"
|
either = "1.5.2"
|
||||||
futures = "0.3.1"
|
futures = "0.3.1"
|
||||||
http = { version = "0.2.0", optional = true }
|
http = { version = "0.2.0", optional = true }
|
||||||
log = "0.4"
|
log = "0.4"
|
||||||
trust-dns-resolver = { version="0.18.0-alpha.2", default-features = false, features=["tokio"] }
|
trust-dns-proto = "=0.18.0-alpha.2"
|
||||||
|
trust-dns-resolver = "=0.18.0-alpha.2"
|
||||||
|
|
||||||
# openssl
|
# openssl
|
||||||
open-ssl = { version="0.10", package = "openssl", optional = true }
|
open-ssl = { version="0.10", package = "openssl", optional = true }
|
||||||
@@ -54,5 +54,5 @@ tokio-rustls = { version = "0.12.0", optional = true }
|
|||||||
webpki = { version = "0.21", optional = true }
|
webpki = { version = "0.21", optional = true }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
bytes = "0.5.2"
|
bytes = "0.5.3"
|
||||||
actix-testing = { version="1.0.0" }
|
actix-testing = { version="1.0.0" }
|
||||||
|
@@ -1,5 +1,13 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [0.4.1] - 2019-12-11
|
||||||
|
|
||||||
|
* Disconnect callback accepts owned state
|
||||||
|
|
||||||
|
## [0.4.0] - 2019-12-11
|
||||||
|
|
||||||
|
* Remove `E` param
|
||||||
|
|
||||||
## [0.3.0-alpha.3] - 2019-12-07
|
## [0.3.0-alpha.3] - 2019-12-07
|
||||||
|
|
||||||
* Migrate to tokio 0.2
|
* Migrate to tokio 0.2
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-ioframe"
|
name = "actix-ioframe"
|
||||||
version = "0.3.0-alpha.3"
|
version = "0.4.1"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix framed service"
|
description = "Actix framed service"
|
||||||
keywords = ["network", "framework", "async", "futures"]
|
keywords = ["network", "framework", "async", "futures"]
|
||||||
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
|
|||||||
documentation = "https://docs.rs/actix-ioframe/"
|
documentation = "https://docs.rs/actix-ioframe/"
|
||||||
categories = ["network-programming", "asynchronous"]
|
categories = ["network-programming", "asynchronous"]
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
|
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
workspace = ".."
|
workspace = ".."
|
||||||
|
|
||||||
@@ -20,7 +19,7 @@ path = "src/lib.rs"
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
actix-service = "1.0.0"
|
actix-service = "1.0.0"
|
||||||
actix-codec = "0.2.0"
|
actix-codec = "0.2.0"
|
||||||
actix-utils = "1.0.0"
|
actix-utils = "1.0.1"
|
||||||
actix-rt = "1.0.0"
|
actix-rt = "1.0.0"
|
||||||
bytes = "0.5"
|
bytes = "0.5"
|
||||||
either = "1.5.2"
|
either = "1.5.2"
|
||||||
@@ -29,5 +28,5 @@ pin-project = "0.4.6"
|
|||||||
log = "0.4"
|
log = "0.4"
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
actix-connect = "1.0.0-alpha.2"
|
actix-connect = "1.0.0"
|
||||||
actix-testing = "1.0.0"
|
actix-testing = "1.0.0"
|
||||||
|
@@ -7,21 +7,21 @@ use futures::Stream;
|
|||||||
|
|
||||||
use crate::sink::Sink;
|
use crate::sink::Sink;
|
||||||
|
|
||||||
pub struct Connect<Io, Codec, Err, St = ()>
|
pub struct Connect<Io, Codec, St = ()>
|
||||||
where
|
where
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
{
|
{
|
||||||
io: Io,
|
io: Io,
|
||||||
sink: Sink<<Codec as Encoder>::Item, Err>,
|
sink: Sink<<Codec as Encoder>::Item>,
|
||||||
_t: PhantomData<(St, Codec)>,
|
_t: PhantomData<(St, Codec)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<Io, Codec, Err> Connect<Io, Codec, Err>
|
impl<Io, Codec> Connect<Io, Codec>
|
||||||
where
|
where
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
{
|
{
|
||||||
pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item, Err>) -> Self {
|
pub(crate) fn new(io: Io, sink: Sink<<Codec as Encoder>::Item>) -> Self {
|
||||||
Self {
|
Self {
|
||||||
io,
|
io,
|
||||||
sink,
|
sink,
|
||||||
@@ -29,7 +29,7 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec, Err> {
|
pub fn codec(self, codec: Codec) -> ConnectResult<Io, (), Codec> {
|
||||||
ConnectResult {
|
ConnectResult {
|
||||||
state: (),
|
state: (),
|
||||||
sink: self.sink,
|
sink: self.sink,
|
||||||
@@ -39,15 +39,15 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[pin_project::pin_project]
|
#[pin_project::pin_project]
|
||||||
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder, Err> {
|
pub struct ConnectResult<Io, St, Codec: Encoder + Decoder> {
|
||||||
pub(crate) state: St,
|
pub(crate) state: St,
|
||||||
pub(crate) framed: Framed<Io, Codec>,
|
pub(crate) framed: Framed<Io, Codec>,
|
||||||
pub(crate) sink: Sink<<Codec as Encoder>::Item, Err>,
|
pub(crate) sink: Sink<<Codec as Encoder>::Item>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
|
impl<Io, St, Codec: Encoder + Decoder> ConnectResult<Io, St, Codec> {
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, Err> {
|
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
|
||||||
&self.sink
|
&self.sink
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -62,7 +62,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec, Err> {
|
pub fn state<S>(self, state: S) -> ConnectResult<Io, S, Codec> {
|
||||||
ConnectResult {
|
ConnectResult {
|
||||||
state,
|
state,
|
||||||
framed: self.framed,
|
framed: self.framed,
|
||||||
@@ -71,7 +71,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<Io, St, Codec, Err> Stream for ConnectResult<Io, St, Codec, Err>
|
impl<Io, St, Codec> Stream for ConnectResult<Io, St, Codec>
|
||||||
where
|
where
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
@@ -83,8 +83,7 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<Io, St, Codec, Err> futures::Sink<<Codec as Encoder>::Item>
|
impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
|
||||||
for ConnectResult<Io, St, Codec, Err>
|
|
||||||
where
|
where
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
|
@@ -13,7 +13,7 @@ use crate::error::ServiceError;
|
|||||||
use crate::item::Item;
|
use crate::item::Item;
|
||||||
use crate::sink::Sink;
|
use crate::sink::Sink;
|
||||||
|
|
||||||
type Request<S, U, E> = Item<S, U, E>;
|
type Request<S, U> = Item<S, U>;
|
||||||
type Response<U> = <U as Encoder>::Item;
|
type Response<U> = <U as Encoder>::Item;
|
||||||
|
|
||||||
pub(crate) enum Message<T> {
|
pub(crate) enum Message<T> {
|
||||||
@@ -25,10 +25,10 @@ pub(crate) enum Message<T> {
|
|||||||
/// FramedTransport - is a future that reads frames from Framed object
|
/// FramedTransport - is a future that reads frames from Framed object
|
||||||
/// and pass then to the service.
|
/// and pass then to the service.
|
||||||
#[pin_project::pin_project]
|
#[pin_project::pin_project]
|
||||||
pub(crate) struct Dispatcher<St, S, T, U, E>
|
pub(crate) struct Dispatcher<St, S, T, U>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
|
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
|
||||||
S::Error: 'static,
|
S::Error: 'static,
|
||||||
S::Future: 'static,
|
S::Future: 'static,
|
||||||
T: AsyncRead + AsyncWrite,
|
T: AsyncRead + AsyncWrite,
|
||||||
@@ -37,19 +37,19 @@ where
|
|||||||
<U as Encoder>::Error: std::fmt::Debug,
|
<U as Encoder>::Error: std::fmt::Debug,
|
||||||
{
|
{
|
||||||
service: S,
|
service: S,
|
||||||
sink: Sink<<U as Encoder>::Item, E>,
|
sink: Sink<<U as Encoder>::Item>,
|
||||||
state: St,
|
state: St,
|
||||||
dispatch_state: FramedState<S, U>,
|
dispatch_state: FramedState<S, U>,
|
||||||
framed: Framed<T, U>,
|
framed: Framed<T, U>,
|
||||||
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
|
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
|
||||||
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, E>>,
|
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, S, T, U, E> Dispatcher<St, S, T, U, E>
|
impl<St, S, T, U> Dispatcher<St, S, T, U>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
|
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
|
||||||
S::Error: 'static,
|
S::Error: 'static,
|
||||||
S::Future: 'static,
|
S::Future: 'static,
|
||||||
T: AsyncRead + AsyncWrite,
|
T: AsyncRead + AsyncWrite,
|
||||||
@@ -61,9 +61,9 @@ where
|
|||||||
framed: Framed<T, U>,
|
framed: Framed<T, U>,
|
||||||
state: St,
|
state: St,
|
||||||
service: F,
|
service: F,
|
||||||
sink: Sink<<U as Encoder>::Item, E>,
|
sink: Sink<<U as Encoder>::Item>,
|
||||||
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
|
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
let tx = rx.sender();
|
let tx = rx.sender();
|
||||||
|
|
||||||
@@ -126,10 +126,10 @@ impl<S: Service, U: Encoder + Decoder> FramedState<S, U> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, S, T, U, E> Dispatcher<St, S, T, U, E>
|
impl<St, S, T, U> Dispatcher<St, S, T, U>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
S: Service<Request = Request<St, U, E>, Response = Option<Response<U>>, Error = E>,
|
S: Service<Request = Request<St, U>, Response = Option<Response<U>>>,
|
||||||
S::Error: 'static,
|
S::Error: 'static,
|
||||||
S::Future: 'static,
|
S::Future: 'static,
|
||||||
T: AsyncRead + AsyncWrite,
|
T: AsyncRead + AsyncWrite,
|
||||||
@@ -245,7 +245,7 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if let Some(ref disconnect) = self.disconnect {
|
if let Some(ref disconnect) = self.disconnect {
|
||||||
(&*disconnect)(&mut self.state, true);
|
(&*disconnect)(self.state.clone(), true);
|
||||||
}
|
}
|
||||||
Poll::Ready(Err(self.dispatch_state.take_error()))
|
Poll::Ready(Err(self.dispatch_state.take_error()))
|
||||||
}
|
}
|
||||||
@@ -265,19 +265,19 @@ where
|
|||||||
let _ = tx.send(());
|
let _ = tx.send(());
|
||||||
}
|
}
|
||||||
if let Some(ref disconnect) = self.disconnect {
|
if let Some(ref disconnect) = self.disconnect {
|
||||||
(&*disconnect)(&mut self.state, false);
|
(&*disconnect)(self.state.clone(), false);
|
||||||
}
|
}
|
||||||
Poll::Ready(Ok(()))
|
Poll::Ready(Ok(()))
|
||||||
}
|
}
|
||||||
FramedState::FramedError(_) => {
|
FramedState::FramedError(_) => {
|
||||||
if let Some(ref disconnect) = self.disconnect {
|
if let Some(ref disconnect) = self.disconnect {
|
||||||
(&*disconnect)(&mut self.state, true);
|
(&*disconnect)(self.state.clone(), true);
|
||||||
}
|
}
|
||||||
Poll::Ready(Err(self.dispatch_state.take_framed_error()))
|
Poll::Ready(Err(self.dispatch_state.take_framed_error()))
|
||||||
}
|
}
|
||||||
FramedState::Stopping => {
|
FramedState::Stopping => {
|
||||||
if let Some(ref disconnect) = self.disconnect {
|
if let Some(ref disconnect) = self.disconnect {
|
||||||
(&*disconnect)(&mut self.state, false);
|
(&*disconnect)(self.state.clone(), false);
|
||||||
}
|
}
|
||||||
Poll::Ready(Ok(()))
|
Poll::Ready(Ok(()))
|
||||||
}
|
}
|
||||||
|
@@ -5,19 +5,19 @@ use actix_codec::{Decoder, Encoder};
|
|||||||
|
|
||||||
use crate::sink::Sink;
|
use crate::sink::Sink;
|
||||||
|
|
||||||
pub struct Item<St, Codec: Encoder + Decoder, E> {
|
pub struct Item<St, Codec: Encoder + Decoder> {
|
||||||
state: St,
|
state: St,
|
||||||
sink: Sink<<Codec as Encoder>::Item, E>,
|
sink: Sink<<Codec as Encoder>::Item>,
|
||||||
item: <Codec as Decoder>::Item,
|
item: <Codec as Decoder>::Item,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Codec, E> Item<St, Codec, E>
|
impl<St, Codec> Item<St, Codec>
|
||||||
where
|
where
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
{
|
{
|
||||||
pub(crate) fn new(
|
pub(crate) fn new(
|
||||||
state: St,
|
state: St,
|
||||||
sink: Sink<<Codec as Encoder>::Item, E>,
|
sink: Sink<<Codec as Encoder>::Item>,
|
||||||
item: <Codec as Decoder>::Item,
|
item: <Codec as Decoder>::Item,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Item { state, sink, item }
|
Item { state, sink, item }
|
||||||
@@ -34,7 +34,7 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, E> {
|
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
|
||||||
&self.sink
|
&self.sink
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -44,18 +44,12 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn into_parts(
|
pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::Item) {
|
||||||
self,
|
|
||||||
) -> (
|
|
||||||
St,
|
|
||||||
Sink<<Codec as Encoder>::Item, E>,
|
|
||||||
<Codec as Decoder>::Item,
|
|
||||||
) {
|
|
||||||
(self.state, self.sink, self.item)
|
(self.state, self.sink, self.item)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Codec, E> Deref for Item<St, Codec, E>
|
impl<St, Codec> Deref for Item<St, Codec>
|
||||||
where
|
where
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
{
|
{
|
||||||
@@ -67,7 +61,7 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Codec, E> DerefMut for Item<St, Codec, E>
|
impl<St, Codec> DerefMut for Item<St, Codec>
|
||||||
where
|
where
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
{
|
{
|
||||||
@@ -77,7 +71,7 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Codec, E> fmt::Debug for Item<St, Codec, E>
|
impl<St, Codec> fmt::Debug for Item<St, Codec>
|
||||||
where
|
where
|
||||||
Codec: Encoder + Decoder,
|
Codec: Encoder + Decoder,
|
||||||
<Codec as Decoder>::Item: fmt::Debug,
|
<Codec as Decoder>::Item: fmt::Debug,
|
||||||
|
@@ -17,7 +17,7 @@ use crate::error::ServiceError;
|
|||||||
use crate::item::Item;
|
use crate::item::Item;
|
||||||
use crate::sink::Sink;
|
use crate::sink::Sink;
|
||||||
|
|
||||||
type RequestItem<S, U, E> = Item<S, U, E>;
|
type RequestItem<S, U> = Item<S, U>;
|
||||||
type ResponseItem<U> = Option<<U as Encoder>::Item>;
|
type ResponseItem<U> = Option<<U as Encoder>::Item>;
|
||||||
type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;
|
type ServiceResult<U, E> = Result<Message<<U as Encoder>::Item>, E>;
|
||||||
|
|
||||||
@@ -37,15 +37,11 @@ impl<St: Clone, Codec> Builder<St, Codec> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Construct framed handler service with specified connect service
|
/// Construct framed handler service with specified connect service
|
||||||
pub fn service<Io, C, F, E>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec, E>
|
pub fn service<Io, C, F>(self, connect: F) -> ServiceBuilder<St, C, Io, Codec>
|
||||||
where
|
where
|
||||||
F: IntoService<C>,
|
F: IntoService<C>,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, E>,
|
|
||||||
Response = ConnectResult<Io, St, Codec, E>,
|
|
||||||
Error = E,
|
|
||||||
>,
|
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
{
|
{
|
||||||
ServiceBuilder {
|
ServiceBuilder {
|
||||||
@@ -56,17 +52,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Construct framed handler new service with specified connect service
|
/// Construct framed handler new service with specified connect service
|
||||||
pub fn factory<Io, C, F, E>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec, E>
|
pub fn factory<Io, C, F>(self, connect: F) -> NewServiceBuilder<St, C, Io, Codec>
|
||||||
where
|
where
|
||||||
F: IntoServiceFactory<C>,
|
F: IntoServiceFactory<C>,
|
||||||
E: 'static,
|
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
C: ServiceFactory<
|
C: ServiceFactory<
|
||||||
Config = (),
|
Config = (),
|
||||||
Request = Connect<Io, Codec, E>,
|
Request = Connect<Io, Codec>,
|
||||||
Response = ConnectResult<Io, St, Codec, E>,
|
Response = ConnectResult<Io, St, Codec>,
|
||||||
Error = E,
|
|
||||||
>,
|
>,
|
||||||
|
C::Error: 'static,
|
||||||
C::Future: 'static,
|
C::Future: 'static,
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
{
|
{
|
||||||
@@ -78,20 +73,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct ServiceBuilder<St, C, Io, Codec, Err> {
|
pub struct ServiceBuilder<St, C, Io, Codec> {
|
||||||
connect: C,
|
connect: C,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
_t: PhantomData<(St, Io, Codec, Err)>,
|
_t: PhantomData<(St, Io, Codec)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, C, Io, Codec, Err> ServiceBuilder<St, C, Io, Codec, Err>
|
impl<St, C, Io, Codec> ServiceBuilder<St, C, Io, Codec>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
<Codec as Encoder>::Item: 'static,
|
<Codec as Encoder>::Item: 'static,
|
||||||
@@ -102,22 +93,22 @@ where
|
|||||||
/// Second parameter indicates error occured during disconnect.
|
/// Second parameter indicates error occured during disconnect.
|
||||||
pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
|
pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
|
||||||
where
|
where
|
||||||
F: Fn(&mut St, bool) + 'static,
|
F: Fn(St, bool) + 'static,
|
||||||
{
|
{
|
||||||
self.disconnect = Some(Rc::new(disconnect));
|
self.disconnect = Some(Rc::new(disconnect));
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Provide stream items handler service and construct service factory.
|
/// Provide stream items handler service and construct service factory.
|
||||||
pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec, Err>
|
pub fn finish<F, T>(self, service: F) -> FramedServiceImpl<St, C, T, Io, Codec>
|
||||||
where
|
where
|
||||||
F: IntoServiceFactory<T>,
|
F: IntoServiceFactory<T>,
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
{
|
{
|
||||||
FramedServiceImpl {
|
FramedServiceImpl {
|
||||||
@@ -129,23 +120,22 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct NewServiceBuilder<St, C, Io, Codec, Err> {
|
pub struct NewServiceBuilder<St, C, Io, Codec> {
|
||||||
connect: C,
|
connect: C,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
_t: PhantomData<(St, Io, Codec, Err)>,
|
_t: PhantomData<(St, Io, Codec)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, C, Io, Codec, Err> NewServiceBuilder<St, C, Io, Codec, Err>
|
impl<St, C, Io, Codec> NewServiceBuilder<St, C, Io, Codec>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
Err: 'static,
|
|
||||||
C: ServiceFactory<
|
C: ServiceFactory<
|
||||||
Config = (),
|
Config = (),
|
||||||
Request = Connect<Io, Codec, Err>,
|
Request = Connect<Io, Codec>,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
Response = ConnectResult<Io, St, Codec>,
|
||||||
Error = Err,
|
|
||||||
>,
|
>,
|
||||||
|
C::Error: 'static,
|
||||||
C::Future: 'static,
|
C::Future: 'static,
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
<Codec as Encoder>::Item: 'static,
|
<Codec as Encoder>::Item: 'static,
|
||||||
@@ -156,21 +146,21 @@ where
|
|||||||
/// Second parameter indicates error occured during disconnect.
|
/// Second parameter indicates error occured during disconnect.
|
||||||
pub fn disconnect<F>(mut self, disconnect: F) -> Self
|
pub fn disconnect<F>(mut self, disconnect: F) -> Self
|
||||||
where
|
where
|
||||||
F: Fn(&mut St, bool) + 'static,
|
F: Fn(St, bool) + 'static,
|
||||||
{
|
{
|
||||||
self.disconnect = Some(Rc::new(disconnect));
|
self.disconnect = Some(Rc::new(disconnect));
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Err, Cfg>
|
pub fn finish<F, T, Cfg>(self, service: F) -> FramedService<St, C, T, Io, Codec, Cfg>
|
||||||
where
|
where
|
||||||
F: IntoServiceFactory<T>,
|
F: IntoServiceFactory<T>,
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
> + 'static,
|
> + 'static,
|
||||||
{
|
{
|
||||||
FramedService {
|
FramedService {
|
||||||
@@ -182,34 +172,32 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct FramedService<St, C, T, Io, Codec, Err, Cfg> {
|
pub struct FramedService<St, C, T, Io, Codec, Cfg> {
|
||||||
connect: C,
|
connect: C,
|
||||||
handler: Rc<T>,
|
handler: Rc<T>,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
_t: PhantomData<(St, Io, Codec, Err, Cfg)>,
|
_t: PhantomData<(St, Io, Codec, Cfg)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, C, T, Io, Codec, Err, Cfg> ServiceFactory
|
impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
|
||||||
for FramedService<St, C, T, Io, Codec, Err, Cfg>
|
|
||||||
where
|
where
|
||||||
St: Clone + 'static,
|
St: Clone + 'static,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
C: ServiceFactory<
|
C: ServiceFactory<
|
||||||
Config = (),
|
Config = (),
|
||||||
Request = Connect<Io, Codec, Err>,
|
Request = Connect<Io, Codec>,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
Response = ConnectResult<Io, St, Codec>,
|
||||||
Error = Err,
|
|
||||||
>,
|
>,
|
||||||
|
C::Error: 'static,
|
||||||
C::Future: 'static,
|
C::Future: 'static,
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
> + 'static,
|
> + 'static,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Err: 'static,
|
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
<Codec as Encoder>::Item: 'static,
|
<Codec as Encoder>::Item: 'static,
|
||||||
<Codec as Encoder>::Error: std::fmt::Debug,
|
<Codec as Encoder>::Error: std::fmt::Debug,
|
||||||
@@ -219,7 +207,7 @@ where
|
|||||||
type Response = ();
|
type Response = ();
|
||||||
type Error = ServiceError<C::Error, Codec>;
|
type Error = ServiceError<C::Error, Codec>;
|
||||||
type InitError = C::InitError;
|
type InitError = C::InitError;
|
||||||
type Service = FramedServiceImpl<St, C::Service, T, Io, Codec, Err>;
|
type Service = FramedServiceImpl<St, C::Service, T, Io, Codec>;
|
||||||
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
|
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::InitError>>;
|
||||||
|
|
||||||
fn new_service(&self, _: Cfg) -> Self::Future {
|
fn new_service(&self, _: Cfg) -> Self::Future {
|
||||||
@@ -241,29 +229,25 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub struct FramedServiceImpl<St, C, T, Io, Codec, Err> {
|
pub struct FramedServiceImpl<St, C, T, Io, Codec> {
|
||||||
connect: C,
|
connect: C,
|
||||||
handler: Rc<T>,
|
handler: Rc<T>,
|
||||||
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
|
disconnect: Option<Rc<dyn Fn(St, bool)>>,
|
||||||
_t: PhantomData<(St, Io, Codec, Err)>,
|
_t: PhantomData<(St, Io, Codec)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, C, T, Io, Codec, Err> Service for FramedServiceImpl<St, C, T, Io, Codec, Err>
|
impl<St, C, T, Io, Codec> Service for FramedServiceImpl<St, C, T, Io, Codec>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
C::Error: 'static,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Err: 'static,
|
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Codec: Decoder + Encoder,
|
Codec: Decoder + Encoder,
|
||||||
@@ -272,8 +256,8 @@ where
|
|||||||
{
|
{
|
||||||
type Request = Io;
|
type Request = Io;
|
||||||
type Response = ();
|
type Response = ();
|
||||||
type Error = ServiceError<Err, Codec>;
|
type Error = ServiceError<C::Error, Codec>;
|
||||||
type Future = FramedServiceImplResponse<St, Io, Codec, Err, C, T>;
|
type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
|
||||||
|
|
||||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||||
self.connect.poll_ready(cx).map_err(|e| e.into())
|
self.connect.poll_ready(cx).map_err(|e| e.into())
|
||||||
@@ -281,7 +265,9 @@ where
|
|||||||
|
|
||||||
fn call(&mut self, req: Io) -> Self::Future {
|
fn call(&mut self, req: Io) -> Self::Future {
|
||||||
let (tx, rx) = mpsc::channel();
|
let (tx, rx) = mpsc::channel();
|
||||||
let sink = Sink::new(tx);
|
let sink = Sink::new(Rc::new(move |msg| {
|
||||||
|
let _ = tx.send(Ok(msg));
|
||||||
|
}));
|
||||||
FramedServiceImplResponse {
|
FramedServiceImplResponse {
|
||||||
inner: FramedServiceImplResponseInner::Connect(
|
inner: FramedServiceImplResponseInner::Connect(
|
||||||
self.connect.call(Connect::new(req, sink.clone())),
|
self.connect.call(Connect::new(req, sink.clone())),
|
||||||
@@ -294,21 +280,17 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[pin_project::pin_project]
|
#[pin_project::pin_project]
|
||||||
pub struct FramedServiceImplResponse<St, Io, Codec, Err, C, T>
|
pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
C::Error: 'static,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Err: 'static,
|
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
@@ -317,24 +299,20 @@ where
|
|||||||
<Codec as Encoder>::Error: std::fmt::Debug,
|
<Codec as Encoder>::Error: std::fmt::Debug,
|
||||||
{
|
{
|
||||||
#[pin]
|
#[pin]
|
||||||
inner: FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
|
inner: FramedServiceImplResponseInner<St, Io, Codec, C, T>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Io, Codec, Err, C, T> Future for FramedServiceImplResponse<St, Io, Codec, Err, C, T>
|
impl<St, Io, Codec, C, T> Future for FramedServiceImplResponse<St, Io, Codec, C, T>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
C::Error: 'static,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Err: 'static,
|
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
@@ -342,7 +320,7 @@ where
|
|||||||
<Codec as Encoder>::Item: 'static,
|
<Codec as Encoder>::Item: 'static,
|
||||||
<Codec as Encoder>::Error: std::fmt::Debug,
|
<Codec as Encoder>::Error: std::fmt::Debug,
|
||||||
{
|
{
|
||||||
type Output = Result<(), ServiceError<Err, Codec>>;
|
type Output = Result<(), ServiceError<C::Error, Codec>>;
|
||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
let mut this = self.as_mut().project();
|
let mut this = self.as_mut().project();
|
||||||
@@ -360,21 +338,17 @@ where
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[pin_project::pin_project]
|
#[pin_project::pin_project]
|
||||||
enum FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
|
enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
C::Error: 'static,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Err: 'static,
|
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
@@ -385,33 +359,29 @@ where
|
|||||||
Connect(
|
Connect(
|
||||||
#[pin] C::Future,
|
#[pin] C::Future,
|
||||||
Rc<T>,
|
Rc<T>,
|
||||||
Option<Rc<dyn Fn(&mut St, bool)>>,
|
Option<Rc<dyn Fn(St, bool)>>,
|
||||||
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
|
Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
|
||||||
),
|
),
|
||||||
Handler(
|
Handler(
|
||||||
#[pin] T::Future,
|
#[pin] T::Future,
|
||||||
Option<ConnectResult<Io, St, Codec, Err>>,
|
Option<ConnectResult<Io, St, Codec>>,
|
||||||
Option<Rc<dyn Fn(&mut St, bool)>>,
|
Option<Rc<dyn Fn(St, bool)>>,
|
||||||
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
|
Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
|
||||||
),
|
),
|
||||||
Dispatcher(Dispatcher<St, T::Service, Io, Codec, Err>),
|
Dispatcher(Dispatcher<St, T::Service, Io, Codec>),
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<St, Io, Codec, Err, C, T> FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
|
impl<St, Io, Codec, C, T> FramedServiceImplResponseInner<St, Io, Codec, C, T>
|
||||||
where
|
where
|
||||||
St: Clone,
|
St: Clone,
|
||||||
C: Service<
|
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
|
||||||
Request = Connect<Io, Codec, Err>,
|
C::Error: 'static,
|
||||||
Response = ConnectResult<Io, St, Codec, Err>,
|
|
||||||
Error = Err,
|
|
||||||
>,
|
|
||||||
Err: 'static,
|
|
||||||
T: ServiceFactory<
|
T: ServiceFactory<
|
||||||
Config = St,
|
Config = St,
|
||||||
Request = RequestItem<St, Codec, Err>,
|
Request = RequestItem<St, Codec>,
|
||||||
Response = ResponseItem<Codec>,
|
Response = ResponseItem<Codec>,
|
||||||
Error = Err,
|
Error = C::Error,
|
||||||
InitError = Err,
|
InitError = C::Error,
|
||||||
>,
|
>,
|
||||||
<T::Service as Service>::Future: 'static,
|
<T::Service as Service>::Future: 'static,
|
||||||
Io: AsyncRead + AsyncWrite,
|
Io: AsyncRead + AsyncWrite,
|
||||||
@@ -424,8 +394,8 @@ where
|
|||||||
self: Pin<&mut Self>,
|
self: Pin<&mut Self>,
|
||||||
cx: &mut Context<'_>,
|
cx: &mut Context<'_>,
|
||||||
) -> Either<
|
) -> Either<
|
||||||
FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
|
FramedServiceImplResponseInner<St, Io, Codec, C, T>,
|
||||||
Poll<Result<(), ServiceError<Err, Codec>>>,
|
Poll<Result<(), ServiceError<C::Error, Codec>>>,
|
||||||
> {
|
> {
|
||||||
#[project]
|
#[project]
|
||||||
match self.project() {
|
match self.project() {
|
||||||
|
@@ -1,43 +1,44 @@
|
|||||||
use std::fmt;
|
use std::fmt;
|
||||||
|
use std::rc::Rc;
|
||||||
|
|
||||||
use actix_utils::{mpsc, oneshot};
|
use actix_utils::oneshot;
|
||||||
use futures::future::{Future, FutureExt};
|
use futures::future::{Future, FutureExt};
|
||||||
|
|
||||||
use crate::dispatcher::Message;
|
use crate::dispatcher::Message;
|
||||||
|
|
||||||
pub struct Sink<T, E>(mpsc::Sender<Result<Message<T>, E>>);
|
pub struct Sink<T>(Rc<dyn Fn(Message<T>)>);
|
||||||
|
|
||||||
impl<T, E> Clone for Sink<T, E> {
|
impl<T> Clone for Sink<T> {
|
||||||
fn clone(&self) -> Self {
|
fn clone(&self) -> Self {
|
||||||
Sink(self.0.clone())
|
Sink(self.0.clone())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T, E> Sink<T, E> {
|
impl<T> Sink<T> {
|
||||||
pub(crate) fn new(tx: mpsc::Sender<Result<Message<T>, E>>) -> Self {
|
pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
|
||||||
Sink(tx)
|
Sink(tx)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Close connection
|
/// Close connection
|
||||||
pub fn close(&self) {
|
pub fn close(&self) {
|
||||||
let _ = self.0.send(Ok(Message::Close));
|
(self.0)(Message::Close);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Close connection
|
/// Close connection
|
||||||
pub fn wait_close(&self) -> impl Future<Output = ()> {
|
pub fn wait_close(&self) -> impl Future<Output = ()> {
|
||||||
let (tx, rx) = oneshot::channel();
|
let (tx, rx) = oneshot::channel();
|
||||||
let _ = self.0.send(Ok(Message::WaitClose(tx)));
|
(self.0)(Message::WaitClose(tx));
|
||||||
|
|
||||||
rx.map(|_| ())
|
rx.map(|_| ())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Send item
|
/// Send item
|
||||||
pub fn send(&self, item: T) {
|
pub fn send(&self, item: T) {
|
||||||
let _ = self.0.send(Ok(Message::Item(item)));
|
(self.0)(Message::Item(item));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T, E> fmt::Debug for Sink<T, E> {
|
impl<T> fmt::Debug for Sink<T> {
|
||||||
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
|
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
fmt.debug_struct("Sink").finish()
|
fmt.debug_struct("Sink").finish()
|
||||||
}
|
}
|
||||||
|
@@ -22,7 +22,7 @@ async fn test_disconnect() -> std::io::Result<()> {
|
|||||||
let disconnect1 = disconnect1.clone();
|
let disconnect1 = disconnect1.clone();
|
||||||
|
|
||||||
Builder::new()
|
Builder::new()
|
||||||
.factory(fn_service(|conn: Connect<_, _, _>| {
|
.factory(fn_service(|conn: Connect<_, _>| {
|
||||||
ok(conn.codec(BytesCodec).state(State))
|
ok(conn.codec(BytesCodec).state(State))
|
||||||
}))
|
}))
|
||||||
.disconnect(move |_, _| {
|
.disconnect(move |_, _| {
|
||||||
@@ -32,7 +32,7 @@ async fn test_disconnect() -> std::io::Result<()> {
|
|||||||
});
|
});
|
||||||
|
|
||||||
let mut client = Builder::new()
|
let mut client = Builder::new()
|
||||||
.service(|conn: Connect<_, _, _>| {
|
.service(|conn: Connect<_, _>| {
|
||||||
let conn = conn.codec(BytesCodec).state(State);
|
let conn = conn.codec(BytesCodec).state(State);
|
||||||
conn.sink().close();
|
conn.sink().close();
|
||||||
ok(conn)
|
ok(conn)
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-macros"
|
name = "actix-macros"
|
||||||
version = "0.1.0"
|
version = "0.1.1"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix runtime macros"
|
description = "Actix runtime macros"
|
||||||
repository = "https://github.com/actix/actix-net"
|
repository = "https://github.com/actix/actix-net"
|
||||||
|
1
actix-macros/LICENSE-APACHE
Symbolic link
1
actix-macros/LICENSE-APACHE
Symbolic link
@@ -0,0 +1 @@
|
|||||||
|
../LICENSE-APACHE
|
1
actix-macros/LICENSE-MIT
Symbolic link
1
actix-macros/LICENSE-MIT
Symbolic link
@@ -0,0 +1 @@
|
|||||||
|
../LICENSE-MIT
|
@@ -71,7 +71,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
|
|||||||
if input.sig.asyncness.is_none() {
|
if input.sig.asyncness.is_none() {
|
||||||
return syn::Error::new_spanned(
|
return syn::Error::new_spanned(
|
||||||
input.sig.fn_token,
|
input.sig.fn_token,
|
||||||
format!("only async fn is supported, {:?}", input.sig.ident),
|
format!("only async fn is supported, {}", input.sig.ident),
|
||||||
)
|
)
|
||||||
.to_compile_error()
|
.to_compile_error()
|
||||||
.into();
|
.into();
|
||||||
|
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-rt/"
|
|||||||
categories = ["network-programming", "asynchronous"]
|
categories = ["network-programming", "asynchronous"]
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
workspace = ".."
|
|
||||||
|
|
||||||
[lib]
|
[lib]
|
||||||
name = "actix_rt"
|
name = "actix_rt"
|
||||||
@@ -21,4 +20,4 @@ actix-macros = "0.1.0"
|
|||||||
actix-threadpool = "0.3"
|
actix-threadpool = "0.3"
|
||||||
futures = "0.3.1"
|
futures = "0.3.1"
|
||||||
copyless = "0.1.4"
|
copyless = "0.1.4"
|
||||||
tokio = { version = "0.2.4", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }
|
tokio = { version = "0.2.6", default-features=false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }
|
||||||
|
@@ -39,4 +39,4 @@ mio-uds = { version = "0.6.7" }
|
|||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
bytes = "0.5"
|
bytes = "0.5"
|
||||||
env_logger = "0.6"
|
env_logger = "0.6"
|
||||||
actix-testing = "1.0.0-alpha.3"
|
actix-testing = "1.0.0"
|
@@ -1,5 +1,9 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [1.0.0] - 2019-12-11
|
||||||
|
|
||||||
|
* Update actix-server to 1.0.0
|
||||||
|
|
||||||
## [1.0.0-alpha.3] - 2019-12-07
|
## [1.0.0-alpha.3] - 2019-12-07
|
||||||
|
|
||||||
* Migrate to tokio 0.2
|
* Migrate to tokio 0.2
|
||||||
|
@@ -1,5 +1,11 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [0.3.1] - 2019-12-12
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
|
||||||
|
* Use parking_lot 0.10
|
||||||
|
|
||||||
## [0.3.0] - 2019-12-02
|
## [0.3.0] - 2019-12-02
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-threadpool"
|
name = "actix-threadpool"
|
||||||
version = "0.3.0"
|
version = "0.3.1"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix thread pool for sync code"
|
description = "Actix thread pool for sync code"
|
||||||
keywords = ["actix", "network", "framework", "async", "futures"]
|
keywords = ["actix", "network", "framework", "async", "futures"]
|
||||||
@@ -19,9 +19,9 @@ path = "src/lib.rs"
|
|||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
derive_more = "0.99.2"
|
derive_more = "0.99.2"
|
||||||
futures = "0.3.1"
|
futures-channel = "0.3.1"
|
||||||
parking_lot = "0.9"
|
parking_lot = "0.10"
|
||||||
lazy_static = "1.2"
|
lazy_static = "1.3"
|
||||||
log = "0.4"
|
log = "0.4"
|
||||||
num_cpus = "1.10"
|
num_cpus = "1.10"
|
||||||
threadpool = "1.7"
|
threadpool = "1.7"
|
||||||
|
@@ -6,7 +6,7 @@ use std::pin::Pin;
|
|||||||
use std::task::{Context, Poll};
|
use std::task::{Context, Poll};
|
||||||
|
|
||||||
use derive_more::Display;
|
use derive_more::Display;
|
||||||
use futures::channel::oneshot;
|
use futures_channel::oneshot;
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
use threadpool::ThreadPool;
|
use threadpool::ThreadPool;
|
||||||
|
|
||||||
@@ -79,9 +79,12 @@ impl<I, E: fmt::Debug> Future for CpuFuture<I, E> {
|
|||||||
|
|
||||||
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||||
let rx = Pin::new(&mut self.rx);
|
let rx = Pin::new(&mut self.rx);
|
||||||
let res = futures::ready!(rx.poll(cx))
|
let res = match rx.poll(cx) {
|
||||||
.map_err(|_| BlockingError::Canceled)
|
Poll::Pending => return Poll::Pending,
|
||||||
.and_then(|res| res.map_err(BlockingError::Error));
|
Poll::Ready(res) => res
|
||||||
|
.map_err(|_| BlockingError::Canceled)
|
||||||
|
.and_then(|res| res.map_err(BlockingError::Error)),
|
||||||
|
};
|
||||||
Poll::Ready(res)
|
Poll::Ready(res)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@@ -1,6 +1,10 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
[1.0.0-alpha.3] - 2019-12-07
|
## [1.0.0] - 2019-12-11
|
||||||
|
|
||||||
|
* 1.0.0 release
|
||||||
|
|
||||||
|
## [1.0.0-alpha.3] - 2019-12-07
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-tls"
|
name = "actix-tls"
|
||||||
version = "1.0.0-alpha.3"
|
version = "1.0.0"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix tls services"
|
description = "Actix tls services"
|
||||||
keywords = ["network", "framework", "async", "futures"]
|
keywords = ["network", "framework", "async", "futures"]
|
||||||
|
@@ -1,5 +1,23 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
|
## [1.0.4] - 2019-12-20
|
||||||
|
|
||||||
|
* Add methods to check `LocalWaker` registration state.
|
||||||
|
|
||||||
|
## [1.0.3] - 2019-12-11
|
||||||
|
|
||||||
|
* Revert InOrder service changes
|
||||||
|
|
||||||
|
## [1.0.2] - 2019-12-11
|
||||||
|
|
||||||
|
* Allow to create `framed::Dispatcher` with custom `mpsc::Receiver`
|
||||||
|
|
||||||
|
* Add `oneshot::Sender::is_canceled()` method
|
||||||
|
|
||||||
|
## [1.0.1] - 2019-12-11
|
||||||
|
|
||||||
|
* Optimize InOrder service
|
||||||
|
|
||||||
## [1.0.0] - 2019-12-11
|
## [1.0.0] - 2019-12-11
|
||||||
|
|
||||||
* Simplify oneshot and mpsc implementations
|
* Simplify oneshot and mpsc implementations
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "actix-utils"
|
name = "actix-utils"
|
||||||
version = "1.0.0"
|
version = "1.0.4"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "Actix utils - various actix net related services"
|
description = "Actix utils - various actix net related services"
|
||||||
keywords = ["network", "framework", "async", "futures"]
|
keywords = ["network", "framework", "async", "futures"]
|
||||||
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-utils/"
|
|||||||
categories = ["network-programming", "asynchronous"]
|
categories = ["network-programming", "asynchronous"]
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
workspace = ".."
|
|
||||||
|
|
||||||
[lib]
|
[lib]
|
||||||
name = "actix_utils"
|
name = "actix_utils"
|
||||||
@@ -20,8 +19,8 @@ path = "src/lib.rs"
|
|||||||
actix-service = "1.0.0"
|
actix-service = "1.0.0"
|
||||||
actix-rt = "1.0.0"
|
actix-rt = "1.0.0"
|
||||||
actix-codec = "0.2.0"
|
actix-codec = "0.2.0"
|
||||||
bytes = "0.5.2"
|
bytes = "0.5.3"
|
||||||
either = "1.5.2"
|
either = "1.5.3"
|
||||||
futures = "0.3.1"
|
futures = "0.3.1"
|
||||||
pin-project = "0.4.6"
|
pin-project = "0.4.6"
|
||||||
log = "0.4"
|
log = "0.4"
|
||||||
|
@@ -127,30 +127,44 @@ where
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Construct new `Dispatcher` instance with customer `mpsc::Receiver`
|
||||||
|
pub fn with_rx<F: IntoService<S>>(
|
||||||
|
framed: Framed<T, U>,
|
||||||
|
service: F,
|
||||||
|
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
|
||||||
|
) -> Self {
|
||||||
|
let tx = rx.sender();
|
||||||
|
Dispatcher {
|
||||||
|
framed,
|
||||||
|
rx,
|
||||||
|
tx,
|
||||||
|
service: service.into_service(),
|
||||||
|
state: State::Processing,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Get sink
|
/// Get sink
|
||||||
pub fn get_sink(&self) -> mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>> {
|
pub fn get_sink(&self) -> mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>> {
|
||||||
self.tx.clone()
|
self.tx.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get reference to a service wrapped by `FramedTransport` instance.
|
/// Get reference to a service wrapped by `Dispatcher` instance.
|
||||||
pub fn get_ref(&self) -> &S {
|
pub fn get_ref(&self) -> &S {
|
||||||
&self.service
|
&self.service
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get mutable reference to a service wrapped by `FramedTransport`
|
/// Get mutable reference to a service wrapped by `Dispatcher` instance.
|
||||||
/// instance.
|
|
||||||
pub fn get_mut(&mut self) -> &mut S {
|
pub fn get_mut(&mut self) -> &mut S {
|
||||||
&mut self.service
|
&mut self.service
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get reference to a framed instance wrapped by `FramedTransport`
|
/// Get reference to a framed instance wrapped by `Dispatcher`
|
||||||
/// instance.
|
/// instance.
|
||||||
pub fn get_framed(&self) -> &Framed<T, U> {
|
pub fn get_framed(&self) -> &Framed<T, U> {
|
||||||
&self.framed
|
&self.framed
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get mutable reference to a framed instance wrapped by `FramedTransport`
|
/// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
|
||||||
/// instance.
|
|
||||||
pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
|
pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
|
||||||
&mut self.framed
|
&mut self.framed
|
||||||
}
|
}
|
||||||
|
@@ -67,6 +67,12 @@ impl<T> Sender<T> {
|
|||||||
Err(val)
|
Err(val)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Tests to see whether this `Sender`'s corresponding `Receiver`
|
||||||
|
/// has gone away.
|
||||||
|
pub fn is_canceled(&self) -> bool {
|
||||||
|
self.inner.strong_count() == 1
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T> Drop for Sender<T> {
|
impl<T> Drop for Sender<T> {
|
||||||
@@ -110,7 +116,9 @@ mod tests {
|
|||||||
assert_eq!(rx.await.unwrap(), "test");
|
assert_eq!(rx.await.unwrap(), "test");
|
||||||
|
|
||||||
let (tx, rx) = channel();
|
let (tx, rx) = channel();
|
||||||
|
assert!(!tx.is_canceled());
|
||||||
drop(rx);
|
drop(rx);
|
||||||
|
assert!(tx.is_canceled());
|
||||||
assert!(tx.send("test").is_err());
|
assert!(tx.send("test").is_err());
|
||||||
|
|
||||||
let (tx, rx) = channel::<&'static str>();
|
let (tx, rx) = channel::<&'static str>();
|
||||||
|
@@ -105,7 +105,7 @@ where
|
|||||||
|
|
||||||
pub struct InOrderService<S: Service> {
|
pub struct InOrderService<S: Service> {
|
||||||
service: S,
|
service: S,
|
||||||
task: Rc<LocalWaker>,
|
waker: Rc<LocalWaker>,
|
||||||
acks: VecDeque<Record<S::Response, S::Error>>,
|
acks: VecDeque<Record<S::Response, S::Error>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -123,7 +123,7 @@ where
|
|||||||
Self {
|
Self {
|
||||||
service: service.into_service(),
|
service: service.into_service(),
|
||||||
acks: VecDeque::new(),
|
acks: VecDeque::new(),
|
||||||
task: Rc::new(LocalWaker::new()),
|
waker: Rc::new(LocalWaker::new()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -142,7 +142,7 @@ where
|
|||||||
|
|
||||||
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
||||||
// poll_ready could be called from different task
|
// poll_ready could be called from different task
|
||||||
self.task.register(cx.waker());
|
self.waker.register(cx.waker());
|
||||||
|
|
||||||
// check acks
|
// check acks
|
||||||
while !self.acks.is_empty() {
|
while !self.acks.is_empty() {
|
||||||
@@ -172,11 +172,11 @@ where
|
|||||||
let (tx2, rx2) = oneshot::channel();
|
let (tx2, rx2) = oneshot::channel();
|
||||||
self.acks.push_back(Record { rx: rx1, tx: tx2 });
|
self.acks.push_back(Record { rx: rx1, tx: tx2 });
|
||||||
|
|
||||||
let task = self.task.clone();
|
let waker = self.waker.clone();
|
||||||
let fut = self.service.call(request);
|
let fut = self.service.call(request);
|
||||||
actix_rt::spawn(async move {
|
actix_rt::spawn(async move {
|
||||||
let res = fut.await;
|
let res = fut.await;
|
||||||
task.wake();
|
waker.wake();
|
||||||
let _ = tx1.send(res);
|
let _ = tx1.send(res);
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -211,7 +211,7 @@ mod tests {
|
|||||||
use super::*;
|
use super::*;
|
||||||
use actix_service::Service;
|
use actix_service::Service;
|
||||||
use futures::channel::oneshot;
|
use futures::channel::oneshot;
|
||||||
use futures::future::{lazy, FutureExt, LocalBoxFuture};
|
use futures::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
|
||||||
|
|
||||||
struct Srv;
|
struct Srv;
|
||||||
|
|
||||||
@@ -245,15 +245,18 @@ mod tests {
|
|||||||
let _ = actix_rt::System::new("test").block_on(async {
|
let _ = actix_rt::System::new("test").block_on(async {
|
||||||
let mut srv = InOrderService::new(Srv);
|
let mut srv = InOrderService::new(Srv);
|
||||||
|
|
||||||
|
let _ = lazy(|cx| srv.poll_ready(cx)).await;
|
||||||
let res1 = srv.call(rx1);
|
let res1 = srv.call(rx1);
|
||||||
let res2 = srv.call(rx2);
|
let res2 = srv.call(rx2);
|
||||||
let res3 = srv.call(rx3);
|
let res3 = srv.call(rx3);
|
||||||
|
|
||||||
let _ = lazy(|cx| srv.poll_ready(cx)).await;
|
actix_rt::spawn(async move {
|
||||||
|
let _ = poll_fn(|cx| {
|
||||||
// dispatcher do this
|
let _ = srv.poll_ready(cx);
|
||||||
actix_rt::time::delay_for(Duration::from_millis(100)).await;
|
Poll::<()>::Pending
|
||||||
let _ = lazy(|cx| srv.poll_ready(cx)).await;
|
})
|
||||||
|
.await;
|
||||||
|
});
|
||||||
|
|
||||||
assert_eq!(res1.await.unwrap(), 1);
|
assert_eq!(res1.await.unwrap(), 1);
|
||||||
assert_eq!(res2.await.unwrap(), 2);
|
assert_eq!(res2.await.unwrap(), 2);
|
||||||
|
@@ -34,11 +34,22 @@ impl LocalWaker {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[inline]
|
||||||
|
/// Check if waker has been registered.
|
||||||
|
pub fn is_registed(&self) -> bool {
|
||||||
|
unsafe { (*self.waker.get()).is_some() }
|
||||||
|
}
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
/// Registers the waker to be notified on calls to `wake`.
|
/// Registers the waker to be notified on calls to `wake`.
|
||||||
pub fn register(&self, waker: &Waker) {
|
///
|
||||||
|
/// Returns `true` if waker was registered before.
|
||||||
|
pub fn register(&self, waker: &Waker) -> bool {
|
||||||
unsafe {
|
unsafe {
|
||||||
*self.waker.get() = Some(waker.clone());
|
let w = self.waker.get();
|
||||||
|
let is_registered = (*w).is_some();
|
||||||
|
*w = Some(waker.clone());
|
||||||
|
is_registered
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -1,9 +1,15 @@
|
|||||||
# Changes
|
# Changes
|
||||||
|
|
||||||
[0.1.1] - 2019-12-07
|
## [0.1.2] - 2019-12-22
|
||||||
|
|
||||||
|
* Fix `new()` method
|
||||||
|
|
||||||
|
* Make `ByteString::from_static()` and `ByteString::from_bytes_unchecked()` methods const.
|
||||||
|
|
||||||
|
## [0.1.1] - 2019-12-07
|
||||||
|
|
||||||
* Fix hash impl
|
* Fix hash impl
|
||||||
|
|
||||||
[0.1.0] - 2019-12-07
|
## [0.1.0] - 2019-12-07
|
||||||
|
|
||||||
* Initial release
|
* Initial release
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "bytestring"
|
name = "bytestring"
|
||||||
version = "0.1.1"
|
version = "0.1.2"
|
||||||
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
|
||||||
description = "A UTF-8 encoded string with Bytes as a storage"
|
description = "A UTF-8 encoded string with Bytes as a storage"
|
||||||
keywords = ["actix"]
|
keywords = ["actix"]
|
||||||
@@ -9,11 +9,10 @@ repository = "https://github.com/actix/actix-net.git"
|
|||||||
documentation = "https://docs.rs/bytestring/"
|
documentation = "https://docs.rs/bytestring/"
|
||||||
license = "MIT/Apache-2.0"
|
license = "MIT/Apache-2.0"
|
||||||
edition = "2018"
|
edition = "2018"
|
||||||
workspace = ".."
|
|
||||||
|
|
||||||
[lib]
|
[lib]
|
||||||
name = "bytestring"
|
name = "bytestring"
|
||||||
path = "src/lib.rs"
|
path = "src/lib.rs"
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
bytes = "0.5.2"
|
bytes = "0.5.3"
|
||||||
|
@@ -6,14 +6,14 @@ use bytes::Bytes;
|
|||||||
|
|
||||||
/// A utf-8 encoded string with [`Bytes`] as a storage.
|
/// A utf-8 encoded string with [`Bytes`] as a storage.
|
||||||
///
|
///
|
||||||
/// [`Bytes`]: https://docs.rs/bytes/0.5.2/bytes/struct.Bytes.html
|
/// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
|
||||||
#[derive(Clone, Eq, PartialEq, Ord, PartialOrd, Default)]
|
#[derive(Clone, Eq, PartialEq, Ord, PartialOrd, Default)]
|
||||||
pub struct ByteString(Bytes);
|
pub struct ByteString(Bytes);
|
||||||
|
|
||||||
impl ByteString {
|
impl ByteString {
|
||||||
/// Creates a new `ByteString`.
|
/// Creates a new `ByteString`.
|
||||||
pub fn new() -> String {
|
pub fn new() -> Self {
|
||||||
String::default()
|
ByteString(Bytes::new())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get a reference to the underlying bytes object.
|
/// Get a reference to the underlying bytes object.
|
||||||
@@ -27,12 +27,12 @@ impl ByteString {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Creates a new `ByteString` from a static str.
|
/// Creates a new `ByteString` from a static str.
|
||||||
pub fn from_static(src: &'static str) -> ByteString {
|
pub const fn from_static(src: &'static str) -> ByteString {
|
||||||
Self(Bytes::from_static(src.as_ref()))
|
Self(Bytes::from_static(src.as_bytes()))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Creates a new `ByteString` from a Bytes.
|
/// Creates a new `ByteString` from a Bytes.
|
||||||
pub unsafe fn from_bytes_unchecked(src: Bytes) -> ByteString {
|
pub const unsafe fn from_bytes_unchecked(src: Bytes) -> ByteString {
|
||||||
Self(src)
|
Self(src)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -147,6 +147,11 @@ mod test {
|
|||||||
use std::collections::hash_map::DefaultHasher;
|
use std::collections::hash_map::DefaultHasher;
|
||||||
use std::hash::{Hash, Hasher};
|
use std::hash::{Hash, Hasher};
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_new() {
|
||||||
|
let _: ByteString = ByteString::new();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_hash() {
|
fn test_hash() {
|
||||||
let mut hasher1 = DefaultHasher::default();
|
let mut hasher1 = DefaultHasher::default();
|
||||||
@@ -171,6 +176,7 @@ mod test {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_from_static_str() {
|
fn test_from_static_str() {
|
||||||
|
const _S: ByteString = ByteString::from_static("hello");
|
||||||
let _ = ByteString::from_static("str");
|
let _ = ByteString::from_static("str");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user