chore(file sink): Upgrade file sink for tokio-compat (#1988)

* Revert "Disable file sink and files bench"

This reverts commit dd5c922.

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Upgrade file sink for tokio 0.2

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Further StreamingSink to refactoring

- switch StreamingSink to impl Stream - provides a significant boot to
  composability;
- simpler compat layer - this compat approach should be much easier to
  understand, it consists of a set of simple and documented components
  that are composed together in a meaningful way;
- switch tests to directly working with FileSink instead of futures
  compat layer - this build upon on StreamingSink taking impl Stream
  instead of mpsc::Receiver.

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Corrected tracing

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Extracted event processing logic to a separate function

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Switch from futures::future::select to tokio02::select!

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Make ExpiringHashMap Unpin

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Move ?Sized to where clause

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Simplified the ExpiringHashMap API

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Unify item kind at ExpiringHashMap::remove and ExpiringHashMap::poll_expired

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Move ExpiringHashMap to expiring_hash_map/mod.rs

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Add some tests

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Add insert_waker, make tests pass

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Add another test

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Improved value assertion test

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Add wake count tests

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Simplify compat, get rid of UnitTypeErrorSink01 in favor of sink_map_err

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Eliminate useless log message

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Switch FileSink to impl Future instead of Stream

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Upgrade tokio for sync::Notify

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Implemented an API that we settled on with Lucio

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Add docs

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Better comment at the example

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Make StreamingSink::run's input 'static

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Remove .get_ref() from ?path

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Correct typos and grammar errors at the docs

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Documented the unreachable! branch

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Move expiring_hash_map tests mod into it's main file

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Compacted nesting

Signed-off-by: MOZGIII <mike-n@narod.ru>

* Fix the typos at the streaming sink compat docs

Signed-off-by: MOZGIII <mike-n@narod.ru>

Cargo.toml

@@ -45,12 +45,13 @@ tracing-limit = { path = "lib/tracing-limit" }
 futures01 = { package = "futures", version = "0.1.25" }
 futures = { version = "0.3", default-features = false, features = ["compat"] }
 tokio = { version = "0.1.22", features = ["io", "uds", "tcp", "rt-full", "experimental-tracing"], default-features = false }
-tokio02 = { package = "tokio", version = "0.2", features = ["blocking"] }
+tokio02 = { package = "tokio", version = "0.2.13", features = ["blocking", "fs", "sync", "macros", "test-util", "rt-core"] }
 tokio-codec = "0.1.0"
 tokio-openssl = "0.3.0"
 tokio-retry = "0.2.0"
 tokio-signal = "0.2.7"
 tokio-compat = { version = "0.1", features = ["rt-full"] }
+async-trait = "0.1"
 
 # Tracing
 tracing = "0.1.9"
@@ -178,6 +179,7 @@ trust-dns-server = "0.17.0"
 trust-dns = "0.17.0"
 trust-dns-proto = "0.8.0"
 dirs = "2.0.2"
+tokio-test = "0.2"
 
 [features]
 # Default features for *-unknown-linux-gnu and *-apple-darwin

benches/bench.rs

@@ -36,7 +36,7 @@ criterion_main!(
     buffering::buffers,
     http::http,
     batch::batch,
-    /* files::files, */
+    files::files,
     lua::lua,
     event::event
 );

benches/files.rs

@@ -1,5 +1,3 @@
-#![cfg(feature = "disabled")]
-
 use bytes::Bytes;
 use criterion::{criterion_group, Benchmark, Criterion, Throughput};
 use futures01::{sink::Sink, stream::Stream, Future};