Migrate Tokio from 0.1.22 to 0.2

[async]: https://docs.rs/tokio/0.1.22/tokio/prelude/enum.Async.html
[poll]: https://doc.rust-lang.org/std/task/enum.Poll.html
[fn_poll]: https://doc.rust-lang.org/std/future/trait.Future.html#tymethod.poll
[pin]: https://doc.rust-lang.org/std/pin/struct.Pin.html
[context]: https://doc.rust-lang.org/std/task/struct.Context.html
[old_poll]: https://docs.rs/tokio/0.1.22/tokio/prelude/type.Poll.html
[old_fn_poll]: https://docs.rs/tokio/0.1.22/tokio/prelude/trait.Future.html#tymethod.poll
[old_s_poll]: https://docs.rs/tokio/0.1.22/tokio/prelude/trait.Stream.html#tymethod.poll
[poll_next]: https://docs.rs/futures/0.3.0/futures/prelude/trait.Stream.html#tymethod.poll_next
[poll_ready]: https://docs.rs/futures/0.3.0/futures/prelude/trait.Sink.html#tymethod.poll_ready
[start_send]: https://docs.rs/futures/0.3.0/futures/prelude/trait.Sink.html#tymethod.start_send
[poll_flush]: https://docs.rs/futures/0.3.0/futures/prelude/trait.Sink.html#tymethod.poll_flush
[poll_close]: https://docs.rs/futures/0.3.0/futures/prelude/trait.Sink.html#tymethod.poll_close
[old_start_send]: https://docs.rs/tokio/0.1.22/tokio/prelude/trait.Sink.html#tymethod.start_send
[poll_complete]: https://docs.rs/tokio/0.1.22/tokio/prelude/trait.Sink.html#tymethod.poll_complete
[asyncsink]: https://docs.rs/tokio/0.1.22/tokio/prelude/enum.AsyncSink.html
[old_delay]: https://docs.rs/tokio/0.1.22/tokio/timer/struct.Delay.html
[delay]: https://docs.rs/tokio/0.2.2/tokio/time/struct.Delay.html
[delay_until]: https://docs.rs/tokio/0.2.2/tokio/time/fn.delay_until.html 
[delay_for]: https://docs.rs/tokio/0.2.2/tokio/time/fn.delay_for.html
[old_interval]: https://docs.rs/tokio/0.1.22/tokio/timer/struct.Interval.html
[interval]: https://docs.rs/tokio/0.2.2/tokio/time/struct.Interval.html
[fn_interval]: https://docs.rs/tokio/0.2.2/tokio/time/fn.interval.html
[interval_at]: https://docs.rs/tokio/0.2.2/tokio/time/fn.interval_at.html
[task]: https://docs.rs/tokio/0.1.22/tokio/prelude/task/struct.Task.html
[waker]: https://doc.rust-lang.org/std/task/struct.Waker.html
[joinhandle]: https://docs.rs/tokio/0.2.2/tokio/task/struct.JoinHandle.html
[shutdown_on_idle]: https://docs.rs/tokio/0.1.22/tokio/runtime/struct.Runtime.html#method.shutdown_on_idle
[spawn]: https://docs.rs/tokio/0.2.2/tokio/runtime/struct.Runtime.html#method.spawn
[executor]: https://docs.rs/tokio/0.1.22/tokio/runtime/struct.Runtime.html#method.executor
[taskexecutor]: https://docs.rs/tokio/0.1.22/tokio/runtime/struct.TaskExecutor.html
[handle]: https://docs.rs/tokio/0.2.2/tokio/runtime/struct.Runtime.html#method.handle
[reactor]: https://docs.rs/tokio/0.1.22/tokio/runtime/struct.Runtime.html#method.reactor
[pollevented]: https://docs.rs/tokio/0.2.4/tokio/io/struct.PollEvented.html

## Tokio Changes Cheatsheet

| | Tokio 0.1.22 | Tokio 0.2 |
|---:|---|---|
| Poll Type | Result\<[Async][async]\<Item>, Error><br>or<br>[Poll][old_poll]\<Item, Error> | [Poll]\<Output> |
| Future Required Method | ***fn [poll][old_fn_poll]***(&mut self) -> [Poll][old_poll]\<Item, Error> | ***fn [poll][fn_poll]***(self: [Pin]<&mut Self>, cx: &mut [Context]) -> [Poll]\<Output> |
| Dependencies | futures v0.1.18~23<br>bytes v0.4.7<br>mio v0.6.14 | futures 0.3.0<br>bytes v0.5.0<br>mio v0.6.20 |
| Stream Required Method | ***fn [poll][old_s_poll]***(&mut self) -> [Poll][old_poll]\<Option\<Item>, Error>|***fn [poll_next]***(self: [Pin]\<&mut Self>, cx: &mut [Context]) -> [Poll]\<Option\<Self::Item>>|
| Sink Required Methods | ***fn [start_send][old_start_send]***(&mut self, item: SinkItem) -> Result<[AsyncSink]<SinkItem>, SinkError><br>***fn [poll_complete]***(&mut self) -> Result<[Async]<()>, SinkError> | ***fn [poll_ready]***(self: [Pin]\<&mut Self>, cx: &mut [Context]) -> [Poll]\<Result\<(), Error>><br> ***fn [start_send]***(self: [Pin]\<&mut Self>, item: Item) -> Result\<(), Error><br> ***fn [poll_flush]***(self: [Pin]\<&mut Self>, cx: &mut [Context]) -> [Poll]\<Result\<(), Error>><br> ***fn [poll_close]***(self: [Pin]<&mut Self>, cx: &mut [Context]) -> [Poll]\<Result\<(), Error>>|
| Interval Path[^1][^2]|[tokio::`timer`::Interval][old_interval]|[tokio::time::Interval][interval]|
| Delay Path[^3][^2] |[tokio::`timer`::Delay][old_delay]|[tokio::time::Delay][delay]|
| Execution Unit |[tokio::prelude::task::Task][task]| [std::task::Waker][waker]|
| Runtime Changes || - replace ***[reactor]\()*** with ***[handle]\()***<br> - remove ***[executor]\()*** and [TaskExecutor]<br>- remove ***[shutdown_on_idle]\()***<br> - ***[spawn]\()*** now returns [JoinHandle] future|

[^1]: [Interval][interval] in 0.2 now doesn't have a ***new*** method. You have to use [interval][fn_interval] or [interval_at].
[^2]: All tokio 0.1 `timer` items have been moved to `time` in 0.2.
[^3]: [Delay][delay] in 0.2 now doesn't have a ***new*** method. You have to use [delay_until] and [delay_for]

## Migrate in Action
Let's start from an example:

```rust
#![feature(test)]

extern crate test;
use futures::future;
use tokio::prelude::*;
use tokio::runtime::Runtime;
use tokio::sync::mpsc::unbounded_channel;

struct Dual<T> {
    r1: T,
    r2: T,
}

impl<E, T: Stream<Item = i32, Error = E>> Stream for Dual<T> {
    type Item = i32;
    type Error = E;
    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        let mut end = false;
        // let us poll both in naiive ways:
        match self.r1.poll()? {
            Async::Ready(Some(e)) => return Ok(Async::Ready(Some(e))),
            Async::Ready(None) => {
                end = true;
            }
            Async::NotReady => {}
        };
        match self.r2.poll()? {
            Async::Ready(Some(e)) => Ok(Async::Ready(Some(e))),
            Async::Ready(None) => {
                if end {
                    Ok(Async::Ready(None))
                } else {
                    Ok(Async::NotReady)
                }
            }
            Async::NotReady => Ok(Async::NotReady),
        }
    }
}

#[test]
fn test_dual_poll() {
    let mut rt = Runtime::new().unwrap();
    let (mut tx, rx) = unbounded_channel();
    let (mut txx, rxx) = unbounded_channel();
    rt.spawn(future::lazy(move || {
        for i in 0..100000_i32 {
            tx.try_send(i).unwrap();
        }
        Ok(())
    }));
    rt.spawn(future::lazy(move || {
        for i in 100000..200000_i32 {
            txx.try_send(i).unwrap();
        }
        Ok(())
    }));
    let dual = Dual { r1: rxx, r2: rx };
    let task = dual.take(200000).collect().map_err(|_| ()).and_then(|x| {
        assert_eq!(x.len(), 200000);
        Ok(())
    });
    rt.spawn(task);
    rt.shutdown_on_idle().wait().unwrap();
}
``` 

Let's start from rewriting old Stream to a newer one. First, we have to change from `poll()` into `poll_next()`.
Notice that `Error`s won't be exported anymore from the Stream unless you define type `Item` as `Result`. Normally, if we want to make it compatible with our old Stream, we'll give the new `Item` with the same type as our old `Item`.

So our code becomes:

```rust
impl<T: std::marker::Unpin + Stream<Item = i32>> Stream for Dual<T> {
    type Item = i32;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
```

To make use of the Pinned self, we could call `Pin::get_mut()` to get the mutable reference. And to poll on our `r1`, we have to pass in a pinned `r1` to the `poll_next()` function. Don't forget to pass the `Context` to `poll_next()`:

```rust
        let pinned = Pin::get_mut(self);
        match Pin::new(&mut pinned.r1).poll_next(cx) {
```

In tokio 0.2, future will return `Poll` enums. We should handle these three possibilities here:  

```rust
Poll::Ready(Some(item)) => {/* meaning we have one item.*/}
Poll::Ready(None) => { /* meaning Stream ends */}
Poll::Pending => { /* meaning Stream is not ready */}
```

Since our `r1` and `r2` have the same return type as our `Stream`, if there's one item exported, we could just return it. So now our `Stream` becomes:

```rust
impl<T: std::marker::Unpin + Stream<Item = i32>> Stream for Dual<T> {
    type Item = i32;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        let mut end = false;
        let pinned = Pin::get_mut(self);
        match Pin::new(&mut pinned.r1).poll_next(cx) {
            x @ Poll::Ready(Some(_)) => return x,
            Poll::Ready(None) => {
                end = true;
            }
            Poll::Pending => (),
        };

        match Pin::new(&mut pinned.r2).poll_next(cx) {
            x @ Poll::Ready(Some(_)) => return x,
            Poll::Ready(None) => {
                if end {
                    Poll::Ready(None)
                } else {
                    Poll::Pending
                }
            }
            Poll::Pending => Poll::Pending,
        }
    }
}
```

For tests, there's no much difference. What you need to do is to modify slightly on the use of `Runtime`. In tokio v0.1.22, `Runtime::spawn()` doesn't return any Object, so you cannot wait on a specific group of spawned `Futures`. Now in tokio v0.2, we have `JoinHandler` - another `Future` that waits for the spawned `Future`. With this `JoinHandler`, now we could use `async/await` syntax sugar to accomplish something like pthread join on our spawned `Future`s:  

```rust
let rt = Runtime::new().unwrap();
let j1 = rt.spawn(async move {/* any code snippet with async/await syntax */});
let j2 = rt.spawn(async move {/* any code snippet with async/await synatx */});
rt.block_on(async move {
    j1.await.expect("first spawn failed");
    j2.await.expect("second spawn failed");
});
```

After fixing several API changes in `unbounded_channel`,
The final code in 0.2 looks like:

```rust
#![feature(test)]

extern crate test;
use futures::prelude::*;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::runtime::Runtime;
use tokio::sync::mpsc::unbounded_channel;

struct Dual<T> {
    r1: T,
    r2: T,
}

impl<T: std::marker::Unpin + Stream<Item = i32>> Stream for Dual<T> {
    type Item = i32;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        let mut end = false;
        let pinned = Pin::get_mut(self);
        match Pin::new(&mut pinned.r1).poll_next(cx) {
            x @ Poll::Ready(Some(_)) => return x,
            Poll::Ready(None) => {
                end = true;
            }
            Poll::Pending => (),
        };

        match Pin::new(&mut pinned.r2).poll_next(cx) {
            x @ Poll::Ready(Some(_)) => return x,
            Poll::Ready(None) => {
                if end {
                    Poll::Ready(None)
                } else {
                    Poll::Pending
                }
            }
            Poll::Pending => Poll::Pending,
        }
    }
}

#[test]
fn test_dual_poll() {
    let mut rt = Runtime::new().unwrap();
    let (mut tx, rx) = unbounded_channel();
    let (mut txx, rxx) = unbounded_channel();
    let j1 = rt.spawn(async move {
        for i in 0..100000_i32 {
            tx.send(i).unwrap();
        }
    });

    let j2 = rt.spawn(async move {
        for i in 100000..200000_i32 {
            txx.send(i).unwrap();
        }
    });

    let dual = Dual { r1: rxx, r2: rx };
    let j3 = rt.spawn(async move {
        let v: Vec<i32> = dual.take(200000).collect().await;
        assert_eq!(v.len(), 200000);
    });
    rt.block_on(async move {
        j1.await.unwrap();
        j2.await.unwrap();
        j3.await.unwrap();
    });
}
```

## For Sink
In v0.1.22, we only need to define `start_send()` and `poll_complete()`. These two maps to `start_send()` and `poll_ready()` in v0.2. Other than these, we have to also define `poll_flush()` - maps to the provided method `flush()` in v0.1.22, and `poll_close()` - maps to privided method `close` in v0.1.22. Details for these APIs are quite similar to old ones.

### `enter()` in v0.2
If you met any issue with the error message of "thread 'main' panicked at 'no current reactor'", then it's time to use the [***enter()***](https://docs.rs/tokio/0.2.4/tokio/runtime/struct.Runtime.html#method.enter) function provided by `Runtime` and `Handle`. Let's first look at the implementation of Handle in tokio:

```rust
    /// Enter the runtime context
    pub fn enter<F, R>(&self, f: F) -> R
    where
        F: FnOnce() -> R,
    {
        self.blocking_spawner.enter(|| {
            let _io = io::set_default(&self.io_handle);

            time::with_default(&self.time_handle, &self.clock, || self.spawner.enter(f))
        })
    }
```

We could see before our function been called, the handler will be assigned (or reactor, in v0.1.22) to \_io. Before \_io been dropped, we'll have this `Handle` as the default reactor. Any function that relies on [PollEvented] and called outside of `spawn()` and `block__on()` should be called through this function.

For example, if we want to create a UdpSocket from std::net::UdpSocket, we will have:  

```rust
let udp = rt.enter(move || {UdpSocket::from_std(socket).unwrap()});
```