Repository: dmolokanov/appinsights-rs (language: rust)

Issue:
Using appinsights in an async context
Trying to start appinsights in an async context gives an error:

<details>
<summary>Cargo.toml</summary>

```
[package]
name = "foo"
version = "0.1.0"
authors = ["Trangar <victor.koenders@gmail.com>"]
edition = "2018"

[dependencies]
appinsights = "0.1"
tokio = { version = "0.2", features = ["full"] }
```

</details>

<details>
<summary>src/main.rs</summary>

```rust
use appinsights::TelemetryClient;

#[tokio::main]
async fn main() {
    let appinsight_key = "...";
    let client = TelemetryClient::new(appinsight_key.to_string());
}
```

</details>

<details>
<summary>Output</summary>

```
thread 'main' panicked at 'Cannot start a runtime from within a runtime. This happens because a function (like `block_on`) attempted to block the current thread while the thread is being used to drive asynchronous tasks.', /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/enter.rs:38:5
stack backtrace:
   0: backtrace::backtrace::libunwind::trace
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.46/src/backtrace/libunwind.rs:86
   1: backtrace::backtrace::trace_unsynchronized
             at /cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.46/src/backtrace/mod.rs:66
   2: std::sys_common::backtrace::_print_fmt
             at src/libstd/sys_common/backtrace.rs:78
   3: <std::sys_common::backtrace::_print::DisplayBacktrace as core::fmt::Display>::fmt
             at src/libstd/sys_common/backtrace.rs:59
   4: core::fmt::write
             at src/libcore/fmt/mod.rs:1076
   5: std::io::Write::write_fmt
             at src/libstd/io/mod.rs:1537
   6: std::sys_common::backtrace::_print
             at src/libstd/sys_common/backtrace.rs:62
   7: std::sys_common::backtrace::print
             at src/libstd/sys_common/backtrace.rs:49
   8: std::panicking::default_hook::{{closure}}
             at src/libstd/panicking.rs:198
   9: std::panicking::default_hook
             at src/libstd/panicking.rs:218
  10: std::panicking::rust_panic_with_hook
             at src/libstd/panicking.rs:477
  11: std::panicking::begin_panic
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libstd/panicking.rs:404
  12: tokio::runtime::enter::enter
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/enter.rs:38
  13: tokio::runtime::blocking::shutdown::Receiver::wait
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/blocking/shutdown.rs:44
  14: tokio::runtime::blocking::pool::BlockingPool::shutdown
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/blocking/pool.rs:132
  15: <tokio::runtime::blocking::pool::BlockingPool as core::ops::drop::Drop>::drop
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/blocking/pool.rs:138
  16: core::ptr::drop_in_place
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libcore/ptr/mod.rs:184
  17: core::ptr::drop_in_place
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libcore/ptr/mod.rs:184
  18: reqwest::blocking::wait::enter
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.10.6/src/blocking/wait.rs:74
  19: reqwest::blocking::wait::timeout
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.10.6/src/blocking/wait.rs:14
  20: reqwest::blocking::client::ClientHandle::new
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.10.6/src/blocking/client.rs:758
  21: reqwest::blocking::client::ClientBuilder::build
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.10.6/src/blocking/client.rs:95
  22: reqwest::blocking::client::Client::new
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/reqwest-0.10.6/src/blocking/client.rs:565
  23: appinsights::transmitter::Transmitter::new
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/appinsights-0.1.4/src/transmitter.rs:27
  24: appinsights::channel::memory::InMemoryChannel::new
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/appinsights-0.1.4/src/channel/memory.rs:28
  25: appinsights::client::TelemetryClient<appinsights::channel::memory::InMemoryChannel>::from_config
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/appinsights-0.1.4/src/client/mod.rs:31
  26: appinsights::client::TelemetryClient<appinsights::channel::memory::InMemoryChannel>::new
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/appinsights-0.1.4/src/client/mod.rs:21
  27: foo::main::{{closure}}
             at src/main.rs:8
  28: <core::future::from_generator::GenFuture<T> as core::future::future::Future>::poll
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libcore/future/mod.rs:73
  29: tokio::runtime::enter::Enter::block_on::{{closure}}
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/enter.rs:163
  30: tokio::coop::with_budget::{{closure}}
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/coop.rs:127
  31: std::thread::local::LocalKey<T>::try_with
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libstd/thread/local.rs:263
  32: std::thread::local::LocalKey<T>::with
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libstd/thread/local.rs:239
  33: tokio::coop::with_budget
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/coop.rs:120
  34: tokio::coop::budget
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/coop.rs:96
  35: tokio::runtime::enter::Enter::block_on
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/enter.rs:163
  36: tokio::runtime::thread_pool::ThreadPool::block_on
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/thread_pool/mod.rs:82
  37: tokio::runtime::Runtime::block_on::{{closure}}
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/mod.rs:446
  38: tokio::runtime::context::enter
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/context.rs:72
  39: tokio::runtime::handle::Handle::enter
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/handle.rs:76
  40: tokio::runtime::Runtime::block_on
             at /home/trangar/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.21/src/runtime/mod.rs:441
  41: foo::main
             at src/main.rs:5
  42: std::rt::lang_start::{{closure}}
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libstd/rt.rs:67
  43: std::rt::lang_start_internal::{{closure}}
             at src/libstd/rt.rs:52
  44: std::panicking::try::do_call
             at src/libstd/panicking.rs:297
  45: std::panicking::try
             at src/libstd/panicking.rs:274
  46: std::panic::catch_unwind
             at src/libstd/panic.rs:394
  47: std::rt::lang_start_internal
             at src/libstd/rt.rs:51
  48: std::rt::lang_start
             at /home/trangar/.rustup/toolchains/nightly-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/src/libstd/rt.rs:67
  49: main
  50: __libc_start_main
  51: _start
note: Some details are omitted, run with `RUST_BACKTRACE=full` for a verbose backtrace.
```

</details>

I believe this is because reqwest is being used in blocking mode.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 392383d..9086238 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -24,17 +24,19 @@ dependencies = [
 name = "appinsights"
 version = "0.1.5"
 dependencies = [
+ "async-trait",
  "chrono",
- "crossbeam-channel",
+ "crossbeam-queue",
  "env_logger",
- "futures",
+ "futures-channel",
+ "futures-util",
  "hostname",
  "http",
  "hyper",
  "lazy_static",
  "log",
  "matches",
- "mockito",
+ "parking_lot",
  "paste",
  "reqwest",
  "serde",
@@ -57,13 +59,14 @@ dependencies = [
 ]
 
 [[package]]
-name = "assert-json-diff"
-version = "2.0.1"
+name = "async-trait"
+version = "0.1.51"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50f1c3703dd33532d7f0ca049168930e9099ecac238e23cf932f3a69c42f06da"
+checksum = "44318e776df68115a881de9a8fd1b9e53368d7a4a5ce4cc48517da3393233a5e"
 dependencies = [
- "serde",
- "serde_json",
+ "proc-macro2 1.0.24",
+ "quote 1.0.9",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -171,12 +174,6 @@ dependencies = [
  "indexmap",
 ]
 
-[[package]]
-name = "const_fn"
-version = "0.4.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ce90df4c658c62f12d78f7508cf92f9173e5184a539c10bfe54a3107b3ffd0f2"
-
 [[package]]
 name = "core-foundation"
 version = "0.9.1"
@@ -194,10 +191,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ea221b5284a47e40033bf9b66f35f984ec0ea2931eb03505246cd27a963f981b"
 
 [[package]]
-name = "crossbeam-channel"
-version = "0.5.1"
+name = "crossbeam-queue"
+version = "0.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "06ed27e177f16d65f0f0c22a213e17c696ace5dd64b14258b52f9417ccb52db4"
+checksum = "9b10ddc024425c88c2ad148c1b0fd53f4c6d38db9697c9f1588381212fa657c9"
 dependencies = [
  "cfg-if 1.0.0",
  "crossbeam-utils",
@@ -205,22 +202,14 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-utils"
-version = "0.8.0"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec91540d98355f690a86367e566ecad2e9e579f230230eb7c21398372be73ea5"
+checksum = "d82cfc11ce7f2c3faef78d8a684447b40d503d9681acebed6cb728d45940c4db"
 dependencies = [
- "autocfg",
  "cfg-if 1.0.0",
- "const_fn",
  "lazy_static",
 ]
 
-[[package]]
-name = "difference"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "524cbf6897b527295dff137cec09ecf3a05f4fddffd7dfcd1585403449e74198"
-
 [[package]]
 name = "encoding_rs"
 version = "0.8.22"
@@ -274,20 +263,6 @@ dependencies = [
  "percent-encoding",
 ]
 
-[[package]]
-name = "futures"
-version = "0.3.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1adc00f486adfc9ce99f77d717836f0c5aa84965eb0b4f051f4e83f7cab53f8b"
-dependencies = [
- "futures-channel",
- "futures-core",
- "futures-io",
- "futures-sink",
- "futures-task",
- "futures-util",
-]
-
 [[package]]
 name = "futures-channel"
 version = "0.3.16"
@@ -295,7 +270,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "74ed2411805f6e4e3d9bc904c95d5d423b89b3b25dc0250aa74729de20629ff9"
 dependencies = [
  "futures-core",
- "futures-sink",
 ]
 
 [[package]]
@@ -304,12 +278,6 @@ version = "0.3.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "af51b1b4a7fdff033703db39de8802c673eb91855f2e0d47dcf3bf2c0ef01f99"
 
-[[package]]
-name = "futures-io"
-version = "0.3.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0b0e06c393068f3a6ef246c75cdca793d6a46347e75286933e5e75fd2fd11582"
-
 [[package]]
 name = "futures-macro"
 version = "0.3.16"
@@ -320,7 +288,7 @@ dependencies = [
  "proc-macro-hack",
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -343,11 +311,8 @@ checksum = "67eb846bfd58e44a8481a00049e82c43e0ccb5d61f8dc071057cb19249dd4d78"
 dependencies = [
  "autocfg",
  "futures-core",
- "futures-io",
  "futures-macro",
- "futures-sink",
  "futures-task",
- "memchr",
  "pin-project-lite",
  "pin-utils",
  "proc-macro-hack",
@@ -537,6 +502,15 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "instant"
+version = "0.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bee0328b1209d157ef001c94dd85b4f8f64139adb0eac2659f4b08382b2f474d"
+dependencies = [
+ "cfg-if 1.0.0",
+]
+
 [[package]]
 name = "ipnet"
 version = "2.3.0"
@@ -566,9 +540,18 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.71"
+version = "0.2.101"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9457b06509d27052635f90d6466700c65095fdf75409b3fbdd903e988b886f49"
+checksum = "3cb00336871be5ed2c8ed44b60ae9959dc5b9f08539422ed43f09e34ecaeba21"
+
+[[package]]
+name = "lock_api"
+version = "0.4.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "712a4d093c9976e24e7dbca41db895dabcbac38eb5f4045393d17a95bdfb1109"
+dependencies = [
+ "scopeguard",
+]
 
 [[package]]
 name = "log"
@@ -626,23 +609,6 @@ dependencies = [
  "winapi",
 ]
 
-[[package]]
-name = "mockito"
-version = "0.30.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d10030163d67f681db11810bc486df3149e6d91c8b4f3f96fa8b62b546c2cef8"
-dependencies = [
- "assert-json-diff",
- "difference",
- "httparse",
- "lazy_static",
- "log",
- "rand 0.8.2",
- "regex",
- "serde_json",
- "serde_urlencoded",
-]
-
 [[package]]
 name = "native-tls"
 version = "0.2.7"
@@ -691,9 +657,9 @@ dependencies = [
 
 [[package]]
 name = "num_cpus"
-version = "1.12.0"
+version = "1.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "46203554f085ff89c235cd12f7075f3233af9b11ed7c9e16dfe2560d03313ce6"
+checksum = "05499f3756671c15885fee9034446956fff3f243d6077b91e5767df161f766b3"
 dependencies = [
  "hermit-abi",
  "libc",
@@ -738,6 +704,31 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "parking_lot"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
+dependencies = [
+ "instant",
+ "lock_api",
+ "parking_lot_core",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d76e8e1493bcac0d2766c42737f34458f1c8c50c0d23bcb24ea953affb273216"
+dependencies = [
+ "cfg-if 1.0.0",
+ "instant",
+ "libc",
+ "redox_syscall 0.2.10",
+ "smallvec",
+ "winapi",
+]
+
 [[package]]
 name = "paste"
 version = "1.0.5"
@@ -776,7 +767,7 @@ checksum = "65ad2ae56b6abe3a1ee25f15ee605bacadb9a764edaba9c2bf4103800d4a1895"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -787,7 +778,7 @@ checksum = "81a4ffa594b66bff340084d4081df649a7dc049ac8d7fc458d8e628bfbbb2f86"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -823,7 +814,7 @@ dependencies = [
  "proc-macro-error-attr",
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
  "version_check",
 ]
 
@@ -835,7 +826,7 @@ checksum = "4f5444ead4e9935abd7f27dc51f7e852a0569ac888096d5ec2499470794e2e53"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
  "syn-mid",
  "version_check",
 ]
@@ -896,21 +887,9 @@ checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03"
 dependencies = [
  "getrandom 0.1.14",
  "libc",
- "rand_chacha 0.2.1",
- "rand_core 0.5.1",
- "rand_hc 0.2.0",
-]
-
-[[package]]
-name = "rand"
-version = "0.8.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "18519b42a40024d661e1714153e9ad0c3de27cd495760ceb09710920f1098b1e"
-dependencies = [
- "libc",
- "rand_chacha 0.3.0",
- "rand_core 0.6.1",
- "rand_hc 0.3.0",
+ "rand_chacha",
+ "rand_core",
+ "rand_hc",
 ]
 
 [[package]]
@@ -920,17 +899,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "03a2a90da8c7523f554344f921aa97283eadf6ac484a6d2a7d0212fa7f8d6853"
 dependencies = [
  "c2-chacha",
- "rand_core 0.5.1",
-]
-
-[[package]]
-name = "rand_chacha"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e12735cf05c9e10bf21534da50a147b924d555dc7a547c42e6bb2d5b6017ae0d"
-dependencies = [
- "ppv-lite86",
- "rand_core 0.6.1",
+ "rand_core",
 ]
 
 [[package]]
@@ -942,38 +911,29 @@ dependencies = [
  "getrandom 0.1.14",
 ]
 
-[[package]]
-name = "rand_core"
-version = "0.6.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c026d7df8b298d90ccbbc5190bd04d85e159eaf5576caeacf8741da93ccbd2e5"
-dependencies = [
- "getrandom 0.2.1",
-]
-
 [[package]]
 name = "rand_hc"
 version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ca3129af7b92a17112d59ad498c6f81eaf463253766b90396d39ea7a39d6613c"
 dependencies = [
- "rand_core 0.5.1",
+ "rand_core",
 ]
 
 [[package]]
-name = "rand_hc"
-version = "0.3.0"
+name = "redox_syscall"
+version = "0.1.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3190ef7066a446f2e7f42e239d161e905420ccab01eb967c9eb27d21b2322a73"
-dependencies = [
- "rand_core 0.6.1",
-]
+checksum = "2439c63f3f6139d1b57529d16bc3b8bb855230c8efcc5d3a896c8bea7c3b1e84"
 
 [[package]]
 name = "redox_syscall"
-version = "0.1.56"
+version = "0.2.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2439c63f3f6139d1b57529d16bc3b8bb855230c8efcc5d3a896c8bea7c3b1e84"
+checksum = "8383f39639269cde97d255a32bdb68c047337295414940c68bdd30c2e13203ff"
+dependencies = [
+ "bitflags",
+]
 
 [[package]]
 name = "regex"
@@ -1085,6 +1045,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "scopeguard"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
+
 [[package]]
 name = "sct"
 version = "0.6.0"
@@ -1135,7 +1101,7 @@ checksum = "e57ae87ad533d9a56427558b516d0adac283614e347abf85b0dc0cbbf0a249f3"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -1189,9 +1155,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.1.0"
+version = "1.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "44e59e0c9fa00817912ae6e4e6e3c4fe04455e75699d06eedc7d85917ed8e8f4"
+checksum = "fe0f37c9e8f3c5a4a66ad655a93c74daac4ad00c441533bf5c6e7990bb42604e"
 
 [[package]]
 name = "socket2"
@@ -1237,7 +1203,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -1253,9 +1219,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "1.0.60"
+version = "1.0.67"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c700597eca8a5a762beb35753ef6b94df201c81cca676604f547495a0d7f0081"
+checksum = "6498a9efc342871f91cc2d0d694c674368b4ceb40f62b65a7a08c3792935e702"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
@@ -1270,7 +1236,7 @@ checksum = "7be3539f6c128a931cf19dcee741c1af532c7fd387baa739c03dd2e96479338a"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -1281,8 +1247,8 @@ checksum = "7a6e24d9338a0a5be79593e2fa15a648add6138caa803e2d5bc782c371732ca9"
 dependencies = [
  "cfg-if 0.1.10",
  "libc",
- "rand 0.7.3",
- "redox_syscall",
+ "rand",
+ "redox_syscall 0.1.56",
  "remove_dir_all",
  "winapi",
 ]
@@ -1305,7 +1271,7 @@ dependencies = [
  "cfg-if 1.0.0",
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
  "version_check",
 ]
 
@@ -1362,7 +1328,7 @@ checksum = "54473be61f4ebe4efd09cec9bd5d16fa51d70ea0192213d754d2d500457db110"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
 ]
 
 [[package]]
@@ -1585,7 +1551,7 @@ dependencies = [
  "log",
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
  "wasm-bindgen-shared",
 ]
 
@@ -1619,7 +1585,7 @@ checksum = "f249f06ef7ee334cc3b8ff031bfc11ec99d00f34d86da7498396dc1e3b1498fe"
 dependencies = [
  "proc-macro2 1.0.24",
  "quote 1.0.9",
- "syn 1.0.60",
+ "syn 1.0.67",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -1661,9 +1627,9 @@ dependencies = [
 
 [[package]]
 name = "winapi"
-version = "0.3.8"
+version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8093091eeb260906a183e6ae1abdba2ef5ef2257a21801128899c3fc699229c6"
+checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419"
 dependencies = [
  "winapi-i686-pc-windows-gnu",
  "winapi-x86_64-pc-windows-gnu",
diff --git a/README.md b/README.md
index 8615707..d12b239 100644
--- a/README.md
+++ b/README.md
@@ -61,6 +61,9 @@ fn main() {
 
     // send trace telemetry to the Application Insights server
     client.track_trace("A database error occurred", SeverityLevel::Warning);
+
+    // stop the client
+    client.close_channel().await;
 }
 ```
 
diff --git a/appinsights/Cargo.toml b/appinsights/Cargo.toml
index 4fb011c..31ef744 100644
--- a/appinsights/Cargo.toml
+++ b/appinsights/Cargo.toml
@@ -14,6 +14,9 @@ categories = [
     "development-tools::profiling"
 ]
 
+[lib]
+doctest = false
+
 [features]
 default = ["reqwest/default-tls"]
 rustls = ["reqwest/rustls-tls"]
@@ -24,19 +27,23 @@ serde_json = "1.0"
 chrono = "0.4"
 http = "0.2"
 uuid = { version = "0.8", features = ["v4"], default-features = false }
-reqwest = { version = "0.11", features = ["blocking", "json"], default-features = false }
+reqwest = { version = "0.11", features = ["json"], default-features = false }
 log = "0.4"
-crossbeam-channel = "0.5"
 sm = "0.9"
+tokio = { version = "1", features = ["macros", "rt"], default-features = false }
 paste = "1.0"
 hostname = "0.3"
+futures-util = { version = "0.3", default-features = false }
+futures-channel = "0.3"
+crossbeam-queue = "0.3"
+async-trait = "0.1.51"
 
 [dev-dependencies]
 test-case = "1.2"
 env_logger = "0.9"
-mockito = { version = "0.30", default-features = false }
 lazy_static = "1.4"
 matches = "0.1"
 hyper = { version = "0.14", features = ["server"], default-features = false }
-tokio = { version = "1.10", features = ["macros"], default-features = false }
-futures = { version = "0.3", default-features = false }
+tokio = { version = "1.10", features = ["macros", "rt-multi-thread"], default-features = false }
+parking_lot = "0.11"
+
diff --git a/appinsights/examples/default.rs b/appinsights/examples/default.rs
index b64a834..e4bcb21 100644
--- a/appinsights/examples/default.rs
+++ b/appinsights/examples/default.rs
@@ -4,7 +4,8 @@ use std::time::Duration;
 use appinsights::TelemetryClient;
 use log::LevelFilter;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     env_logger::builder().filter_level(LevelFilter::Debug).init();
 
     let i_key = env::var("APPINSIGHTS_INSTRUMENTATIONKEY").expect("Set APPINSIGHTS_INSTRUMENTATIONKEY first");
@@ -20,5 +21,5 @@ fn main() {
         }
     }
 
-    ai.close_channel();
+    ai.close_channel().await;
 }
diff --git a/appinsights/src/channel/memory.rs b/appinsights/src/channel/memory.rs
index 12b209c..16bee8a 100644
--- a/appinsights/src/channel/memory.rs
+++ b/appinsights/src/channel/memory.rs
@@ -1,7 +1,10 @@
-use std::thread::{self, JoinHandle};
+use std::sync::Arc;
 
-use crossbeam_channel::{unbounded, Sender};
+use async_trait::async_trait;
+use crossbeam_queue::SegQueue;
+use futures_channel::mpsc::UnboundedSender;
 use log::{debug, trace, warn};
+use tokio::task::JoinHandle;
 
 use crate::{
     channel::{command::Command, state::Worker, TelemetryChannel},
@@ -12,75 +15,72 @@ use crate::{
 
 /// A telemetry channel that stores events exclusively in memory.
 pub struct InMemoryChannel {
-    event_sender: Sender<Envelope>,
-    command_sender: Option<Sender<Command>>,
-    thread: Option<JoinHandle<()>>,
+    items: Arc<SegQueue<Envelope>>,
+    command_sender: Option<UnboundedSender<Command>>,
+    join: Option<JoinHandle<()>>,
 }
 
 impl InMemoryChannel {
     /// Creates a new instance of in-memory channel and starts a submission routine.
     pub fn new(config: &TelemetryConfig) -> Self {
-        let (event_sender, event_receiver) = unbounded();
-        let (command_sender, command_receiver) = unbounded();
+        let items = Arc::new(SegQueue::new());
 
+        let (command_sender, command_receiver) = futures_channel::mpsc::unbounded();
         let worker = Worker::new(
             Transmitter::new(config.endpoint()),
-            event_receiver,
+            items.clone(),
             command_receiver,
             config.interval(),
         );
 
-        let thread = thread::spawn(move || {
-            worker.run();
-        });
+        let handle = tokio::spawn(worker.run());
 
         Self {
-            event_sender,
+            items,
             command_sender: Some(command_sender),
-            thread: Some(thread),
+            join: Some(handle),
         }
     }
 
-    fn shutdown(&mut self, command: Command) {
+    async fn shutdown(mut self, command: Command) {
+        // send shutdown command
         if let Some(sender) = self.command_sender.take() {
-            Self::send_command(&sender, command);
+            send_command(&sender, command);
         }
 
-        if let Some(thread) = self.thread.take() {
+        // wait until worker is finished
+        if let Some(handle) = self.join.take() {
             debug!("Shutting down worker");
-            thread.join().unwrap();
+            handle.await.unwrap();
         }
     }
-
-    fn send_command(sender: &Sender<Command>, command: Command) {
-        debug!("Sending {} command to channel", command);
-        if let Err(err) = sender.send(command.clone()) {
-            warn!("Unable to send {} command to channel: {}", command, err);
-        }
-    }
-}
-
-impl Drop for InMemoryChannel {
-    fn drop(&mut self) {
-        self.shutdown(Command::Terminate);
-    }
 }
 
+#[async_trait]
 impl TelemetryChannel for InMemoryChannel {
     fn send(&self, envelop: Envelope) {
         trace!("Sending telemetry to channel");
-        if let Err(err) = self.event_sender.send(envelop) {
-            warn!("Unable to send telemetry to channel: {}", err);
-        }
+        self.items.push(envelop);
     }
 
     fn flush(&self) {
         if let Some(sender) = &self.command_sender {
-            Self::send_command(sender, Command::Flush);
+            send_command(sender, Command::Flush);
         }
     }
 
-    fn close(&mut self) {
-        self.shutdown(Command::Close);
+    async fn close(self) {
+        self.shutdown(Command::Close).await
+    }
+
+    async fn terminate(self) {
+        self.shutdown(Command::Terminate).await;
+    }
+}
+
+fn send_command(sender: &UnboundedSender<Command>, command: Command) {
+    debug!("Sending {} command to channel", command);
+    if let Err(err) = sender.unbounded_send(command.clone()) {
+        warn!("Unable to send {} command to channel: {}", command, err);
     }
 }
diff --git a/appinsights/src/channel/mod.rs b/appinsights/src/channel/mod.rs
index b6e6ba7..694c642 100644
--- a/appinsights/src/channel/mod.rs
+++ b/appinsights/src/channel/mod.rs
@@ -1,23 +1,35 @@
 mod command;
+
 mod memory;
+pub use memory::InMemoryChannel;
+
 mod retry;
+
 mod state;
 
-pub use memory::InMemoryChannel;
+use async_trait::async_trait;
 
 use crate::contracts::Envelope;
 
 /// An implementation of [TelemetryChannel](trait.TelemetryChannel.html) is responsible for queueing
 /// and periodically submitting telemetry events.
+#[async_trait]
 pub trait TelemetryChannel {
     /// Queues a single telemetry item.
     fn send(&self, envelop: Envelope);
 
-    /// Forces all pending telemetry items to be submitted. The current thread will not be blocked.
+    /// Forces all pending telemetry items to be submitted. The current task will not be blocked.
     fn flush(&self);
 
     /// Flushes and tears down the submission flow and closes internal channels.
-    /// It blocks current thread until all pending telemetry items have been submitted and it is safe to
+    /// It blocks the current task until all pending telemetry items have been submitted and it is safe to
+    /// shutdown without losing telemetry.
+    async fn close(self);
+
+    /// Flushes and tears down the submission flow and closes internal channels.
+    /// It blocks the current task until all pending telemetry items have been submitted and it is safe to
     /// shutdown without losing telemetry.
-    fn close(&mut self);
+    /// Tears down the submission flow and closes internal channels. Any telemetry waiting to be sent is discarded.
+    /// This is a more abrupt version of [close](#method.close).
+    async fn terminate(self);
 }
diff --git a/appinsights/src/channel/state.rs b/appinsights/src/channel/state.rs
index 4410dce..20cb990 100644
--- a/appinsights/src/channel/state.rs
+++ b/appinsights/src/channel/state.rs
@@ -1,6 +1,8 @@
-use std::{mem, time::Duration};
+use std::{mem, sync::Arc, time::Duration};
 
-use crossbeam_channel::{select, Receiver};
+use crossbeam_queue::SegQueue;
+use futures_channel::mpsc::UnboundedReceiver;
+use futures_util::{Future, Stream, StreamExt};
 use log::{debug, error, trace};
 use sm::{sm, Event};
 
@@ -57,27 +59,27 @@ sm! {
 
 pub struct Worker {
     transmitter: Transmitter,
-    event_receiver: Receiver<Envelope>,
-    command_receiver: Receiver<Command>,
+    items: Arc<SegQueue<Envelope>>,
+    command_receiver: UnboundedReceiver<Command>,
     interval: Duration,
 }
 
 impl Worker {
     pub fn new(
         transmitter: Transmitter,
-        event_receiver: Receiver<Envelope>,
-        command_receiver: Receiver<Command>,
+        items: Arc<SegQueue<Envelope>>,
+        command_receiver: UnboundedReceiver<Command>,
         interval: Duration,
     ) -> Self {
         Self {
             transmitter,
-            event_receiver,
+            items,
             command_receiver,
             interval,
         }
     }
 
-    pub fn run(&self) {
+    pub async fn run(mut self) {
         let mut state = Machine::new(Receiving).as_enum();
 
         let mut items: Vec<Envelope> = Default::default();
@@ -85,13 +87,13 @@ impl Worker {
 
         loop {
             state = match state {
-                InitialReceiving(m) => self.handle_receiving(m, &mut items),
-                ReceivingByItemsSentAndContinue(m) => self.handle_receiving(m, &mut items),
-                ReceivingByRetryExhausted(m) => self.handle_receiving(m, &mut items),
-                SendingByTimeoutExpired(m) => self.handle_sending_with_retry(m, &mut items, &mut retry),
-                SendingByFlushRequested(m) => self.handle_sending_with_retry(m, &mut items, &mut retry),
-                SendingByCloseRequested(m) => self.handle_sending_once_and_terminate(m, &mut items, &mut retry),
-                WaitingByRetryRequested(m) => self.handle_waiting(m, &mut retry),
+                InitialReceiving(m) => self.handle_receiving(m, &mut items).await,
+                ReceivingByItemsSentAndContinue(m) => self.handle_receiving(m, &mut items).await,
+                ReceivingByRetryExhausted(m) => self.handle_receiving(m, &mut items).await,
+                SendingByTimeoutExpired(m) => self.handle_sending_with_retry(m, &mut items, &mut retry).await,
+                SendingByFlushRequested(m) => self.handle_sending_with_retry(m, &mut items, &mut retry).await,
+                SendingByCloseRequested(m) => self.handle_sending_once_and_terminate(m, &mut items, &mut retry).await,
+                WaitingByRetryRequested(m) => self.handle_waiting(m, &mut retry).await,
                 StoppedByItemsSentAndStop(_) => break,
                 StoppedByCloseRequested(_) => break,
                 StoppedByTerminateRequested(_) => break,
@@ -99,17 +101,17 @@ impl Worker {
         }
     }
 
-    fn handle_receiving<E: Event>(&self, m: Machine<Receiving, E>, items: &mut Vec<Envelope>) -> Variant {
+    async fn handle_receiving<E: Event>(&mut self, m: Machine<Receiving, E>, items: &mut Vec<Envelope>) -> Variant {
         debug!("Receiving messages triggered by {:?}", m.trigger());
 
-        let timeout = timeout::after(self.interval);
+        let timeout = timeout::sleep(self.interval);
         items.clear();
 
         loop {
-            select! {
-                recv(self.command_receiver) -> command => {
+            tokio::select! {
+                command = self.command_receiver.next() => {
                     match command {
-                        Ok(command) => {
+                        Some(command) => {
                             trace!("Command received: {}", command);
                             match command {
                                 Command::Flush => return m.transition(FlushRequested).as_enum(),
@@ -117,13 +119,13 @@ impl Worker {
                                 Command::Close => return m.transition(CloseRequested).as_enum(),
                             }
                         },
-                        Err(err) => {
-                            error!("commands channel closed: {}", err);
+                        None => {
+                            error!("commands channel closed");
                             return m.transition(TerminateRequested).as_enum()
                         },
                     }
                 },
-                recv(timeout) -> _ => {
+                _ = timeout => {
                     debug!("Timeout expired");
                     return m.transition(TimeoutExpired).as_enum()
                 },
@@ -131,32 +133,33 @@ impl Worker {
         }
     }
 
-    fn handle_sending_with_retry<E: Event>(
-        &self,
+    async fn handle_sending_with_retry<E: Event>(
+        &mut self,
         m: Machine<Sending, E>,
         items: &mut Vec<Envelope>,
         retry: &mut Retry,
     ) -> Variant {
         *retry = Retry::exponential();
-        self.handle_sending(m, items)
+        self.handle_sending(m, items).await
     }
 
-    fn handle_sending_once_and_terminate<E: Event>(
-        &self,
+    async fn handle_sending_once_and_terminate<E: Event>(
+        &mut self,
         m: Machine<Sending, E>,
         items: &mut Vec<Envelope>,
         retry: &mut Retry,
     ) -> Variant {
         *retry = Retry::once();
         let cloned = m.clone(); // clone here
-        self.handle_sending(m, items);
+        self.handle_sending(m, items).await;
         cloned.transition(TerminateRequested).as_enum()
     }
 
-    fn handle_sending<E: Event>(&self, m: Machine<Sending, E>, items: &mut Vec<Envelope>) -> Variant {
-        // read items from a channel
-        let pending_items = self.event_receiver.try_iter();
-        items.extend(pending_items);
+    async fn handle_sending<E: Event>(&mut self, m: Machine<Sending, E>, items: &mut Vec<Envelope>) -> Variant {
+        // read pending items from a channel
+        while let Some(item) = self.items.pop() {
+            items.push(item);
+        }
 
         debug!(
             "Sending {} telemetry items triggered by {:?}",
@@ -170,7 +173,7 @@ impl Worker {
             m.transition(ItemsSentAndContinue).as_enum()
         } else {
             // attempt to send items
-            match self.transmitter.send(mem::take(items)) {
+            match self.transmitter.send(mem::take(items)).await {
                 Ok(Response::Success) => m.transition(ItemsSentAndContinue).as_enum(),
                 Ok(Response::Retry(retry_items)) => {
                     *items = retry_items;
@@ -190,7 +193,7 @@ impl Worker {
         }
     }
 
-    fn handle_waiting<E: Event>(&self, m: Machine<Waiting, E>, retry: &mut Retry) -> Variant {
+    async fn handle_waiting<E: Event>(&mut self, m: Machine<Waiting, E>, retry: &mut Retry) -> Variant {
         if let Some(timeout) = retry.next() {
             debug!(
                 "Waiting for retry timeout {:?} or stop command triggered by {:?}",
@@ -198,29 +201,25 @@ impl Worker {
                 m.state()
             );
             // sleep until next sending attempt
-            let timeout = timeout::after(timeout);
+            let timeout = timeout::sleep(timeout);
 
             // wait for either retry timeout expired or stop command received
-            loop {
-                select! {
-                    recv(self.command_receiver) -> command => {
-                        match command {
-                            Ok(command) => match command {
-                                Command::Flush => continue,
-                                Command::Terminate => return m.transition(TerminateRequested).as_enum(),
-                                Command::Close => return m.transition(CloseRequested).as_enum(),
-                            },
-                            Err(err) => {
-                                error!("commands channel closed: {}", err);
-                                return m.transition(TerminateRequested).as_enum()
-                            }
+            tokio::select! {
+                command = skip_flush(&mut self.command_receiver) => {
+                    match command {
+                        Some(Command::Terminate) => m.transition(TerminateRequested).as_enum(),
+                        Some(Command::Close) => m.transition(CloseRequested).as_enum(),
+                        Some(Command::Flush) => panic!("whoops Flush is not supported here"),
+                        None => {
+                            error!("commands channel closed");
+                            m.transition(TerminateRequested).as_enum()
                         }
-                    },
-                    recv(timeout) -> _ => {
-                        debug!("Retry timeout expired");
-                        return m.transition(TimeoutExpired).as_enum()
-                    },
-                }
+                    }
+                },
+                _ = timeout => {
+                    debug!("Retry timeout expired");
+                    m.transition(TimeoutExpired).as_enum()
+                },
             }
         } else {
             debug!("All retries exhausted by {:?}", m.state());
@@ -228,3 +227,23 @@ impl Worker {
         }
     }
 }
+
+fn skip_flush<St>(stream: &mut St) -> SkipFlush<'_, St> {
+    SkipFlush { stream }
+}
+
+struct SkipFlush<'a, St: ?Sized> {
+    stream: &'a mut St,
+}
+
+impl<St: ?Sized + Stream<Item = Command> + Unpin> Future for SkipFlush<'_, St> {
+    type Output = Option<St::Item>;
+
+    fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
+        match self.stream.poll_next_unpin(cx) {
+            std::task::Poll::Ready(Some(Command::Flush)) => std::task::Poll::Pending,
+            std::task::Poll::Ready(command) => std::task::Poll::Ready(command),
+            std::task::Poll::Pending => std::task::Poll::Pending,
+        }
+    }
+}
diff --git a/appinsights/src/client/mod.rs b/appinsights/src/client/mod.rs
index e3c8181..4e4364c 100644
--- a/appinsights/src/client/mod.rs
+++ b/appinsights/src/client/mod.rs
@@ -236,8 +236,36 @@ where
         }
     }
 
+    /// Forces all pending telemetry items to be submitted. The current task will not be blocked.
+    ///
+    /// # Examples
+    ///
+    /// ```rust, no_run
+    /// # use appinsights::TelemetryClient;
+    /// # let client = TelemetryClient::new("<instrumentation key>".to_string());
+    /// let mut counter = 0;
+    ///
+    /// // send heartbeats while application is running
+    /// let running = true;
+    /// while running {
+    ///     client.track_event("app is running");
+    ///     counter += 1;
+    ///
+    ///     // if the rate is bigger than submission interval you can make sure that data is
+    ///     // triggered for submission (each 100 items)
+    ///     if counter == 100 {
+    ///         // trigger submission of all pending items
+    ///         client.flush_channel();
+    ///         counter = 0;
+    ///     }
+    /// }
+    /// ```
+    pub fn flush_channel(&self) {
+        self.channel.flush();
+    }
+
     /// Flushes and tears down the submission flow and closes internal channels.
-    /// It block current thread until all pending telemetry items have been submitted and it is safe to
+    /// It blocks the current task until all pending telemetry items have been submitted and it is safe to
     /// shutdown without losing telemetry.
     /// This method consumes the value of client so it makes impossible to use a client with close
     /// channel.
@@ -254,42 +282,42 @@ where
     /// }
     ///
     /// // wait until pending telemetry is sent at most once and tear down submission flow
-    /// client.close_channel();
+    /// client.close_channel().await;
     ///
     /// // unable to sent any telemetry after client closes its channel
     /// // client.track_event("app is stopped".to_string());
     /// ```
-    pub fn close_channel(self) {
-        let mut channel = self.channel;
-        channel.close()
+    pub async fn close_channel(self) {
+        self.channel.close().await;
     }
 
-    /// Forces all pending telemetry items to be submitted. The current thread will not be blocked.
+    /// Tears down the submission flow and closes internal channels.
+    /// Any telemetry waiting to be sent is discarded. This is a more abrupt version of [`close_channel`](#method.close_channel).
+    /// This method consumes the value of client so it makes impossible to use a client with close
+    /// channel.
+    ///
+    /// This method should be used in cases when the client should be stopped. It is a separate function until
+    /// `async_drop` is implemented in rust.
     ///
     /// # Examples
     ///
     /// ```rust, no_run
     /// # use appinsights::TelemetryClient;
     /// # let client = TelemetryClient::new("<instrumentation key>".to_string());
-    /// let mut counter = 0;
-    ///
     /// // send heartbeats while application is running
     /// let running = true;
     /// while running {
     ///     client.track_event("app is running");
-    ///     counter += 1;
-    ///
-    ///     // if the rate is bigger than submission interval you can make sure that data is
-    ///     // triggered for submission (each 100 items)
-    ///     if counter == 100 {
-    ///         // trigger submission of all pending items
-    ///         client.flush_channel();
-    ///         counter = 0;
-    ///     }
     /// }
+    ///
+    /// // wait until pending telemetry is sent at most once and tear down submission flow
+    /// client.terminate().await;
+    ///
+    /// // unable to sent any telemetry after client closes its channel
+    /// // client.track_event("app is stopped".to_string());
     /// ```
-    pub fn flush_channel(&self) {
-        self.channel.flush();
+    pub async fn terminate(self) {
+        self.channel.terminate().await;
     }
 }
 
@@ -307,20 +335,21 @@ impl From<(TelemetryConfig, TelemetryContext)> for TelemetryClient<InMemoryChann
 mod tests {
     use std::cell::RefCell;
 
+    use async_trait::async_trait;
     use chrono::{DateTime, Utc};
     use matches::assert_matches;
 
     use super::*;
     use crate::telemetry::{ContextTags, Properties};
 
-    #[test]
-    fn it_enabled_by_default() {
+    #[tokio::test]
+    async fn it_enabled_by_default() {
         let client = TelemetryClient::new("key".into());
         assert!(client.is_enabled())
     }
 
-    #[test]
-    fn it_disables_telemetry() {
+    #[tokio::test]
+    async fn it_disables_telemetry() {
         let mut client = TelemetryClient::new("key".into());
 
         client.enabled(false);
@@ -328,8 +357,8 @@ mod tests {
         assert!(!client.is_enabled())
     }
 
-    #[test]
-    fn it_submits_telemetry() {
+    #[tokio::test]
+    async fn it_submits_telemetry() {
         let client = create_client();
 
         client.track(TestTelemetry {});
@@ -338,8 +367,8 @@ mod tests {
         assert_eq!(events.len(), 1)
     }
 
-    #[test]
-    fn it_swallows_telemetry_when_disabled() {
+    #[tokio::test]
+    async fn it_swallows_telemetry_when_disabled() {
         let mut client = create_client();
         client.enabled(false);
 
@@ -349,8 +378,8 @@ mod tests {
         assert!(events.is_empty())
     }
 
-    #[test]
-    fn it_creates_client_with_default_tags() {
+    #[tokio::test]
+    async fn it_creates_client_with_default_tags() {
         let client = TelemetryClient::new("instrumentation".into());
 
         let tags = client.context().tags();
@@ -358,6 +387,12 @@ mod tests {
         assert_matches!(tags.device().os_version(), Some(_))
     }
 
+    #[tokio::test]
+    async fn it_does_not_fail_with_tokio() {
+        let client = TelemetryClient::new("instrumentation".into());
+        assert!(client.is_enabled())
+    }
+
     fn create_client() -> TelemetryClient<TestChannel> {
         let config = TelemetryConfig::new("instrumentation".into());
         TelemetryClient {
@@ -406,6 +441,7 @@ mod tests {
         events: RefCell<Vec<Envelope>>,
     }
 
+    #[async_trait]
     impl TelemetryChannel for TestChannel {
         fn send(&self, envelop: Envelope) {
             self.events.borrow_mut().push(envelop);
@@ -415,7 +451,11 @@ mod tests {
             unimplemented!()
         }
 
-        fn close(&mut self) {
+        async fn close(self) {
+            unimplemented!()
+        }
+
+        async fn terminate(self) {
             unimplemented!()
         }
     }
diff --git a/appinsights/src/lib.rs b/appinsights/src/lib.rs
index ede9996..9c9de5a 100644
--- a/appinsights/src/lib.rs
+++ b/appinsights/src/lib.rs
@@ -62,6 +62,9 @@
 //!
 //! // send trace telemetry to the Application Insights server
 //! client.track_trace("A database error occurred", SeverityLevel::Warning);
+//!
+//! // stop the client
+//! client.close_channel().await
 //! ```
 //!
 //! ## Telemetry submission
@@ -118,6 +121,9 @@
 //!
 //! // send telemetry to the Application Insights server
 //! client.track(telemetry);
+//!
+//! // stop the client
+//! client.close_channel().awai
 //! ```
 //!
 //! ## Common properties
@@ -142,19 +148,18 @@
 //!
 //! ## Shutdown
 //!
-//! The telemetry item submission happens asynchronously. The internal channel starts a new worker
-//! thread that used to accept and send telemetry to the server. While telemetry is not sent the
+//! The telemetry item submission happens asynchronously. The internal channel spawns a new worker
+//! task that accepts and sends telemetry to the server in the background. While telemetry is not sent the
 //! worker stores it in memory, so when application crashes the data will be lost. Luckily SDK
 //! provides several convenient methods to deal with this issue.
 //! * [`flush_channel`](struct.TelemetryClient.html#method.flush_channel) will trigger telemetry submission
 //! as soon as possible. It returns immediately and telemetry is no guaranteed to be sent.
 //! * [`close_channel`](struct.TelemetryClient.html#method.close_channel) will cause the channel to
-//! stop accepting any new telemetry items, submit all pending ones, block current thread and
+//! stop accepting any new telemetry items, submit all pending ones, block current task and
 //! wait until data will be sent at most once. If telemetry submission fails, it will not retry.
 //! This method consumes the value of client so it makes impossible to use a client with close channel.
-//! * Once [`TelemetryClient`](struct.TelemetryClient.html) is out of scope `drop` method for channel
-//! will be called. It will trigger termination of submission flow, all pending items discarded,
-//! block current thread until all resources freed. It is __default__ "exit" mode for client.
+//! * [`terminate`](struct.TelemetryClient.html#method.terminate) will trigger termination of submission flow, all pending items discarded and
+//! current task will be blocked until all resources freed.
 #![deny(unused_extern_crates)]
 #![deny(missing_docs)]
 
diff --git a/appinsights/src/telemetry/tags.rs b/appinsights/src/telemetry/tags.rs
index 4f47431..d0d7adb 100644
--- a/appinsights/src/telemetry/tags.rs
+++ b/appinsights/src/telemetry/tags.rs
@@ -206,7 +206,7 @@ tags!(
     internal,
     /// Tag helper type that provides access to context fields grouped under 'internal'.
     InternalTags {
-        /// SDK version. See https://github.com/Microsoft/ApplicationInsights-Home/blob/master/SDK-AUTHORING.md#sdk-version-specification for information.
+        /// SDK version. See `https://github.com/Microsoft/ApplicationInsights-Home/blob/master/SDK-AUTHORING.md#sdk-version-specification` for information.
         sdk_version: "ai.internal.sdkVersion",
         /// Agent version. Used to indicate the version of StatusMonitor installed on the computer if it is used for data collection.
         agent_version: "ai.internal.agentVersion",
diff --git a/appinsights/src/timeout.rs b/appinsights/src/timeout.rs
index f87f1fd..b5203f8 100644
--- a/appinsights/src/timeout.rs
+++ b/appinsights/src/timeout.rs
@@ -2,58 +2,62 @@ pub use imp::*;
 
 #[cfg(not(test))]
 mod imp {
-    use std::time::{Duration, Instant};
+    use std::time::Duration;
 
-    use crossbeam_channel::Receiver;
+    use tokio::time::{self, Instant};
 
     /// Creates a receiver that reliably delivers only one message when given interval expires.
-    pub fn after(duration: Duration) -> Receiver<Instant> {
-        crossbeam_channel::after(duration)
+    pub async fn sleep(duration: Duration) {
+        let timeout = Instant::now() + duration;
+        time::sleep_until(timeout).await;
     }
 }
 
 #[cfg(test)]
 mod imp {
-    use std::sync::RwLock;
-    use std::time::{Duration, Instant};
+    use std::{sync::Arc, time::Duration};
 
-    use crossbeam_channel::{Receiver, Sender};
     use lazy_static::lazy_static;
+    use parking_lot::Mutex;
+    use tokio::{sync::Notify, time::Instant};
 
     lazy_static! {
-        static ref CHANNEL: RwLock<Option<(Sender<Instant>, Receiver<Instant>)>> = RwLock::new(None);
+        static ref CHANNEL: Mutex<Option<Arc<Notify>>> = Mutex::new(None);
     }
 
     /// Initializes a channel which emulates timeout expiration event. External code should run
     /// [`expire`](#method.expire) method in order to emulate timeout expiration.
-    pub fn init() {
-        let mut channel = CHANNEL.write().expect("lock");
-        *channel = Some(crossbeam_channel::bounded(1));
+    pub async fn init() {
+        let mut channel = CHANNEL.lock();
+        *channel = Some(Arc::new(Notify::new()));
     }
 
     /// Creates a copy of a receiver that delivers a current time stamp in order to emulate
     /// timeout expiration for tests.
-    pub fn after(duration: Duration) -> Receiver<Instant> {
-        CHANNEL
-            .read()
-            .expect("lock")
-            .as_ref()
-            .map_or_else(|| crossbeam_channel::after(duration), |(_, receiver)| receiver.clone())
+    pub async fn sleep(duration: Duration) {
+        let maybe_notify = CHANNEL.lock().clone();
+
+        if let Some(notify) = maybe_notify {
+            notify.notified().await;
+        } else {
+            let timeout = Instant::now() + duration;
+            tokio::time::sleep_until(timeout).await;
+        }
     }
 
     /// Emulates timeout expiration event.
     /// It sends a current time stamp to receiver in order to trigger an action if a channel was
-    /// initialized in advance. Does't nothing otherwise.
-    pub fn expire() {
-        if let Some((sender, _)) = CHANNEL.read().expect("lock").as_ref() {
-            sender.send(Instant::now()).unwrap();
+    /// initialized in advance. Does nothing otherwise.
+    pub async fn expire() {
+        if let Some(notify) = CHANNEL.lock().clone() {
+            notify.notify_one();
         }
     }
 
     /// Resets a channel that emulates timeout expiration event with default
-    /// crossbeam_channel::bounded() instead.
-    pub fn reset() {
-        let mut channel = CHANNEL.write().expect("lock");
+    /// timer base timeout expiration instead.
+    pub async fn reset() {
+        let mut channel = CHANNEL.lock();
         *channel = None;
     }
 }
diff --git a/appinsights/src/transmitter.rs b/appinsights/src/transmitter.rs
index 3f54f71..c6bc910 100644
--- a/appinsights/src/transmitter.rs
+++ b/appinsights/src/transmitter.rs
@@ -1,7 +1,7 @@
 use chrono::{DateTime, Utc};
 use http::{header::RETRY_AFTER, StatusCode};
 use log::debug;
-use reqwest::blocking::Client;
+use reqwest::Client;
 
 use crate::{
     contracts::{Envelope, Transmission, TransmissionItem},
@@ -33,17 +33,17 @@ impl Transmitter {
     }
 
     /// Sends a telemetry items to the server.
-    pub fn send(&self, mut items: Vec<Envelope>) -> Result<Response> {
+    pub async fn send(&self, mut items: Vec<Envelope>) -> Result<Response> {
         let payload = serde_json::to_string(&items)?;
 
-        let response = self.client.post(&self.url).body(payload).send()?;
+        let response = self.client.post(&self.url).body(payload).send().await?;
         let response = match response.status() {
             StatusCode::OK => {
                 debug!("Successfully sent {} items", items.len());
                 Response::Success
             }
             StatusCode::PARTIAL_CONTENT => {
-                let content: Transmission = response.json()?;
+                let content: Transmission = response.json().await?;
                 let log_prefix = format!(
                     "Successfully sent {}/{} telemetry items",
                     content.items_accepted, content.items_received
@@ -65,7 +65,7 @@ impl Transmitter {
             StatusCode::TOO_MANY_REQUESTS | StatusCode::REQUEST_TIMEOUT => {
                 let retry_after = response.headers().get(RETRY_AFTER).cloned();
 
-                if let Ok(content) = response.json::<Transmission>() {
+                if let Ok(content) = response.json::<Transmission>().await {
                     retain_retry_items(&mut items, content);
                 }
 
@@ -88,7 +88,7 @@ impl Transmitter {
                 Response::Retry(items.to_vec())
             }
             StatusCode::INTERNAL_SERVER_ERROR => {
-                if let Ok(content) = response.json::<Transmission>() {
+                if let Ok(content) = response.json::<Transmission>().await {
                     retain_retry_items(&mut items, content);
                     if items.is_empty() {
                         debug!("Service error. Nothing to re-send");
@@ -106,7 +106,7 @@ impl Transmitter {
                 debug!(
                     "Unknown status: {}. {}. Nothing to re-send",
                     response.status(),
-                    response.text().unwrap_or_default()
+                    response.text().await.unwrap_or_default()
                 );
                 Response::NoRetry
             }
@@ -139,8 +139,11 @@ fn can_retry_item(item: &TransmissionItem) -> bool {
 #[cfg(test)]
 mod tests {
     use chrono::TimeZone;
-    use http::StatusCode;
-    use mockito::mock;
+    use http::{Request, StatusCode};
+    use hyper::{
+        service::{make_service_fn, service_fn},
+        Body, Server,
+    };
     use serde_json::{json, Value};
     use test_case::test_case;
 
@@ -164,25 +167,51 @@ mod tests {
     fn it_sends_telemetry_and_handles_server_response(
         items: Vec<Envelope>,
         status_code: StatusCode,
-        retry_after: Option<&str>,
+        retry_after: Option<&'static str>,
         body: Option<Value>,
         expected: Response,
     ) {
-        let mut _server = mock("POST", "/track").with_status(status_code.as_u16() as usize);
-        if let Some(retry_after) = retry_after {
-            _server = _server.with_header("Retry-After", retry_after);
-        }
-        if let Some(body) = body {
-            _server = _server.with_body(body.to_string());
-        }
-        _server = _server.create();
-        let url = mockito::server_url();
+        let rt = tokio::runtime::Runtime::new().expect("runtime");
+        rt.block_on(async {
+            let url = create_server(status_code, retry_after, body);
+
+            let transmitter = Transmitter::new(&format!("{}/track", url));
+
+            let response = transmitter.send(items).await.unwrap();
+
+            assert_eq!(response, expected);
+        });
+    }
+
+    fn create_server(status_code: StatusCode, retry_after: Option<&'static str>, body: Option<Value>) -> String {
+        let make_service = make_service_fn(move |_| {
+            let retry_after = retry_after.map(ToString::to_string);
+            let body = body.clone();
+            async move {
+                Ok::<_, hyper::Error>(service_fn(move |_: Request<Body>| {
+                    let retry_after = retry_after.clone();
+                    let body = body.clone();
+                    async move {
+                        let mut builder = hyper::Response::builder().status(status_code);
+
+                        if let Some(retry_after) = retry_after {
+                            builder = builder.header("Retry-After", retry_after);
+                        }
+
+                        let body = body.map(move |body| Body::from(body.to_string())).unwrap_or_default();
+
+                        builder.body(body)
+                    }
+                }))
+            }
+        });
 
-        let transmitter = Transmitter::new(&format!("{}/track", url));
+        let server = Server::bind(&([0, 0, 0, 0], 0).into()).serve(make_service);
+        let url = format!("http://{}", server.local_addr());
 
-        let response = transmitter.send(items).unwrap();
+        tokio::spawn(server);
 
-        assert_eq!(response, expected)
+        url
     }
 
     fn partial_no_retries() -> Value {


Test patch:
diff --git a/appinsights/src/client/integration_tests.rs b/appinsights/src/client/integration_tests.rs
index 770e83b..201ad8b 100644
--- a/appinsights/src/client/integration_tests.rs
+++ b/appinsights/src/client/integration_tests.rs
@@ -1,13 +1,12 @@
 use std::{
     sync::{
         atomic::{AtomicUsize, Ordering},
-        Arc, Mutex,
+        Arc,
     },
     time::Duration,
 };
 
 use chrono::{DateTime, Utc};
-use crossbeam_channel::{bounded, unbounded, Receiver, RecvTimeoutError};
 use hyper::{
     body::Buf,
     service::{make_service_fn, service_fn},
@@ -15,8 +14,12 @@ use hyper::{
 };
 use lazy_static::lazy_static;
 use matches::assert_matches;
+use parking_lot::Mutex;
 use serde_json::json;
-use tokio::sync::oneshot;
+use tokio::sync::{
+    mpsc::{self, Receiver},
+    oneshot,
+};
 
 use crate::{channel::InMemoryChannel, timeout, TelemetryClient, TelemetryConfig};
 
@@ -26,65 +29,68 @@ lazy_static! {
 }
 
 macro_rules! manual_timeout_test {
-    (fn $name: ident() $body: block) => {
+    (async fn $name: ident() $body: block) => {
         #[test]
         fn $name() {
-            let guard = SERIAL_TEST_MUTEX.lock().unwrap();
-            timeout::init();
+            let _guard = SERIAL_TEST_MUTEX.lock();
 
-            // Catch any panics to not poison the lock.
-            if let Err(err) = std::panic::catch_unwind(|| $body) {
-                drop(guard);
+            let rt = tokio::runtime::Runtime::new().expect("runtime");
+            rt.block_on(async {
+                timeout::init().await;
 
-                std::panic::resume_unwind(err);
-            }
+                $body;
 
-            timeout::reset();
+                timeout::reset().await;
+            });
         }
     };
 }
 
 manual_timeout_test! {
-    fn it_sends_one_telemetry_item() {
-        let server = server().status(StatusCode::OK).create();
+    async fn it_sends_one_telemetry_item() {
+        let mut server = server().status(StatusCode::OK).create();
 
         let client = create_client(server.url());
         client.track_event("--event--");
 
-        timeout::expire();
+        timeout::expire().await;
 
         // expect one requests available so far
-        let receiver = server.requests();
-        assert_matches!(receiver.recv_timeout(Duration::from_millis(500)), Ok(_));
+        assert_matches!(server.next_request_timeout().await, Ok(_));
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_does_not_resend_submitted_telemetry_items() {
-        let server = server().status(StatusCode::OK).create();
+    async fn it_does_not_resend_submitted_telemetry_items() {
+        let mut server = server().status(StatusCode::OK).create();
 
         let client = create_client(server.url());
         client.track_event("--event--");
 
         // verify 1 items is sent after first interval expired
-        let receiver = server.requests();
 
         // "wait" until interval expired
-        timeout::expire();
-        assert_matches!(receiver.recv_timeout(Duration::from_millis(500)), Ok(_));
+        timeout::expire().await;
+        assert_matches!(server.next_request_timeout().await, Ok(_));
 
         // verify no items is sent after next interval expired
-        timeout::expire();
+        timeout::expire().await;
         assert_matches!(
-            receiver.recv_timeout(Duration::from_millis(500)),
+            server.next_request_timeout().await,
             Err(RecvTimeoutError::Timeout)
         );
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_sends_telemetry_items_in_several_batches() {
-        let server = server().status(StatusCode::OK).status(StatusCode::OK).create();
+    async fn it_sends_telemetry_items_in_several_batches() {
+        let mut server = server().status(StatusCode::OK).status(StatusCode::OK).create();
 
         let client = create_client(server.url());
 
@@ -94,7 +100,7 @@ manual_timeout_test! {
         }
 
         // "wait" until interval expired
-        timeout::expire();
+        timeout::expire().await;
 
         // send next 5 items and then interval expired
         for i in 10..15 {
@@ -102,10 +108,10 @@ manual_timeout_test! {
         }
 
         // "wait" until next interval expired
-        timeout::expire();
+        timeout::expire().await;
 
         // verify that all items were send
-        let requests = server.wait_for_requests(2);
+        let requests = server.wait_for_requests(2).await;
         let content = requests.into_iter().fold(String::new(), |mut content, body| {
             content.push_str(&body);
             content
@@ -114,12 +120,15 @@ manual_timeout_test! {
             .filter(|i| content.contains(&format!("--event {}--", i)))
             .count();
         assert_eq!(items_count, 15);
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_flushes_all_pending_telemetry_items() {
-        let server = server().status(StatusCode::OK).status(StatusCode::OK).create();
+    async fn it_flushes_all_pending_telemetry_items() {
+        let mut server = server().status(StatusCode::OK).status(StatusCode::OK).create();
 
         let client = create_client(server.url());
 
@@ -133,7 +142,7 @@ manual_timeout_test! {
 
         // NOTE no timeout expired
         // assert that 1 request has been sent
-        let requests = server.wait_for_requests(1);
+        let requests = server.wait_for_requests(1).await;
         assert_eq!(requests.len(), 1);
 
         // verify request contains all items we submitted to the client
@@ -145,12 +154,15 @@ manual_timeout_test! {
             .filter(|i| content.contains(&format!("--event {}--", i)))
             .count();
         assert_eq!(items_count, 15);
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_does_not_send_any_pending_telemetry_items_when_drop_client() {
-        let server = server().status(StatusCode::OK).status(StatusCode::OK).create();
+    async fn it_does_not_send_any_pending_telemetry_items_when_drop_client() {
+        let mut server = server().status(StatusCode::OK).status(StatusCode::OK).create();
 
         let client = create_client(server.url());
 
@@ -163,18 +175,19 @@ manual_timeout_test! {
         drop(client);
 
         // verify that nothing has been sent to the server
-        let receiver = server.requests();
         assert_matches!(
-            receiver.recv_timeout(Duration::from_millis(500)),
+            server.next_request_timeout().await,
             Err(RecvTimeoutError::Timeout)
         );
 
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_tries_to_send_pending_telemetry_items_when_close_channel_requested() {
-        let server = server().status(StatusCode::OK).status(StatusCode::OK).create();
+    async fn it_tries_to_send_pending_telemetry_items_when_close_channel_requested() {
+        let mut server = server().status(StatusCode::OK).status(StatusCode::OK).create();
 
         let client = create_client(server.url());
 
@@ -185,11 +198,11 @@ manual_timeout_test! {
 
         // close internal channel means that client will make an attempt to send telemetry items once
         // and then tear down submission flow
-        client.close_channel();
+        client.close_channel().await;
 
         // NOTE no timeout expired
         // verify that 1 request has been sent
-        let requests = server.wait_for_requests(1);
+        let requests = server.wait_for_requests(1).await;
         assert_eq!(requests.len(), 1);
 
         // verify request contains all items we submitted to the client
@@ -201,12 +214,39 @@ manual_timeout_test! {
             .filter(|i| content.contains(&format!("--event {}--", i)))
             .count();
         assert_eq!(items_count, 15);
+
+        // terminate server
+        server.terminate().await;
+    }
+}
+
+manual_timeout_test! {
+    async fn it_does_not_try_to_send_pending_telemetry_items_when_client_terminated() {
+        let mut server = server().status(StatusCode::OK).status(StatusCode::OK).create();
+
+        let client = create_client(server.url());
+
+        // send 15 items and then interval expired
+        for i in 0..15 {
+            client.track_event(format!("--event {}--", i));
+        }
+
+        // terminate client
+        client.terminate().await;
+
+        // NOTE no timeout expired
+        // verify that no request has been sent
+        let requests = server.wait_for_requests(1).await;
+        assert!(requests.is_empty());
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_retries_when_previous_submission_failed() {
-        let server = server()
+    async fn it_retries_when_previous_submission_failed() {
+        let mut server = server()
             .response(StatusCode::INTERNAL_SERVER_ERROR, json!({}), None)
             .response(
                 StatusCode::OK,
@@ -228,21 +268,24 @@ manual_timeout_test! {
         }
 
         // "wait" until interval expired
-        timeout::expire();
+        timeout::expire().await;
 
         // "wait" until retry logic handled
-        timeout::expire();
+        timeout::expire().await;
 
         // verify there are 2 identical requests
-        let requests = server.wait_for_requests(2);
+        let requests = server.wait_for_requests(2).await;
         assert_eq!(requests.len(), 2);
         assert_eq!(requests[0], requests[1]);
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
 manual_timeout_test! {
-    fn it_retries_when_partial_content() {
-        let server = server()
+    async fn it_retries_when_partial_content() {
+        let mut server = server()
             .response(
                 StatusCode::PARTIAL_CONTENT,
                 json!(
@@ -289,13 +332,13 @@ manual_timeout_test! {
         }
 
         // "wait" until interval expired
-        timeout::expire();
+        timeout::expire().await;
 
         // "wait" until retry logic handled
-        timeout::expire();
+        timeout::expire().await;
 
         // verify it sends a first request with all items
-        let requests = server.wait_for_requests(1);
+        let requests = server.wait_for_requests(1).await;
         assert_eq!(requests.len(), 1);
 
         let content = requests.into_iter().fold(String::new(), |mut content, body| {
@@ -308,7 +351,7 @@ manual_timeout_test! {
         assert_eq!(items_count, 15);
 
         // verify it re-send only errors that previously were invalid
-        let requests = server.wait_for_requests(1);
+        let requests = server.wait_for_requests(1).await;
         assert_eq!(requests.len(), 1);
 
         let content = requests.into_iter().fold(String::new(), |mut content, body| {
@@ -320,6 +363,9 @@ manual_timeout_test! {
             .filter(|i| content.contains(&format!("--event {}--", i)))
             .count();
         assert_eq!(items_count, 3);
+
+        // terminate server
+        server.terminate().await;
     }
 }
 
@@ -335,18 +381,14 @@ fn create_client(endpoint: &str) -> TelemetryClient<InMemoryChannel> {
     TelemetryClient::from_config(config)
 }
 
-struct Builder {
-    responses: Vec<Response<String>>,
-}
-
 fn server() -> Builder {
     Builder { responses: Vec::new() }
 }
 
 struct HyperTestServer {
     url: String,
-    requests: Receiver<String>,
-    shutdown: Option<oneshot::Sender<()>>,
+    request_recv: Receiver<String>,
+    shutdown_send: Option<oneshot::Sender<()>>,
 }
 
 impl HyperTestServer {
@@ -354,15 +396,19 @@ impl HyperTestServer {
         &self.url
     }
 
-    fn requests(&self) -> Receiver<String> {
-        self.requests.clone()
+    async fn next_request_timeout(&mut self) -> Result<String, RecvTimeoutError> {
+        match tokio::time::timeout(Duration::from_millis(100), self.request_recv.recv()).await {
+            Ok(Some(x)) => Ok(x),
+            Ok(None) => Err(RecvTimeoutError::Closed),
+            Err(_) => Err(RecvTimeoutError::Timeout),
+        }
     }
 
-    fn wait_for_requests(&self, count: usize) -> Vec<String> {
+    async fn wait_for_requests(&mut self, count: usize) -> Vec<String> {
         let mut requests = Vec::new();
 
         for _ in 0..count {
-            match self.requests.recv_timeout(Duration::from_millis(1000)) {
+            match self.next_request_timeout().await {
                 Result::Ok(request) => requests.push(request),
                 Result::Err(err) => {
                     log::error!("{:?}", err);
@@ -372,16 +418,24 @@ impl HyperTestServer {
 
         requests
     }
-}
 
-impl Drop for HyperTestServer {
-    fn drop(&mut self) {
-        if let Some(shutdown) = self.shutdown.take() {
+    async fn terminate(mut self) {
+        if let Some(shutdown) = self.shutdown_send.take() {
             shutdown.send(()).unwrap();
         }
     }
 }
 
+#[derive(Debug)]
+enum RecvTimeoutError {
+    Closed,
+    Timeout,
+}
+
+struct Builder {
+    responses: Vec<Response<String>>,
+}
+
 impl Builder {
     fn response(mut self, status: StatusCode, body: impl ToString, retry_after: Option<DateTime<Utc>>) -> Self {
         let mut builder = Response::builder().status(status);
@@ -411,21 +465,20 @@ impl Builder {
     }
 
     fn create(self) -> HyperTestServer {
-        let (shutdown_sender, shutdown_receiver) = oneshot::channel::<()>();
-        let (url_sender, url_receiver) = bounded::<String>(1);
-        let (request_sender, request_receiver) = unbounded::<String>();
+        let (shutdown_send, shutdown_recv) = oneshot::channel();
+        let (request_sender, request_receiver) = mpsc::channel(100);
 
         let responses = Arc::new(self.responses);
         let counter = Arc::new(AtomicUsize::new(0));
 
         let make_service = make_service_fn(move |_| {
-            let request_sender = request_sender.clone();
+            let request_send = request_sender.clone();
             let counter = counter.clone();
             let responses = responses.clone();
 
             async move {
                 Ok::<_, hyper::Error>(service_fn(move |req: Request<Body>| {
-                    let request_sender = request_sender.clone();
+                    let request_send = request_send.clone();
                     let counter = counter.clone();
                     let responses = responses.clone();
 
@@ -435,7 +488,7 @@ impl Builder {
 
                         let mut content = String::default();
                         body.reader().read_to_string(&mut content).unwrap();
-                        request_sender.send(content).unwrap();
+                        request_send.send(content).await.expect("send request");
 
                         let count = counter.fetch_add(1, Ordering::AcqRel);
 
@@ -457,30 +510,24 @@ impl Builder {
             }
         });
 
-        std::thread::spawn(move || {
-            let rt = tokio::runtime::Runtime::new().expect("runtime");
-            rt.block_on(async move {
-                let server = Server::bind(&([0, 0, 0, 0], 0).into()).serve(make_service);
+        let server = Server::bind(&([0, 0, 0, 0], 0).into()).serve(make_service);
 
-                let url = format!("http://{}", server.local_addr());
-                url_sender.send(url).unwrap();
+        let url = format!("http://{}", server.local_addr());
 
-                let graceful = server.with_graceful_shutdown(async {
-                    shutdown_receiver.await.ok();
-                });
-
-                if let Err(e) = graceful.await {
-                    log::error!("server error: {}", e);
-                }
-            });
+        let graceful = server.with_graceful_shutdown(async {
+            shutdown_recv.await.ok();
         });
 
-        let url = url_receiver.recv().expect("url");
+        tokio::spawn(async move {
+            if let Err(e) = graceful.await {
+                log::error!("server error: {}", e);
+            }
+        });
 
         HyperTestServer {
             url,
-            requests: request_receiver,
-            shutdown: Some(shutdown_sender),
+            request_recv: request_receiver,
+            shutdown_send: Some(shutdown_send),
         }
     }
 }
diff --git a/appinsights/tests/telemetry.rs b/appinsights/tests/telemetry.rs
index a557181..f7444a9 100644
--- a/appinsights/tests/telemetry.rs
+++ b/appinsights/tests/telemetry.rs
@@ -7,8 +7,8 @@ use std::{
 use appinsights::{telemetry::SeverityLevel, TelemetryClient};
 use hyper::{Method, Uri};
 
-#[test]
-fn it_tracks_all_telemetry_items() {
+#[tokio::test]
+async fn it_tracks_all_telemetry_items() {
     let entries = Arc::new(RwLock::new(Vec::new()));
     logger::builder(entries.clone()).output(true).init();
 
@@ -38,9 +38,9 @@ fn it_tracks_all_telemetry_items() {
         true,
     );
 
-    ai.close_channel();
+    ai.close_channel().await;
 
-    logger::wait_until(&entries, "Successfully sent 6 items", Duration::from_secs(10));
+    logger::wait_until(&entries, "Successfully sent 6 items", Duration::from_secs(10)).await;
 }
 
 pub mod logger {
@@ -52,18 +52,19 @@ pub mod logger {
     use chrono::Utc;
     use log::{Level, LevelFilter, Log, Metadata, Record, SetLoggerError};
 
-    pub fn wait_until(entries: &Arc<RwLock<Vec<String>>>, msg: &str, panic_after: Duration) {
+    pub async fn wait_until(entries: &Arc<RwLock<Vec<String>>>, msg: &str, panic_after: Duration) {
         let panic_after = Utc::now() + chrono::Duration::from_std(panic_after).unwrap();
         loop {
             let entries = entries.read().unwrap();
             if entries.iter().any(|entry| entry.contains(msg)) {
                 break;
             }
+            drop(entries);
 
             if Utc::now() > panic_after {
                 panic!("Test took too long to finish");
             }
-            std::thread::sleep(Duration::from_millis(100))
+            tokio::time::sleep(Duration::from_millis(100)).await
         }
     }