Forem: x1957

stop async-std task

x1957 — Wed, 22 Jan 2020 12:52:00 +0000

I have a situation:

connect to a websocket & send ping message via rest.(It's ridiculous, but the biggest cryptocurrency exchange Binance's websocket api require this).

fn main() {
    task::block_on(async {
        loop { // loop for handle reconnect
            connect to websocket
            task::spawn(async move {
                loop {
                    // delay 30 minutes
                    // rest put
                }
            }); // send rest ping/pong
            read the push message
            // when websocket disconnect finish this iteration
        }
    });
}

The problem is: when iteration finished, the spawn task will run forever.

We can use async-std channel here to recv the stop message and break the loop.

The new problem: When we delay this task, how to get notice that out loop iteration finished?

futures::select solved this problem.

use async_std::sync::{channel, Receiver};
use async_std::task;
use futures::future::{Fuse, FusedFuture, FutureExt};
use futures::select;
use std::time::Duration;

fn main() {
    task::block_on(async move {
        let mut idx: i32 = 0;
        loop {
            idx = idx + 1;
            let (_s, r) = channel::<i32>(1);
            let h = task::spawn(async move {
                loop {
                    let fut = async {
                        task::sleep(Duration::from_secs(2)).await;
                        println!("{} do something!", idx);
                    }.fuse();
                    let mut fut = Box::pin(fut);
                    let mut stop = Box::pin(r.recv().fuse());
                    select! {
                        _ = fut=> {},
                        _ = stop => break,
                    }
                }
            });

            task::sleep(Duration::from_secs(5)).await;
            println!("{} exit", idx);
        }
    });
}

nodejs event loop

x1957 — Mon, 02 Sep 2019 12:47:43 +0000

最近遇到个问题，在for loop里面调用http.get请求，测试发现，会慢个几ms然后请求才会发出去,代码如下：


async function run() {
    for (let i = 0; i < 100; i++) {
        for (let j = 0; j < 40; j++) {
            let local_ts = Date.now();
            let cid = 'cid_' + i * 1000 + j;
            let url = `http://127.0.0.1:8890/ts?id=${cid}&ts=${local_ts}`;
            http.get(url).then(ret => {
                let server_ts = ret.split(' ')[1];
                let finish_ts = Date.now();
                console.log(cid, ret, 'elapsed', finish_ts - local_ts, 'server_diff', server_ts - local_ts, 'finish_diff', finish_ts - server_ts);
            });
            console.log(cid, 'fired', local_ts, Date.now());
        }
        await Sleep(3000);
    }
}

run();

调用的是个http服务，传本地的timestamp给他，他返回server的timestamp，方便我们计算网络的latency。

这时候就非常疑惑了，为什么server收到我们发送的请求都要6-10ms，这中间网络肯定没这么慢。

怀疑是event loop的锅，首先看了下http的实现，在最后会调用一个process.nextTick

event loop里面分了很多个loop，代码可以参考

r = uv__loop_alive(loop);
  if (!r)
    uv__update_time(loop);

while (r != 0 && loop->stop_flag == 0) {
    uv__update_time(loop);
    uv__run_timers(loop);
    ran_pending = uv__run_pending(loop);
    uv__run_idle(loop);
    uv__run_prepare(loop);

    timeout = 0;
    if ((mode == UV_RUN_ONCE && !ran_pending) || mode == UV_RUN_DEFAULT)
      timeout = uv_backend_timeout(loop);

    uv__io_poll(loop, timeout);
    uv__run_check(loop);
    uv__run_closing_handles(loop);

    if (mode == UV_RUN_ONCE) {
      uv__update_time(loop);
      uv__run_timers(loop);
    }

    r = uv__loop_alive(loop);
    if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
      break;
}

而process.nextTick是在这里面任意一个loop结束后就执行的，所以我们的请求其实都堆积到了sleep那儿，因为http get都在for loop里面，所以这里我们如果要立马执行http get，需要切出我们的loop回到event loop. 在循环立马加入一个await Sleep(0)


async function run() {
    for (let i = 0; i < 100; i++) {
        for (let j = 0; j < 40; j++) {
            let local_ts = Date.now();
            let cid = 'cid_' + i * 1000 + j;
            let url = `http://127.0.0.1:8890/ts?id=${cid}&ts=${local_ts}`;
            http.get(url).then(ret => {
                let server_ts = ret.split(' ')[1];
                let finish_ts = Date.now();
                console.log(cid, ret, 'elapsed', finish_ts - local_ts, 'server_diff', server_ts - local_ts, 'finish_diff', finish_ts - server_ts);
            });
            console.log(cid, 'fired', local_ts, Date.now());
            await Sleep(0);
        }
        await Sleep(3000);
    }
}

run();

这时候堆积之后再请求的问题就解决了，不过问题还是有，一毫秒左右才一个请求，感觉太慢。

PS. Sleep代码

const Promise = require('bluebird');
module.exports = function(duration) {
    return new Promise(resolve => {
        setTimeout(() => {
            resolve(null);
        }, duration);
    });
}

使用指定glibc

x1957 — Fri, 23 Aug 2019 04:35:58 +0000

使用Rust在编译机器上copy过来之后，运行报错: "GLIBC_2.18" not found

通过 strings /lib64/libc.so.6 | grep GLIBC_ 命令查看发现这台机器上只有2.17，开始本来想升级glibc，但是这个东西确实太底层了，怕影响其他东西，所以决定编译到自定义目录然后指定加载就行。

编译流程很简单 http://ftp.gnu.org/gnu/glibc 下载一个需要的版本，我这儿就选了2.18。

解压后新建一个目录glibc-build(因为它不让在源代码目录build)，和一个我们放编译好的glibc目录，然后在build目录里面 ../glibc-xxx/configure --prefix=../glibc/

configure好之后 make make install

我们编译好的glibc就在glibc这个目录啦，运行是这样

RUST_LOG=info ./glibc/lib/ld-linux-x86-64.so.2  --library-path ./glibc/lib:/lib64:/usr/lib64 ./order-server

Rust failure

x1957 — Sat, 20 Jul 2019 13:45:10 +0000

大家都建议error handling用failure这个库，看了看（：

先贴代码

use failure::{Error, Fail};
use reqwest;

#[derive(Fail, Debug)]
enum MyError {
    #[fail(display = "filed {} not exist", _0)]
    FileNotExist(String),
    // #[fail(display = "http error: {}", _0)]
    // HttpError(#[fail(cause)] reqwest::Error),
}

fn http_get(url: Option<String>) -> std::result::Result<(), Error> {
    if url == None {
        // just test
        return Err(MyError::FileNotExist("url".to_string()).into());
    }
    let client = reqwest::Client::new();
    let result = client.get(&url.unwrap()).send()?.text()?;
    println!("{}", result);
    Ok(())
}

fn main() {
    http_get(Some("https://www.baidu.com".to_string())).unwrap();
}

我们可以用failure::Error来替换掉std::error::Error，我们自己的error呢就去实现Fail这个trait，这里我们用了他的过程宏，让它给我们默认实现了，可以cargo expand展开看下

#[allow(non_upper_case_globals)]
#[doc(hidden)]
const _DERIVE_failure_Fail_FOR_MyError: () = {
    impl ::failure::Fail for MyError {
        fn name(&self) -> Option<&str> {
            Some("fail::MyError")
        }
        #[allow(unreachable_code)]
        fn cause(&self) -> ::failure::_core::option::Option<&dyn ::failure::Fail> {
            match *self {
                MyError::FileNotExist(ref __binding_0) => return None,
            }
            None
        }
        #[allow(unreachable_code)]
        fn backtrace(&self) -> ::failure::_core::option::Option<&::failure::Backtrace> {
            match *self {
                MyError::FileNotExist(ref __binding_0) => return None,
            }
            None
        }
    }
};

其实遇到的最大的问题不是这个，而是，我

return Err(MyError::FileNotExist("url".to_string()));

的时候要报错说要的是failure::failure::Error而我返回的是这个我自定义的Error，但是不是说好的是实现了Fail接口就行吗？文档也没写，又来在某个issue里面发现，说要么into()，要么?

所以得写成

return Err(MyError::FileNotExist("url".to_string()).into());

或者

Err(MyError::FileNotExist("url".to_string()))?;

Rust existential type

x1957 — Fri, 19 Jul 2019 11:08:09 +0000

最近在封装一点lib，本来想在trait里面定义async fn的，但是目前来说是不允许的.Async Methods I: generic associated types，只能想其他办法咯，那我们就再调用async fn返回个future吧，也一样的。

然后问题又来了，本来想是return 一个 impl Future，但是。。。trait里面又不允许，sad！

Async Methods I: generic associated types这里也解释了为什么直接associated types不行。

原因就是async返回的Future是个闭包，把所有东西的lifetime都丢进去了，如果我们要写associated types需要这样:

trait Foo {
    type _Future<'a>: Future<Output = i32> + 'a;
    fn foo_method<'a>(&'a self) -> Self::_Future<'a>;
}

但是呢。。。问题又来了，目前associated types不支持lifetime参数，这个东西叫generic associated types

不过还好，我们还有existential type，参考RFC

#![feature(async_await, existential_type)]

use futures::compat::Future01CompatExt;
use futures::compat::Stream01CompatExt;
use futures::future::{FutureExt, TryFutureExt};
use futures::stream::{StreamExt, TryStreamExt};
use futures::Future;
use reqwest::r#async::{Client, ClientBuilder, Decoder};
use std::io::{self, Write};

trait Fuck {
    type AsyncResult: Future<Output=()>;
    fn download(&self) -> Self::AsyncResult;
}

struct Down;

impl Down {
    pub fn new() -> Self{
        Down{}
    }
}

impl Fuck for Down {
    existential type AsyncResult: Future<Output=()>;
    fn download(&self) -> Self::AsyncResult {
        download()
    }
}

async fn download() {
    let client = ClientBuilder::new().build().unwrap();
    let res = client
        .get("https://dev.to/x1957/rust-async-await-2l4c")
        .send()
        .compat()
        .await;
    let mut body = res.unwrap().into_body().compat();
      while let Some(next) = body.next().await {
        let chunk = next.unwrap();
        io::stdout().write_all(&chunk).unwrap();
    }
    println!("\nDone");
}

fn call<T: Fuck>(x: T)->T::AsyncResult {
    x.download()
}

fn main() {
    // let fut = download().unit_error().boxed().compat();
    let fut = call(Down::new()).unit_error().boxed().compat();
    tokio::run(fut);
}

我们定义了Fuck这个trait里面有个associated type AsyncResult，在我们impl这个trait的使用通过existential type来指出AsyncResult是impl Future的即可。

现在使用的时候需要打开feature existential_type

Rust async await (2)

x1957 — Tue, 16 Jul 2019 06:21:33 +0000

Rust async await (1) 里面我们提到了简单的async await目前的语法，想用于实战试试。

我们用异步http请求试试，依赖如下:

[dependencies]
tokio = "0.1.22"
futures-preview = { version = "=0.3.0-alpha.17", features = ["compat"] }
reqwest = "0.9.18"

#![feature(async_await)]

use futures::compat::Future01CompatExt;
use futures::compat::Stream01CompatExt;
use futures::future::{FutureExt, TryFutureExt};
use futures::stream::{StreamExt, TryStreamExt};
use futures::Future;
use reqwest::r#async::{Client, ClientBuilder, Decoder};
use std::io::{self, Write};

async fn download() {
    let client = ClientBuilder::new().build().unwrap();
    let res = client
        .get("https://dev.to/x1957/rust-async-await-2l4c")
        .send()
        .compat()
        .await;
    let mut body = res.unwrap().into_body().compat();
      while let Some(next) = body.next().await {
        let chunk = next.unwrap();
        io::stdout().write_all(&chunk).unwrap();
    }
    println!("\nDone");
}

fn main() {
    let fut = download().unit_error().boxed().compat();
    tokio::run(fut);
}

需要注意几点：
1、use reqwest::r#async::{Client, ClientBuilder, Decoder} 因为reqwest还没有升级，之前的async现在已经是关键字了，所以得用r#
2、.send之后的future是futures01需要compat
3、into_body之后的stream是stream01需要compat

Rust async await (1)

x1957 — Mon, 15 Jul 2019 07:08:54 +0000

Rust version:

rustc 1.38.0-nightly (69656fa4c 2019-07-13)

打开async/await的future:

#![feature(async_await)]

#![feature(async_await)]

async fn get()->i32 {
    println!("start");
    std::thread::sleep(std::time::Duration::from_millis(1000));
    1
}

async fn p() {
    let n = get().await;
    println!("n = {}", n);
}

fn main() {
    let fut = p();
    println!("called p");
    futures::executor::block_on(fut);
}

定义了一个简单的get函数，返回1

p里面调用get并打印1

async和JavaScript的类似，返回的是future(js的是promise)，所以在Rust里面我们还需要一个executor来Run这个future，这里我们future用的版本是:

futures-preview = { version = "=0.3.0-alpha.17", features = ["compat"] }

主要原因就是async返回的是std::future::Future,而我们大部分三方库还是用的futures这个库，比如tokio，我们用tokio::run来试试会有个报错

  --> src/main.rs:21:5
   |
21 |     tokio::run(fut);
   |     ^^^^^^^^^^ the trait `futures::future::Future` is not implemented for `impl std::future::Future`
   |
   = note: required by `tokio::runtime::threadpool::run`

error: aborting due to previous error

For more information about this error, try `rustc --explain E0277`.
error: Could not compile `async-await`.

To learn more, run the command again with --verbose.

tokio::run 是给 futures::future::Future,但是我们 async 的是 std::future::Future

现在最大的问题就是三方库还没有支持std future，所以如果要用就得compat一下

#![feature(async_await)]

use futures::compat::Future01CompatExt;
use futures::future::{FutureExt, TryFutureExt};

async fn get()->i32 {
    println!("start");
    std::thread::sleep(std::time::Duration::from_millis(1000));
    1
}

async fn p() {
    let n = get().await;
    println!("n = {}", n);
}

fn main() {
    let fut = p().unit_error().boxed().compat();
    println!("called p");
    tokio::run(fut);
}

Rust future(stream) timeout

x1957 — Wed, 12 Jun 2019 15:55:44 +0000

打开tokio-timer可以看到timout

#[must_use = "futures do nothing unless polled"]
#[derive(Debug)]
pub struct Timeout<T> {
    value: T,
    delay: Delay,
}

很简单的结构，value就是我们要加timeout的future或者stream，delay就是多久之后就timeout（：

看看他实现的Future和Stream trait

impl<T> Future for Timeout<T>
where
    T: Future,
{
    type Item = T::Item;
    type Error = Error<T::Error>;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        // First, try polling the future
        match self.value.poll() {
            Ok(Async::Ready(v)) => return Ok(Async::Ready(v)),
            Ok(Async::NotReady) => {}
            Err(e) => return Err(Error::inner(e)),
        }

        // Now check the timer
        match self.delay.poll() {
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Ok(Async::Ready(_)) => Err(Error::elapsed()),
            Err(e) => Err(Error::timer(e)),
        }
    }
}

impl<T> Stream for Timeout<T>
where
    T: Stream,
{
    type Item = T::Item;
    type Error = Error<T::Error>;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        // First, try polling the future
        match self.value.poll() {
            Ok(Async::Ready(v)) => {
                if v.is_some() {
                    self.delay.reset_timeout();
                }
                return Ok(Async::Ready(v));
            }
            Ok(Async::NotReady) => {}
            Err(e) => return Err(Error::inner(e)),
        }

        // Now check the timer
        match self.delay.poll() {
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Ok(Async::Ready(_)) => {
                self.delay.reset_timeout();
                Err(Error::elapsed())
            }
            Err(e) => Err(Error::timer(e)),
        }
    }

都是正常去poll value，如果ready了就直接返回，如果没有ready，那么再去poll dealy看看是否timeout，如果没有timeout返回notready，如果timeout了那就返回time elapsed error.

非常简单的逻辑实现了timeout（：

如果我要方便的使用timeout可能需要use tokio::prelude::FutureExt;

fn timeout(self, timeout: Duration) -> Timeout<Self>
    where
        Self: Sized,
    {
        Timeout::new(self, timeout)
    }

其实自己用Timeout::new也可以，但是xxx.timeout看着舒服一些

Cargo build script

x1957 — Tue, 30 Apr 2019 14:25:55 +0000

试用了下tower-grpc，作为萌新很多东西不知道，以前用这种都是protoc ....先生成代码，看了下example，结果就没发现哪儿有生成代码的过程。仔细看了下

pub mod hello_world {
    include!(concat!(env!("OUT_DIR"), "/helloworld.rs"));
}

use hello_world::{server, HelloReply, HelloRequest};

在 hello_world mod里面用 include! 把生成的代码给搞了进来。

ok，我们继续纠结生成的代码在哪儿？怎么生成的？

发现目录里面有个 build.rs 萌新虽然不了解这个，但是看他名字就知道和编译相关了，于是Google了一下
https://doc.rust-lang.org/cargo/reference/build-scripts.html

其实就是在build的时候执行的东西，还能指定build-dependencies什么的。简单看了下，这里面就是编译了.proto，然后输出到了OUT_DIR，所以上面的include！就能用了。

Protobuf log

x1957 — Fri, 26 Apr 2019 13:57:45 +0000

我有一些数据是mq定时推送的，现在要在他们基础上做模型，之前都是收到就过了，没存过。
推送过来的数据是protobuf序列化过的，想也没啥必要反序列化了再打log。

方案1：
直接log，
pb1pb2pb3pb4pb5....pbn
最后发现，解析不了，开始想法是美好的，毕竟就那么几个field，读了就好了呀，但是后来发现因为pb可能会处理optional的字段，那就不知道一个message有多长啦。

方案2：
加换行符
pb1\n
pb2\n
pb3\n
...
pbn\n
这里就是自己傻逼了，pb里面可能会包含换行符等乱七八糟的东西（毕竟binary），所以不可行。

方案3：
按照之前tcp发送数据包那样的写法，在头部放个length
length1_pb1 length2_pb2 ... lengthn_pbn
这样我们解析的时候就先读有多少个bytes，然后拿这些bytes来反序列化pb messsage。

并且我们length最要也用pb编码，免得处理不一致

c.log.Write(proto.EncodeVarint(uint64(len(bs))))
c.log.Write(bs)

然后这边我们处理数据用的Rust

protobuf 这个库看用的人最多，并且在他的 CodedInputStream read_message正好就是处理我们这种结构的数据的.

    pub fn read_message<M : Message>(&mut self) -> ProtobufResult<M> {
        let mut r: M = Message::new();
        self.merge_message(&mut r)?;
        r.check_initialized()?;
        Ok(r)
    }

Rust处理pb的代码如下

mod message;
mod file;

#[macro_use]
extern crate slog;
extern crate slog_async;
extern crate slog_term;

use slog::Drain;

use clap::{App, Arg};
use message::Message;
use protobuf::CodedInputStream;
use std::fs;
use std::io::BufReader;

fn vist_dir(path: &str) {}

fn main() -> Result<(), Box<std::error::Error>> {
    let matches = App::new("pig data process")
        .version(env!("CARGO_PKG_VERSION"))
        .author(env!("CARGO_PKG_AUTHORS"))
        .arg(
            Arg::with_name("dir")
                .long("dir")
                .short("d")
                .help("dir of data")
                .value_name("DIR")
                .takes_value(true),
        )
        .get_matches();

    let decorator = slog_term::TermDecorator::new().build();
    let drain = slog_term::FullFormat::new(decorator).build().fuse();
    let drain = slog_async::Async::new(drain).build().fuse();
    let log = slog::Logger::root(drain, o!());

    let dir = matches.value_of("dir").expect("need --dir parameter");
    info!(log, "dir = {}", dir);

    let mut files = file::read_dir(dir);
    files.sort_by(|a, b| {
        a.file_name()
            .into_string()
            .unwrap()
            .cmp(&b.file_name().into_string().unwrap())
    });
    info!(log, "{:?}", files);
    let mut cnt = 0;
    for file in &files {
        let fd = fs::File::open(file.path()).unwrap();
        info!(log, "read {}", file.file_name().into_string().unwrap());
        let mut buffer = BufReader::new(fd);
        let mut is = CodedInputStream::new(&mut buffer);
        while let Ok(message) = is.read_message::<Message>(){
//            info!(log, "{:?}", message);
            cnt+=1;
        }
    }
    info!(log, "total = {}", cnt);
    Ok(())
}

Rust Future then map difference

x1957 — Wed, 10 Apr 2019 13:55:00 +0000

一直好奇and_then, map有什么区别，感觉都是输入一个fn，然后返回一个Future。

看看文档就发现了

and_then

fn and_then<F, B>(self, f: F) -> AndThen<Self, B, F> 
where F: FnOnce(Self::Item) -> B,
        B: IntoFuture<Error=Self::Error>,
        Self: Sized
[−]
Execute another future after this one has resolved successfully.

This function can be used to chain two futures together and ensure that the final future isn't resolved until both have finished. The closure provided is yielded the successful result of this future and returns another value which can be converted into a future.

Note that because Result implements the IntoFuture trait this method can also be useful for chaining fallible and serial computations onto the end of one future.

If this future is dropped, panics, or completes with an error then the provided closure f is never called.

Note that this function consumes the receiving future and returns a wrapped version of it.

Examples
use futures::future::*;

let future_of_1 = ok::<u32, u32>(1);
let future_of_4 = future_of_1.and_then(|x| {
    Ok(x + 3)
});

let future_of_err_1 = err::<u32, u32>(1);
future_of_err_1.and_then(|_| -> FutureResult<u32, u32> {
    panic!("should not be called in case of an error");
});

map

fn map<F, U>(self, f: F) -> Map<Self, F> 
where F: FnOnce(Self::Item) -> U, Self: Sized
[−]
Map this future's result to a different type, returning a new future of the resulting type.

This function is similar to the Option::map or Iterator::map where it will change the type of the underlying future. This is useful to chain along a computation once a future has been resolved.

The closure provided will only be called if this future is resolved successfully. If this future returns an error, panics, or is dropped, then the closure provided will never be invoked.

Note that this function consumes the receiving future and returns a wrapped version of it, similar to the existing map methods in the standard library.

Examples
use futures::future::*;

let future_of_1 = ok::<u32, u32>(1);
let future_of_4 = future_of_1.map(|x| x + 3);

可以看到and_then的f返回的future应该是个unresolved的，map返回的是个solved的。

看看代码

#[derive(Debug)]
#[must_use = "futures do nothing unless polled"]
pub struct AndThen<A, B, F> where A: Future, B: IntoFuture {
    state: Chain<A, B::Future, F>,
}

pub fn new<A, B, F>(future: A, f: F) -> AndThen<A, B, F>
    where A: Future,
          B: IntoFuture,
{
    AndThen {
        state: Chain::new(future, f),
    }
}

impl<A, B, F> Future for AndThen<A, B, F>
    where A: Future,
          B: IntoFuture<Error=A::Error>,
          F: FnOnce(A::Item) -> B,
{
    type Item = B::Item;
    type Error = B::Error;

    fn poll(&mut self) -> Poll<B::Item, B::Error> {
        self.state.poll(|result, f| {
            result.map(|e| {
                Err(f(e).into_future())
            })
        })
    }
}

#[derive(Debug)]
#[must_use = "futures do nothing unless polled"]
pub struct Map<A, F> where A: Future {
    future: A,
    f: Option<F>,
}

pub fn new<A, F>(future: A, f: F) -> Map<A, F>
    where A: Future,
{
    Map {
        future: future,
        f: Some(f),
    }
}

impl<U, A, F> Future for Map<A, F>
    where A: Future,
          F: FnOnce(A::Item) -> U,
{
    type Item = U;
    type Error = A::Error;

    fn poll(&mut self) -> Poll<U, A::Error> {
        let e = match self.future.poll() {
            Ok(Async::NotReady) => return Ok(Async::NotReady),
            Ok(Async::Ready(e)) => Ok(e),
            Err(e) => Err(e),
        };
        e.map(self.f.take().expect("cannot poll Map twice"))
         .map(Async::Ready)
    }
}

map的实现比较简单，当future ready之后把返回值用f处理后返回新的Async::Ready包装后的Result. 不过and_then稍微复杂了点，多了个chain，我们看看chain的代码

#[derive(Debug)]
pub enum Chain<A, B, C> where A: Future {
    First(A, C),
    Second(B),
    Done,
}

impl<A, B, C> Chain<A, B, C>
    where A: Future,
          B: Future,
{
    pub fn new(a: A, c: C) -> Chain<A, B, C> {
        Chain::First(a, c)
    }

    pub fn poll<F>(&mut self, f: F) -> Poll<B::Item, B::Error>
        where F: FnOnce(Result<A::Item, A::Error>, C)
                        -> Result<Result<B::Item, B>, B::Error>,
    {
        let a_result = match *self {
            Chain::First(ref mut a, _) => {
                match a.poll() {
                    Ok(Async::NotReady) => return Ok(Async::NotReady),
                    Ok(Async::Ready(t)) => Ok(t),
                    Err(e) => Err(e),
                }
            }
            Chain::Second(ref mut b) => return b.poll(),
            Chain::Done => panic!("cannot poll a chained future twice"),
        };
        let data = match mem::replace(self, Chain::Done) {
            Chain::First(_, c) => c,
            _ => panic!(),
        };
        match f(a_result, data)? {
            Ok(e) => Ok(Async::Ready(e)),
            Err(mut b) => {
                let ret = b.poll();
                *self = Chain::Second(b);
                ret
            }
        }
    }
}

可以看到当a.poll执行完之后，拿到之前的fn，然后执行f(a_result, data)，f是之前and_then里面的的函数

|result, f| {
    result.map(|e| {
        Err(f(e).into_future())
})

所以chain里面的match这里一定是个Err啦，这里的b就是f(e)之后生成的Future，并且这里改变了Chain的状态，Chain::Second(b)，如果这里b.poll()返回Ready那么就结束了，如果是NotReady，那么下次poll的时候再match *self，就是b了，就直接poll b。

impl Trait for Fn

x1957 — Sun, 31 Mar 2019 15:23:33 +0000

用ws-rs的时候，看demo：

connect(HUOBI, |out| {
        out.send(r#"{"event":"addChannel","channel":"ltcbtc_depth"}"#);

        move |msg| {
            println!("Client got message '{}'. ", msg);
            out.close(CloseCode::Normal)
        }
    });

后面返回个 closure，开始以为是啥奇怪的语法呢，msg就拿到了服务端发送的东西，好神奇！

看connect代码:

pub fn connect<U, F, H>(url: U, factory: F) -> Result<()>
where
    U: Borrow<str>,
    F: FnMut(Sender) -> H,
    H: Handler,
{
    let mut ws = WebSocket::new(factory)?;
    let parsed = url::Url::parse(url.borrow()).map_err(|err| {
        Error::new(
            ErrorKind::Internal,
            format!("Unable to parse {} as url due to {:?}", url.borrow(), err),
        )
    })?;
    ws.connect(parsed)?;
    ws.run()?;
    Ok(())
}

我们输入的应该是个 factory，然后通过factory创建了ws。
我们再看看factory定义

/// A trait for creating new WebSocket handlers.
pub trait Factory {
    type Handler: Handler;

    fn connection_made(&mut self, _: Sender) -> Self::Handler;

    #[inline]
    fn on_shutdown(&mut self) {
        debug!("Factory received WebSocket shutdown request.");
    }

    #[inline]
    fn client_connected(&mut self, ws: Sender) -> Self::Handler {
        self.connection_made(ws)
    }

    #[inline]
    fn server_connected(&mut self, ws: Sender) -> Self::Handler {
        self.connection_made(ws)
    }

    #[inline]
    fn connection_lost(&mut self, _: Self::Handler) {}
}

但是我们输入的是个closure啊？？？

再看看我们输入函数在where的限定

F: FnMut(Sender) -> H
H: Handler

输入是个Sender，返回是个Handler，所以我们demo里面的out是个Sender，返回的

move |msg| {
            println!("Client got message '{}'. ", msg);
            out.close(CloseCode::Normal)
        }

是个Handler.

继续疑问：我输入是个函数呀，怎么就符合Factory这个呢

看看Factory trait里面

impl<F, H> Factory for F
where
    H: Handler,
    F: FnMut(Sender) -> H,
{
    type Handler = H;

    fn connection_made(&mut self, out: Sender) -> H {
        self(out)
    }
}

他给 FnMut(Sender) -> H实现了Factory trait，connection_made也是Factory trait里面唯一没有默认实现的，所以当connection创建之后，就调用我们的这个函数，把out给他，并返回一个Handler。

OK到这儿就只有一个问题了，我们返回的是个closuzre啊，同理。。。也是给这个closuzre实现了Handler trait，看看代码

impl<F> Handler for F
where
    F: Fn(Message) -> Result<()>,
{
    fn on_message(&mut self, msg: Message) -> Result<()> {
        self(msg)
    }
}

同上，on_message也是Handler唯一没有默认实现的了，所以这里这个等于

        move |msg| {
            println!("Client got message '{}'. ", msg);
            out.close(CloseCode::Normal)
        }