cleanups

.github/workflows/linux.yml

@@ -48,13 +48,20 @@ jobs:
  path: target
  key: ${{ matrix.version }}-x86_64-unknown-linux-gnu-cargo-build-trimmed-${{ hashFiles('**/Cargo.lock') }}
 
- - name: Run tests
+ - name: Run tests [tokio]
  uses: actions-rs/cargo@v1
  timeout-minutes: 40
  with:
  command: test
  args: --all --features=ntex/tokio -- --nocapture
 
+ - name: Run tests [compio]
+ uses: actions-rs/cargo@v1
+ timeout-minutes: 40
+ with:
+ command: test
+ args: --all --features=ntex/compio -- --nocapture
+
  - name: Install cargo-cache
  continue-on-error: true
  run: |

.github/workflows/windows.yml

@@ -69,4 +69,4 @@ jobs:
  uses: actions-rs/cargo@v1
  with:
  command: test
- args: --all --features=ntex/tokio -- --nocapture
+ args: --all --features=ntex/compio -- --nocapture

Cargo.toml

@@ -36,4 +36,4 @@ ntex-tls = "2"
 ntex-macros = "0.1"
 openssl = "0.10"
 test-case = "3.2"
-ntex = { version = "2", features = ["tokio", "openssl"] }
+ntex = { version = "2", features = ["openssl"] }

src/inflight.rs

@@ -33,7 +33,7 @@ impl InFlightService {
 
  /// Number of inbound in-flight concurrent messages.
  ///
- /// By default inbound is set to 16 messages
+ /// By default max receive number is set to 16 messages
  pub fn max_receive(mut self, val: u16) -> Self {
  self.max_receive = val;
  self
@@ -65,28 +65,32 @@ pub struct InFlightServiceImpl<S> {
  service: S,
 }
 
-impl<T, R> Service<R> for InFlightServiceImpl<T>
+impl<S, R> Service<R> for InFlightServiceImpl<S>
 where
- T: Service<R>,
+ S: Service<R>,
  R: SizedRequest + 'static,
 {
- type Response = T::Response;
- type Error = T::Error;
+ type Response = S::Response;
+ type Error = S::Error;
 
  ntex_service::forward_shutdown!(service);
 
  #[inline]
- async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), Self::Error> {
+ async fn ready(&self, ctx: ServiceCtx<'_, Self>) -> Result<(), S::Error> {
  ctx.ready(&self.service).await?;
+
+ // check if we have capacity
  self.count.available().await;
  Ok(())
  }
 
  #[inline]
- async fn call(&self, req: R, ctx: ServiceCtx<'_, Self>) -> Result<T::Response, T::Error> {
+ async fn call(&self, req: R, ctx: ServiceCtx<'_, Self>) -> Result<S::Response, S::Error> {
  let size = if self.count.0.max_size > 0 { req.size() } else { 0 };
- let _task_guard = self.count.get(size);
- ctx.call(&self.service, req).await
+ let task_guard = self.count.get(size);
+ let result = ctx.call(&self.service, req).await;
+ drop(task_guard);
+ result
  }
 }
 

src/io.rs

@@ -789,6 +789,7 @@ mod tests {
  assert_eq!(buf, Bytes::from_static(b"GET /test HTTP/1\r\n\r\n"));
 
  // write side must be closed, dispatcher waiting for read side to close
+ sleep(Millis(50)).await;
  assert!(client.is_closed());
 
  // close read side
@@ -837,6 +838,7 @@ mod tests {
  assert_eq!(buf, Bytes::from_static(b"GET /test HTTP/1\r\n\r\n"));
 
  // write side must be closed, dispatcher waiting for read side to close
+ sleep(Millis(50)).await;
  assert!(client.is_closed());
 
  // close read side