1
0
mirror of https://github.com/fafhrd91/actix-net synced 2025-08-13 11:58:23 +02:00

Compare commits

...

25 Commits

Author SHA1 Message Date
Nikolay Kim
a2a9d9764d introduce IntoPattern trait 2019-12-25 19:54:20 +04:00
Nikolay Kim
bf0a9d2f6e Add IntoPatterns trait 2019-12-25 15:34:21 +04:00
Nikolay Kim
119027f822 fmt 2019-12-25 15:10:13 +04:00
Nikolay Kim
0fe8038d23 allow specify set of resource patters 2019-12-25 15:10:01 +04:00
Nikolay Kim
b599bc4a0c map_config() and unit_config() accepts IntoServiceFactory type 2019-12-22 16:30:49 +04:00
Nikolay Kim
a80e1f8370 fix new() method and make from_static and from_bytes_unchecked methods const 2019-12-22 16:24:28 +04:00
Nikolay Kim
5fe759cc02 Merge branch 'master' of github.com:actix/actix-net 2019-12-20 09:15:19 +06:00
Nikolay Kim
05549f0b42 Add methods to check LocalWaker registration state 2019-12-20 09:13:11 +06:00
Yuki Okushi
b1430eaded Run tests for all features as possible (#78) 2019-12-19 16:31:32 +09:00
Nikolay Kim
0d3f9e74c5 Use .advance() intead of .split_to() 2019-12-19 09:50:31 +06:00
Nikolay Kim
cab73791ed pin trsut-dns-proto 2019-12-15 13:04:26 +06:00
Nikolay Kim
a7ac1a76ed add license files to actix-macros 2019-12-14 23:01:55 +06:00
Nikolay Kim
37bedff6fb use parking_lot 0.10 2019-12-12 06:57:40 +06:00
Nikolay Kim
33fd6adc11 better InOrder test 2019-12-12 06:56:45 +06:00
Nikolay Kim
4305cdba2c Revert InOrder service changes 2019-12-11 23:10:02 +06:00
Nikolay Kim
52ecb4bcc5 Add oneshot::Sender::is_canceled() method 2019-12-11 20:52:57 +06:00
Nikolay Kim
b28f32e82c Allow to create framed::Dispatcher with custom mpsc::Receiver 2019-12-11 20:23:14 +06:00
Nikolay Kim
081205a02f Disconnect callback accepts owned state 2019-12-11 18:57:43 +06:00
Nikolay Kim
8bb81c0768 optimize InOrder service 2019-12-11 18:55:53 +06:00
Nikolay Kim
c7a8743bf9 remove E param 2019-12-11 16:44:09 +06:00
Nikolay Kim
f26fcc703b prep release 2019-12-11 14:56:05 +06:00
Nikolay Kim
ce4587df82 prepare actix-tls release 2019-12-11 14:53:58 +06:00
Nikolay Kim
9957f28137 prepare actix-testing release 2019-12-11 14:49:26 +06:00
Nikolay Kim
9d84d14ef4 update deps 2019-12-11 14:47:30 +06:00
Nikolay Kim
60bfa1bfb1 prepare actix-server release 2019-12-11 14:43:26 +06:00
48 changed files with 688 additions and 305 deletions

View File

@@ -2,6 +2,9 @@ name: CI
on: [push, pull_request]
env:
VCPKGRS_DYNAMIC: 1
jobs:
build_and_test:
strategy:
@@ -18,10 +21,12 @@ jobs:
include:
- toolchain: x86_64-pc-windows-msvc
os: windows-latest
arch: x64
- toolchain: x86_64-pc-windows-gnu
os: windows-latest
- toolchain: i686-pc-windows-msvc
os: windows-latest
arch: x86
- toolchain: x86_64-apple-darwin
os: macOS-latest
@@ -37,6 +42,13 @@ jobs:
toolchain: ${{ matrix.version }}-${{ matrix.toolchain }}
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
if: matrix.version == 'nightly'
uses: actions-rs/cargo@v1
@@ -52,6 +64,14 @@ jobs:
args: --all --bins --examples --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
with:
command: test

View File

@@ -1,5 +1,7 @@
# Changes
* Use `.advance()` intead of `.split_to()`
## [0.2.0] - 2019-12-10
* Use specific futures dependencies

View File

@@ -2,7 +2,7 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use std::{fmt, io};
use bytes::BytesMut;
use bytes::{Buf, BytesMut};
use futures_core::{ready, Stream};
use futures_sink::Sink;
@@ -288,7 +288,7 @@ impl<T, U> Framed<T, U> {
}
// remove written data
let _ = self.write_buf.split_to(n);
self.write_buf.advance(n);
}
// Try flushing the underlying IO

View File

@@ -1,5 +1,13 @@
# 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
### Changed

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-connect"
version = "1.0.0-alpha.3"
version = "1.0.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix connect - tcp connector service"
keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-connect/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
edition = "2018"
workspace = ".."
@@ -35,14 +34,15 @@ uri = ["http"]
[dependencies]
actix-service = "1.0.0"
actix-codec = "0.2.0"
actix-utils = "1.0.0-alpha.3"
actix-utils = "1.0.3"
actix-rt = "1.0.0"
derive_more = "0.99.2"
either = "1.5.2"
futures = "0.3.1"
http = { version = "0.2.0", optional = true }
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
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 }
[dev-dependencies]
bytes = "0.5.2"
actix-testing = { version="1.0.0-alpha.2" }
bytes = "0.5.3"
actix-testing = { version="1.0.0" }

View File

@@ -1,5 +1,13 @@
# 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
* Migrate to tokio 0.2

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-ioframe"
version = "0.3.0-alpha.3"
version = "0.4.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix framed service"
keywords = ["network", "framework", "async", "futures"]
@@ -9,7 +9,6 @@ repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-ioframe/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
edition = "2018"
workspace = ".."
@@ -20,7 +19,7 @@ path = "src/lib.rs"
[dependencies]
actix-service = "1.0.0"
actix-codec = "0.2.0"
actix-utils = "1.0.0-alpha.2"
actix-utils = "1.0.1"
actix-rt = "1.0.0"
bytes = "0.5"
either = "1.5.2"
@@ -29,5 +28,5 @@ pin-project = "0.4.6"
log = "0.4"
[dev-dependencies]
actix-connect = "1.0.0-alpha.2"
actix-testing = "1.0.0-alpha.2"
actix-connect = "1.0.0"
actix-testing = "1.0.0"

View File

@@ -7,21 +7,21 @@ use futures::Stream;
use crate::sink::Sink;
pub struct Connect<Io, Codec, Err, St = ()>
pub struct Connect<Io, Codec, St = ()>
where
Codec: Encoder + Decoder,
{
io: Io,
sink: Sink<<Codec as Encoder>::Item, Err>,
sink: Sink<<Codec as Encoder>::Item>,
_t: PhantomData<(St, Codec)>,
}
impl<Io, Codec, Err> Connect<Io, Codec, Err>
impl<Io, Codec> Connect<Io, Codec>
where
Io: AsyncRead + AsyncWrite,
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 {
io,
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 {
state: (),
sink: self.sink,
@@ -39,15 +39,15 @@ where
}
#[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) 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]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, Err> {
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink
}
@@ -62,7 +62,7 @@ impl<Io, St, Codec: Encoder + Decoder, Err> ConnectResult<Io, St, Codec, Err> {
}
#[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 {
state,
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
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,
@@ -83,8 +83,7 @@ where
}
}
impl<Io, St, Codec, Err> futures::Sink<<Codec as Encoder>::Item>
for ConnectResult<Io, St, Codec, Err>
impl<Io, St, Codec> futures::Sink<<Codec as Encoder>::Item> for ConnectResult<Io, St, Codec>
where
Io: AsyncRead + AsyncWrite,
Codec: Encoder + Decoder,

View File

@@ -13,7 +13,7 @@ use crate::error::ServiceError;
use crate::item::Item;
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;
pub(crate) enum Message<T> {
@@ -25,10 +25,10 @@ pub(crate) enum Message<T> {
/// FramedTransport - is a future that reads frames from Framed object
/// and pass then to the service.
#[pin_project::pin_project]
pub(crate) struct Dispatcher<St, S, T, U, E>
pub(crate) struct Dispatcher<St, S, T, U>
where
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::Future: 'static,
T: AsyncRead + AsyncWrite,
@@ -37,19 +37,19 @@ where
<U as Encoder>::Error: std::fmt::Debug,
{
service: S,
sink: Sink<<U as Encoder>::Item, E>,
sink: Sink<<U as Encoder>::Item>,
state: St,
dispatch_state: FramedState<S, U>,
framed: Framed<T, U>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, E>>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
tx: mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>>,
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
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::Future: 'static,
T: AsyncRead + AsyncWrite,
@@ -61,9 +61,9 @@ where
framed: Framed<T, U>,
state: St,
service: F,
sink: Sink<<U as Encoder>::Item, E>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, E>>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
sink: Sink<<U as Encoder>::Item>,
rx: mpsc::Receiver<Result<Message<<U as Encoder>::Item>, S::Error>>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
) -> Self {
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
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::Future: 'static,
T: AsyncRead + AsyncWrite,
@@ -245,7 +245,7 @@ where
}
}
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()))
}
@@ -265,19 +265,19 @@ where
let _ = tx.send(());
}
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(&mut self.state, false);
(&*disconnect)(self.state.clone(), false);
}
Poll::Ready(Ok(()))
}
FramedState::FramedError(_) => {
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()))
}
FramedState::Stopping => {
if let Some(ref disconnect) = self.disconnect {
(&*disconnect)(&mut self.state, false);
(&*disconnect)(self.state.clone(), false);
}
Poll::Ready(Ok(()))
}

View File

@@ -5,19 +5,19 @@ use actix_codec::{Decoder, Encoder};
use crate::sink::Sink;
pub struct Item<St, Codec: Encoder + Decoder, E> {
pub struct Item<St, Codec: Encoder + Decoder> {
state: St,
sink: Sink<<Codec as Encoder>::Item, E>,
sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item,
}
impl<St, Codec, E> Item<St, Codec, E>
impl<St, Codec> Item<St, Codec>
where
Codec: Encoder + Decoder,
{
pub(crate) fn new(
state: St,
sink: Sink<<Codec as Encoder>::Item, E>,
sink: Sink<<Codec as Encoder>::Item>,
item: <Codec as Decoder>::Item,
) -> Self {
Item { state, sink, item }
@@ -34,7 +34,7 @@ where
}
#[inline]
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item, E> {
pub fn sink(&self) -> &Sink<<Codec as Encoder>::Item> {
&self.sink
}
@@ -44,18 +44,12 @@ where
}
#[inline]
pub fn into_parts(
self,
) -> (
St,
Sink<<Codec as Encoder>::Item, E>,
<Codec as Decoder>::Item,
) {
pub fn into_parts(self) -> (St, Sink<<Codec as Encoder>::Item>, <Codec as Decoder>::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
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
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
Codec: Encoder + Decoder,
<Codec as Decoder>::Item: fmt::Debug,

View File

@@ -17,7 +17,7 @@ use crate::error::ServiceError;
use crate::item::Item;
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 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
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
F: IntoService<C>,
Io: AsyncRead + AsyncWrite,
C: Service<
Request = Connect<Io, Codec, E>,
Response = ConnectResult<Io, St, Codec, E>,
Error = E,
>,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Codec: Decoder + Encoder,
{
ServiceBuilder {
@@ -56,17 +52,16 @@ impl<St: Clone, Codec> Builder<St, Codec> {
}
/// 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
F: IntoServiceFactory<C>,
E: 'static,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec, E>,
Response = ConnectResult<Io, St, Codec, E>,
Error = E,
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
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,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_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
St: Clone,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
Io: AsyncRead + AsyncWrite,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
@@ -102,22 +93,22 @@ where
/// Second parameter indicates error occured during disconnect.
pub fn disconnect<F, Out>(mut self, disconnect: F) -> Self
where
F: Fn(&mut St, bool) + 'static,
F: Fn(St, bool) + 'static,
{
self.disconnect = Some(Rc::new(disconnect));
self
}
/// 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
F: IntoServiceFactory<T>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
{
FramedServiceImpl {
@@ -129,23 +120,22 @@ where
}
}
pub struct NewServiceBuilder<St, C, Io, Codec, Err> {
pub struct NewServiceBuilder<St, C, Io, Codec> {
connect: C,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_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
St: Clone,
Io: AsyncRead + AsyncWrite,
Err: 'static,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
@@ -156,21 +146,21 @@ where
/// Second parameter indicates error occured during disconnect.
pub fn disconnect<F>(mut self, disconnect: F) -> Self
where
F: Fn(&mut St, bool) + 'static,
F: Fn(St, bool) + 'static,
{
self.disconnect = Some(Rc::new(disconnect));
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
F: IntoServiceFactory<T>,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
> + 'static,
{
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,
handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err, Cfg)>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_t: PhantomData<(St, Io, Codec, Cfg)>,
}
impl<St, C, T, Io, Codec, Err, Cfg> ServiceFactory
for FramedService<St, C, T, Io, Codec, Err, Cfg>
impl<St, C, T, Io, Codec, Cfg> ServiceFactory for FramedService<St, C, T, Io, Codec, Cfg>
where
St: Clone + 'static,
Io: AsyncRead + AsyncWrite,
C: ServiceFactory<
Config = (),
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
Request = Connect<Io, Codec>,
Response = ConnectResult<Io, St, Codec>,
>,
C::Error: 'static,
C::Future: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
> + 'static,
<T::Service as Service>::Future: 'static,
Err: 'static,
Codec: Decoder + Encoder,
<Codec as Encoder>::Item: 'static,
<Codec as Encoder>::Error: std::fmt::Debug,
@@ -219,7 +207,7 @@ where
type Response = ();
type Error = ServiceError<C::Error, Codec>;
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>>;
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,
handler: Rc<T>,
disconnect: Option<Rc<dyn Fn(&mut St, bool)>>,
_t: PhantomData<(St, Io, Codec, Err)>,
disconnect: Option<Rc<dyn Fn(St, bool)>>,
_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
St: Clone,
Io: AsyncRead + AsyncWrite,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Future: 'static,
Codec: Decoder + Encoder,
@@ -272,8 +256,8 @@ where
{
type Request = Io;
type Response = ();
type Error = ServiceError<Err, Codec>;
type Future = FramedServiceImplResponse<St, Io, Codec, Err, C, T>;
type Error = ServiceError<C::Error, Codec>;
type Future = FramedServiceImplResponse<St, Io, Codec, C, T>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.connect.poll_ready(cx).map_err(|e| e.into())
@@ -281,7 +265,9 @@ where
fn call(&mut self, req: Io) -> Self::Future {
let (tx, rx) = mpsc::channel();
let sink = Sink::new(tx);
let sink = Sink::new(Rc::new(move |msg| {
let _ = tx.send(Ok(msg));
}));
FramedServiceImplResponse {
inner: FramedServiceImplResponseInner::Connect(
self.connect.call(Connect::new(req, sink.clone())),
@@ -294,21 +280,17 @@ where
}
#[pin_project::pin_project]
pub struct FramedServiceImplResponse<St, Io, Codec, Err, C, T>
pub struct FramedServiceImplResponse<St, Io, Codec, C, T>
where
St: Clone,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
@@ -317,24 +299,20 @@ where
<Codec as Encoder>::Error: std::fmt::Debug,
{
#[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
St: Clone,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
@@ -342,7 +320,7 @@ where
<Codec as Encoder>::Item: 'static,
<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> {
let mut this = self.as_mut().project();
@@ -360,21 +338,17 @@ where
}
#[pin_project::pin_project]
enum FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>
enum FramedServiceImplResponseInner<St, Io, Codec, C, T>
where
St: Clone,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
@@ -385,33 +359,29 @@ where
Connect(
#[pin] C::Future,
Rc<T>,
Option<Rc<dyn Fn(&mut St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
Option<Rc<dyn Fn(St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, C::Error>>>,
),
Handler(
#[pin] T::Future,
Option<ConnectResult<Io, St, Codec, Err>>,
Option<Rc<dyn Fn(&mut St, bool)>>,
Option<mpsc::Receiver<ServiceResult<Codec, Err>>>,
Option<ConnectResult<Io, St, Codec>>,
Option<Rc<dyn Fn(St, bool)>>,
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
St: Clone,
C: Service<
Request = Connect<Io, Codec, Err>,
Response = ConnectResult<Io, St, Codec, Err>,
Error = Err,
>,
Err: 'static,
C: Service<Request = Connect<Io, Codec>, Response = ConnectResult<Io, St, Codec>>,
C::Error: 'static,
T: ServiceFactory<
Config = St,
Request = RequestItem<St, Codec, Err>,
Request = RequestItem<St, Codec>,
Response = ResponseItem<Codec>,
Error = Err,
InitError = Err,
Error = C::Error,
InitError = C::Error,
>,
<T::Service as Service>::Future: 'static,
Io: AsyncRead + AsyncWrite,
@@ -424,8 +394,8 @@ where
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Either<
FramedServiceImplResponseInner<St, Io, Codec, Err, C, T>,
Poll<Result<(), ServiceError<Err, Codec>>>,
FramedServiceImplResponseInner<St, Io, Codec, C, T>,
Poll<Result<(), ServiceError<C::Error, Codec>>>,
> {
#[project]
match self.project() {

View File

@@ -1,43 +1,44 @@
use std::fmt;
use std::rc::Rc;
use actix_utils::{mpsc, oneshot};
use actix_utils::oneshot;
use futures::future::{Future, FutureExt};
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 {
Sink(self.0.clone())
}
}
impl<T, E> Sink<T, E> {
pub(crate) fn new(tx: mpsc::Sender<Result<Message<T>, E>>) -> Self {
impl<T> Sink<T> {
pub(crate) fn new(tx: Rc<dyn Fn(Message<T>)>) -> Self {
Sink(tx)
}
/// Close connection
pub fn close(&self) {
let _ = self.0.send(Ok(Message::Close));
(self.0)(Message::Close);
}
/// Close connection
pub fn wait_close(&self) -> impl Future<Output = ()> {
let (tx, rx) = oneshot::channel();
let _ = self.0.send(Ok(Message::WaitClose(tx)));
(self.0)(Message::WaitClose(tx));
rx.map(|_| ())
}
/// Send item
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 {
fmt.debug_struct("Sink").finish()
}

View File

@@ -22,7 +22,7 @@ async fn test_disconnect() -> std::io::Result<()> {
let disconnect1 = disconnect1.clone();
Builder::new()
.factory(fn_service(|conn: Connect<_, _, _>| {
.factory(fn_service(|conn: Connect<_, _>| {
ok(conn.codec(BytesCodec).state(State))
}))
.disconnect(move |_, _| {
@@ -32,7 +32,7 @@ async fn test_disconnect() -> std::io::Result<()> {
});
let mut client = Builder::new()
.service(|conn: Connect<_, _, _>| {
.service(|conn: Connect<_, _>| {
let conn = conn.codec(BytesCodec).state(State);
conn.sink().close();
ok(conn)

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-macros"
version = "0.1.0"
version = "0.1.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix runtime macros"
repository = "https://github.com/actix/actix-net"

1
actix-macros/LICENSE-APACHE Symbolic link
View File

@@ -0,0 +1 @@
../LICENSE-APACHE

1
actix-macros/LICENSE-MIT Symbolic link
View File

@@ -0,0 +1 @@
../LICENSE-MIT

View File

@@ -71,7 +71,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
if input.sig.asyncness.is_none() {
return syn::Error::new_spanned(
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()
.into();

View File

@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-rt/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
edition = "2018"
workspace = ".."
[lib]
name = "actix_rt"
@@ -21,4 +20,4 @@ actix-macros = "0.1.0"
actix-threadpool = "0.3"
futures = "0.3.1"
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"] }

View File

@@ -1,5 +1,12 @@
# Changes
## [1.0.0] - 2019-12-11
### Changed
* Use actix-net releases
## [1.0.0-alpha.4] - 2019-12-08
### Changed

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-server"
version = "1.0.0-alpha.4"
version = "1.0.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix server - General purpose tcp server"
keywords = ["network", "framework", "async", "futures"]
@@ -24,10 +24,10 @@ default = []
actix-service = "1.0.0"
actix-rt = "1.0.0"
actix-codec = "0.2.0"
actix-utils = "1.0.0-alpha.3"
actix-utils = "1.0.0"
log = "0.4"
num_cpus = "1.0"
num_cpus = "1.11"
mio = "0.6.19"
net2 = "0.2"
futures = "0.3.1"
@@ -39,4 +39,4 @@ mio-uds = { version = "0.6.7" }
[dev-dependencies]
bytes = "0.5"
env_logger = "0.6"
actix-testing = "1.0.0-alpha.3"
actix-testing = "1.0.0"

View File

@@ -391,7 +391,7 @@ impl ServerBuilder {
.await;
System::current().stop();
}
.boxed(),
.boxed(),
);
}
ready(())

View File

@@ -153,7 +153,7 @@ impl InternalServiceFactory for ConfiguredService {
}
return Ok(res);
}
.boxed_local()
.boxed_local()
}
}
@@ -280,6 +280,6 @@ where
}
};
}
.boxed_local()
.boxed_local()
}
}

View File

@@ -217,7 +217,7 @@ impl Worker {
wrk.await
});
}
.boxed(),
.boxed(),
);
WorkerClient::new(idx, tx1, tx2, avail)

View File

@@ -1,5 +1,12 @@
# Changes
## [1.0.1] - 2019-12-22
### Changed
* `map_config()` and `unit_config()` accepts `IntoServiceFactory` type
## [1.0.0] - 2019-12-11
### Added

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-service"
version = "1.0.0"
version = "1.0.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix service"
keywords = ["network", "framework", "async", "futures"]

View File

@@ -1,28 +1,30 @@
use std::marker::PhantomData;
use super::ServiceFactory;
use super::{IntoServiceFactory, ServiceFactory};
/// Adapt external config argument to a config for provided service factory
///
/// Note that this function consumes the receiving service factory and returns
/// a wrapped version of it.
pub fn map_config<T, F, C>(factory: T, f: F) -> MapConfig<T, F, C>
pub fn map_config<T, U, F, C>(factory: U, f: F) -> MapConfig<T, F, C>
where
T: ServiceFactory,
U: IntoServiceFactory<T>,
F: Fn(C) -> T::Config,
{
MapConfig::new(factory, f)
MapConfig::new(factory.into_factory(), f)
}
/// Replace config with unit
pub fn unit_config<T, C>(new_service: T) -> UnitConfig<T, C>
pub fn unit_config<T, U, C>(factory: U) -> UnitConfig<T, C>
where
T: ServiceFactory<Config = ()>,
U: IntoServiceFactory<T>,
{
UnitConfig::new(new_service)
UnitConfig::new(factory.into_factory())
}
/// `.map_config()` service combinator
/// `map_config()` adapter service factory
pub struct MapConfig<A, F, C> {
a: A,
f: F,

View File

@@ -1,5 +1,9 @@
# Changes
## [1.0.0] - 2019-12-11
* Update actix-server to 1.0.0
## [1.0.0-alpha.3] - 2019-12-07
* Migrate to tokio 0.2

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-testing"
version = "1.0.0-alpha.3"
version = "1.0.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix testing utils"
keywords = ["network", "framework", "async", "futures"]
@@ -19,7 +19,7 @@ path = "src/lib.rs"
[dependencies]
actix-rt = "1.0.0"
actix-macros = "0.1.0"
actix-server = "1.0.0-alpha.3"
actix-server = "1.0.0"
actix-service = "1.0.0"
log = "0.4"

View File

@@ -1,5 +1,11 @@
# Changes
## [0.3.1] - 2019-12-12
### Changed
* Use parking_lot 0.10
## [0.3.0] - 2019-12-02
### Changed

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-threadpool"
version = "0.3.0"
version = "0.3.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix thread pool for sync code"
keywords = ["actix", "network", "framework", "async", "futures"]
@@ -19,9 +19,9 @@ path = "src/lib.rs"
[dependencies]
derive_more = "0.99.2"
futures = "0.3.1"
parking_lot = "0.9"
lazy_static = "1.2"
futures-channel = "0.3.1"
parking_lot = "0.10"
lazy_static = "1.3"
log = "0.4"
num_cpus = "1.10"
threadpool = "1.7"

View File

@@ -6,7 +6,7 @@ use std::pin::Pin;
use std::task::{Context, Poll};
use derive_more::Display;
use futures::channel::oneshot;
use futures_channel::oneshot;
use parking_lot::Mutex;
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> {
let rx = Pin::new(&mut self.rx);
let res = futures::ready!(rx.poll(cx))
.map_err(|_| BlockingError::Canceled)
.and_then(|res| res.map_err(BlockingError::Error));
let res = match rx.poll(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(res) => res
.map_err(|_| BlockingError::Canceled)
.and_then(|res| res.map_err(BlockingError::Error)),
};
Poll::Ready(res)
}
}

View File

@@ -1,6 +1,10 @@
# 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

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-tls"
version = "1.0.0-alpha.3"
version = "1.0.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix tls services"
keywords = ["network", "framework", "async", "futures"]
@@ -34,7 +34,7 @@ nativetls = ["native-tls", "tokio-tls"]
[dependencies]
actix-service = "1.0.0"
actix-codec = "0.2.0"
actix-utils = "1.0.0-alpha.3"
actix-utils = "1.0.0"
actix-rt = "1.0.0"
derive_more = "0.99.2"
either = "1.5.2"
@@ -57,4 +57,4 @@ tokio-tls = { version="0.3", optional = true }
[dev-dependencies]
bytes = "0.5"
actix-testing = { version="1.0.0-alpha.3" }
actix-testing = { version="1.0.0" }

View File

@@ -88,7 +88,7 @@ impl<T: AsyncRead + AsyncWrite + Unpin + 'static> Service for AcceptorService<T>
let acc = acc;
tokio_openssl::accept(&acc, req).await
}
.boxed_local(),
.boxed_local(),
}
}
}

View File

@@ -1,5 +1,23 @@
# 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
* Simplify oneshot and mpsc implementations

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-utils"
version = "1.0.0"
version = "1.0.4"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix utils - various actix net related services"
keywords = ["network", "framework", "async", "futures"]
@@ -10,7 +10,6 @@ documentation = "https://docs.rs/actix-utils/"
categories = ["network-programming", "asynchronous"]
license = "MIT/Apache-2.0"
edition = "2018"
workspace = ".."
[lib]
name = "actix_utils"
@@ -20,8 +19,8 @@ path = "src/lib.rs"
actix-service = "1.0.0"
actix-rt = "1.0.0"
actix-codec = "0.2.0"
bytes = "0.5.2"
either = "1.5.2"
bytes = "0.5.3"
either = "1.5.3"
futures = "0.3.1"
pin-project = "0.4.6"
log = "0.4"

View File

@@ -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
pub fn get_sink(&self) -> mpsc::Sender<Result<Message<<U as Encoder>::Item>, S::Error>> {
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 {
&self.service
}
/// Get mutable reference to a service wrapped by `FramedTransport`
/// instance.
/// Get mutable reference to a service wrapped by `Dispatcher` instance.
pub fn get_mut(&mut self) -> &mut S {
&mut self.service
}
/// Get reference to a framed instance wrapped by `FramedTransport`
/// Get reference to a framed instance wrapped by `Dispatcher`
/// instance.
pub fn get_framed(&self) -> &Framed<T, U> {
&self.framed
}
/// Get mutable reference to a framed instance wrapped by `FramedTransport`
/// instance.
/// Get mutable reference to a framed instance wrapped by `Dispatcher` instance.
pub fn get_framed_mut(&mut self) -> &mut Framed<T, U> {
&mut self.framed
}

View File

@@ -67,6 +67,12 @@ impl<T> Sender<T> {
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> {
@@ -110,7 +116,9 @@ mod tests {
assert_eq!(rx.await.unwrap(), "test");
let (tx, rx) = channel();
assert!(!tx.is_canceled());
drop(rx);
assert!(tx.is_canceled());
assert!(tx.send("test").is_err());
let (tx, rx) = channel::<&'static str>();

View File

@@ -105,7 +105,7 @@ where
pub struct InOrderService<S: Service> {
service: S,
task: Rc<LocalWaker>,
waker: Rc<LocalWaker>,
acks: VecDeque<Record<S::Response, S::Error>>,
}
@@ -123,7 +123,7 @@ where
Self {
service: service.into_service(),
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>> {
// poll_ready could be called from different task
self.task.register(cx.waker());
self.waker.register(cx.waker());
// check acks
while !self.acks.is_empty() {
@@ -172,11 +172,11 @@ where
let (tx2, rx2) = oneshot::channel();
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);
actix_rt::spawn(async move {
let res = fut.await;
task.wake();
waker.wake();
let _ = tx1.send(res);
});
@@ -211,7 +211,7 @@ mod tests {
use super::*;
use actix_service::Service;
use futures::channel::oneshot;
use futures::future::{lazy, FutureExt, LocalBoxFuture};
use futures::future::{lazy, poll_fn, FutureExt, LocalBoxFuture};
struct Srv;
@@ -245,15 +245,18 @@ mod tests {
let _ = actix_rt::System::new("test").block_on(async {
let mut srv = InOrderService::new(Srv);
let _ = lazy(|cx| srv.poll_ready(cx)).await;
let res1 = srv.call(rx1);
let res2 = srv.call(rx2);
let res3 = srv.call(rx3);
let _ = lazy(|cx| srv.poll_ready(cx)).await;
// dispatcher do this
actix_rt::time::delay_for(Duration::from_millis(100)).await;
let _ = lazy(|cx| srv.poll_ready(cx)).await;
actix_rt::spawn(async move {
let _ = poll_fn(|cx| {
let _ = srv.poll_ready(cx);
Poll::<()>::Pending
})
.await;
});
assert_eq!(res1.await.unwrap(), 1);
assert_eq!(res2.await.unwrap(), 2);

View File

@@ -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]
/// 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 {
*self.waker.get() = Some(waker.clone());
let w = self.waker.get();
let is_registered = (*w).is_some();
*w = Some(waker.clone());
is_registered
}
}

View File

@@ -1,5 +1,11 @@
# Changes
## [0.2.1] - 2019-12-25
* Add `IntoPattern` trait
* Add multi-pattern resources
## [0.2.0] - 2019-12-07
* Update http to 0.2

View File

@@ -1,6 +1,6 @@
[package]
name = "actix-router"
version = "0.2.0"
version = "0.2.1"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Path router"
keywords = ["actix"]
@@ -8,9 +8,7 @@ homepage = "https://actix.rs"
repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/actix-router/"
license = "MIT/Apache-2.0"
exclude = [".gitignore", ".travis.yml", ".cargo/config", "appveyor.yml"]
edition = "2018"
workspace = ".."
[lib]
name = "actix_router"
@@ -21,8 +19,8 @@ default = ["http"]
[dependencies]
regex = "1.3.1"
serde = "1.0.80"
bytestring = "0.1.0"
serde = "1.0.104"
bytestring = "0.1.2"
log = "0.4.8"
http = { version="0.2.0", optional=true }

View File

@@ -492,7 +492,7 @@ mod tests {
#[derive(Deserialize)]
struct Id {
id: String,
_id: String,
}
#[derive(Debug, Deserialize)]

View File

@@ -35,6 +35,90 @@ impl ResourcePath for bytestring::ByteString {
}
}
/// Helper trait for type that could be converted to path pattern
pub trait IntoPattern {
/// Signle patter
fn is_single(&self) -> bool;
fn patterns(&self) -> Vec<String>;
}
impl IntoPattern for String {
fn is_single(&self) -> bool {
true
}
fn patterns(&self) -> Vec<String> {
vec![self.clone()]
}
}
impl<'a> IntoPattern for &'a str {
fn is_single(&self) -> bool {
true
}
fn patterns(&self) -> Vec<String> {
vec![self.to_string()]
}
}
impl<T: AsRef<str>> IntoPattern for Vec<T> {
fn is_single(&self) -> bool {
self.len() == 1
}
fn patterns(&self) -> Vec<String> {
self.into_iter().map(|v| v.as_ref().to_string()).collect()
}
}
macro_rules! array_patterns (($tp:ty, $num:tt) => {
impl IntoPattern for [$tp; $num] {
fn is_single(&self) -> bool {
$num == 1
}
fn patterns(&self) -> Vec<String> {
self.iter().map(|v| v.to_string()).collect()
}
}
});
array_patterns!(&str, 1);
array_patterns!(&str, 2);
array_patterns!(&str, 3);
array_patterns!(&str, 4);
array_patterns!(&str, 5);
array_patterns!(&str, 6);
array_patterns!(&str, 7);
array_patterns!(&str, 8);
array_patterns!(&str, 9);
array_patterns!(&str, 10);
array_patterns!(&str, 11);
array_patterns!(&str, 12);
array_patterns!(&str, 13);
array_patterns!(&str, 14);
array_patterns!(&str, 15);
array_patterns!(&str, 16);
array_patterns!(String, 1);
array_patterns!(String, 2);
array_patterns!(String, 3);
array_patterns!(String, 4);
array_patterns!(String, 5);
array_patterns!(String, 6);
array_patterns!(String, 7);
array_patterns!(String, 8);
array_patterns!(String, 9);
array_patterns!(String, 10);
array_patterns!(String, 11);
array_patterns!(String, 12);
array_patterns!(String, 13);
array_patterns!(String, 14);
array_patterns!(String, 15);
array_patterns!(String, 16);
#[cfg(feature = "http")]
mod url;

View File

@@ -2,10 +2,10 @@ use std::cmp::min;
use std::hash::{Hash, Hasher};
use std::rc::Rc;
use regex::{escape, Regex};
use regex::{escape, Regex, RegexSet};
use crate::path::{Path, PathItem};
use crate::{Resource, ResourcePath};
use crate::{IntoPattern, Resource, ResourcePath};
const MAX_DYNAMIC_SEGMENTS: usize = 16;
@@ -32,21 +32,53 @@ enum PatternType {
Static(String),
Prefix(String),
Dynamic(Regex, Vec<Rc<String>>, usize),
DynamicSet(RegexSet, Vec<(Regex, Vec<Rc<String>>, usize)>),
}
impl ResourceDef {
/// Parse path pattern and create new `Pattern` instance.
///
/// Panics if path pattern is wrong.
pub fn new(path: &str) -> Self {
ResourceDef::with_prefix(path, false)
/// Panics if path pattern is malformed.
pub fn new<T: IntoPattern>(path: T) -> Self {
if path.is_single() {
let patterns = path.patterns();
ResourceDef::with_prefix(&patterns[0], false)
} else {
let set = path.patterns();
let mut data = Vec::new();
let mut re_set = Vec::new();
for path in set {
let (pattern, _, _, len) = ResourceDef::parse(&path, false);
let re = match Regex::new(&pattern) {
Ok(re) => re,
Err(err) => panic!("Wrong path pattern: \"{}\" {}", path, err),
};
// actix creates one router per thread
let names: Vec<_> = re
.capture_names()
.filter_map(|name| name.map(|name| Rc::new(name.to_owned())))
.collect();
data.push((re, names, len));
re_set.push(pattern);
}
ResourceDef {
id: 0,
tp: PatternType::DynamicSet(RegexSet::new(re_set).unwrap(), data),
elements: Vec::new(),
name: String::new(),
pattern: "".to_owned(),
}
}
}
/// Parse path pattern and create new `Pattern` instance.
///
/// Use `prefix` type instead of `static`.
///
/// Panics if path regex pattern is wrong.
/// Panics if path regex pattern is malformed.
pub fn prefix(path: &str) -> Self {
ResourceDef::with_prefix(path, true)
}
@@ -57,7 +89,7 @@ impl ResourceDef {
///
/// Use `prefix` type instead of `static`.
///
/// Panics if path regex pattern is wrong.
/// Panics if path regex pattern is malformed.
pub fn root_prefix(path: &str) -> Self {
ResourceDef::with_prefix(&insert_slash(path), true)
}
@@ -123,8 +155,9 @@ impl ResourceDef {
pub fn is_match(&self, path: &str) -> bool {
match self.tp {
PatternType::Static(ref s) => s == path,
PatternType::Dynamic(ref re, _, _) => re.is_match(path),
PatternType::Prefix(ref s) => path.starts_with(s),
PatternType::Dynamic(ref re, _, _) => re.is_match(path),
PatternType::DynamicSet(ref re, _) => re.is_match(path),
}
}
@@ -176,6 +209,30 @@ impl ResourceDef {
};
Some(min(plen, len))
}
PatternType::DynamicSet(ref re, ref params) => {
if let Some(idx) = re.matches(path).into_iter().next() {
let (ref pattern, _, len) = params[idx];
if let Some(captures) = pattern.captures(path) {
let mut pos = 0;
let mut passed = false;
for capture in captures.iter() {
if let Some(ref m) = capture {
if !passed {
passed = true;
continue;
}
pos = m.end();
}
}
Some(pos + len)
} else {
None
}
} else {
None
}
}
}
}
@@ -190,6 +247,25 @@ impl ResourceDef {
false
}
}
PatternType::Prefix(ref s) => {
let rpath = path.path();
let len = if s == rpath {
s.len()
} else if rpath.starts_with(s)
&& (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
{
if s.ends_with('/') {
s.len() - 1
} else {
s.len()
}
} else {
return false;
};
let rpath_len = rpath.len();
path.skip(min(rpath_len, len) as u16);
true
}
PatternType::Dynamic(ref re, ref names, len) => {
let mut idx = 0;
let mut pos = 0;
@@ -219,24 +295,40 @@ impl ResourceDef {
path.skip((pos + len) as u16);
true
}
PatternType::Prefix(ref s) => {
let rpath = path.path();
let len = if s == rpath {
s.len()
} else if rpath.starts_with(s)
&& (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
{
if s.ends_with('/') {
s.len() - 1
PatternType::DynamicSet(ref re, ref params) => {
if let Some(idx) = re.matches(path.path()).into_iter().next() {
let (ref pattern, ref names, len) = params[idx];
let mut idx = 0;
let mut pos = 0;
let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
[PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
if let Some(captures) = pattern.captures(path.path()) {
for (no, name) in names.iter().enumerate() {
if let Some(m) = captures.name(&name) {
idx += 1;
pos = m.end();
segments[no] =
PathItem::Segment(m.start() as u16, m.end() as u16);
} else {
log::error!(
"Dynamic path match but not all segments found: {}",
name
);
return false;
}
}
} else {
s.len()
return false;
}
for idx in 0..idx {
path.add(names[idx].clone(), segments[idx]);
}
path.skip((pos + len) as u16);
true
} else {
return false;
};
let rpath_len = rpath.len();
path.skip(min(rpath_len, len) as u16);
true
false
}
}
}
}
@@ -263,6 +355,30 @@ impl ResourceDef {
false
}
}
PatternType::Prefix(ref s) => {
let len = {
let rpath = res.resource_path().path();
if s == rpath {
s.len()
} else if rpath.starts_with(s)
&& (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
{
if s.ends_with('/') {
s.len() - 1
} else {
s.len()
}
} else {
return false;
}
};
if !check(res, user_data) {
return false;
}
let path = res.resource_path();
path.skip(min(path.path().len(), len) as u16);
true
}
PatternType::Dynamic(ref re, ref names, len) => {
let mut idx = 0;
let mut pos = 0;
@@ -298,29 +414,47 @@ impl ResourceDef {
path.skip((pos + len) as u16);
true
}
PatternType::Prefix(ref s) => {
let len = {
let rpath = res.resource_path().path();
if s == rpath {
s.len()
} else if rpath.starts_with(s)
&& (s.ends_with('/') || rpath.split_at(s.len()).1.starts_with('/'))
{
if s.ends_with('/') {
s.len() - 1
} else {
s.len()
PatternType::DynamicSet(ref re, ref params) => {
let path = res.resource_path().path();
if let Some(idx) = re.matches(path).into_iter().next() {
let (ref pattern, ref names, len) = params[idx];
let mut idx = 0;
let mut pos = 0;
let mut segments: [PathItem; MAX_DYNAMIC_SEGMENTS] =
[PathItem::Static(""); MAX_DYNAMIC_SEGMENTS];
if let Some(captures) = pattern.captures(path) {
for (no, name) in names.iter().enumerate() {
if let Some(m) = captures.name(&name) {
idx += 1;
pos = m.end();
segments[no] =
PathItem::Segment(m.start() as u16, m.end() as u16);
} else {
log::error!(
"Dynamic path match but not all segments found: {}",
name
);
return false;
}
}
} else {
return false;
}
};
if !check(res, user_data) {
return false;
if !check(res, user_data) {
return false;
}
let path = res.resource_path();
for idx in 0..idx {
path.add(names[idx].clone(), segments[idx]);
}
path.skip((pos + len) as u16);
true
} else {
false
}
let path = res.resource_path();
path.skip(min(path.path().len(), len) as u16);
true
}
}
}
@@ -348,7 +482,10 @@ impl ResourceDef {
}
}
}
};
PatternType::DynamicSet(..) => {
return false;
}
}
true
}
@@ -476,7 +613,7 @@ impl<'a> From<&'a str> for ResourceDef {
impl From<String> for ResourceDef {
fn from(path: String) -> ResourceDef {
ResourceDef::new(&path)
ResourceDef::new(path)
}
}
@@ -558,6 +695,65 @@ mod tests {
assert_eq!(path.get("id").unwrap(), "012345");
}
#[test]
fn test_dynamic_set() {
let re = ResourceDef::new(vec![
"/user/{id}",
"/v{version}/resource/{id}",
"/{id:[[:digit:]]{6}}",
]);
assert!(re.is_match("/user/profile"));
assert!(re.is_match("/user/2345"));
assert!(!re.is_match("/user/2345/"));
assert!(!re.is_match("/user/2345/sdg"));
let mut path = Path::new("/user/profile");
assert!(re.match_path(&mut path));
assert_eq!(path.get("id").unwrap(), "profile");
let mut path = Path::new("/user/1245125");
assert!(re.match_path(&mut path));
assert_eq!(path.get("id").unwrap(), "1245125");
assert!(re.is_match("/v1/resource/320120"));
assert!(!re.is_match("/v/resource/1"));
assert!(!re.is_match("/resource"));
let mut path = Path::new("/v151/resource/adahg32");
assert!(re.match_path(&mut path));
assert_eq!(path.get("version").unwrap(), "151");
assert_eq!(path.get("id").unwrap(), "adahg32");
assert!(re.is_match("/012345"));
assert!(!re.is_match("/012"));
assert!(!re.is_match("/01234567"));
assert!(!re.is_match("/XXXXXX"));
let mut path = Path::new("/012345");
assert!(re.match_path(&mut path));
assert_eq!(path.get("id").unwrap(), "012345");
let re = ResourceDef::new([
"/user/{id}",
"/v{version}/resource/{id}",
"/{id:[[:digit:]]{6}}",
]);
assert!(re.is_match("/user/profile"));
assert!(re.is_match("/user/2345"));
assert!(!re.is_match("/user/2345/"));
assert!(!re.is_match("/user/2345/sdg"));
let re = ResourceDef::new([
"/user/{id}".to_string(),
"/v{version}/resource/{id}".to_string(),
"/{id:[[:digit:]]{6}}".to_string(),
]);
assert!(re.is_match("/user/profile"));
assert!(re.is_match("/user/2345"));
assert!(!re.is_match("/user/2345/"));
assert!(!re.is_match("/user/2345/sdg"));
}
#[test]
fn test_parse_tail() {
let re = ResourceDef::new("/user/-{id}*");

View File

@@ -1,9 +1,15 @@
# 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
[0.1.0] - 2019-12-07
## [0.1.0] - 2019-12-07
* Initial release

View File

@@ -1,6 +1,6 @@
[package]
name = "bytestring"
version = "0.1.1"
version = "0.1.2"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "A UTF-8 encoded string with Bytes as a storage"
keywords = ["actix"]
@@ -9,11 +9,10 @@ repository = "https://github.com/actix/actix-net.git"
documentation = "https://docs.rs/bytestring/"
license = "MIT/Apache-2.0"
edition = "2018"
workspace = ".."
[lib]
name = "bytestring"
path = "src/lib.rs"
[dependencies]
bytes = "0.5.2"
bytes = "0.5.3"

View File

@@ -6,14 +6,14 @@ use bytes::Bytes;
/// 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)]
pub struct ByteString(Bytes);
impl ByteString {
/// Creates a new `ByteString`.
pub fn new() -> String {
String::default()
pub fn new() -> Self {
ByteString(Bytes::new())
}
/// Get a reference to the underlying bytes object.
@@ -27,12 +27,12 @@ impl ByteString {
}
/// Creates a new `ByteString` from a static str.
pub fn from_static(src: &'static str) -> ByteString {
Self(Bytes::from_static(src.as_ref()))
pub const fn from_static(src: &'static str) -> ByteString {
Self(Bytes::from_static(src.as_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)
}
}
@@ -147,6 +147,11 @@ mod test {
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
#[test]
fn test_new() {
let _: ByteString = ByteString::new();
}
#[test]
fn test_hash() {
let mut hasher1 = DefaultHasher::default();
@@ -171,6 +176,7 @@ mod test {
#[test]
fn test_from_static_str() {
const _S: ByteString = ByteString::from_static("hello");
let _ = ByteString::from_static("str");
}