client.rs

    let client = || {
        connect_async(Url::parse(url).unwrap(), handle.remote().clone())
        // .retry(|_| RetryPolicy::Repeat)
        .map_err(|e| eprintln!("connect_async: {}", e))
        .and_then(|(ws_stream, _)| {
            println!("Connected to {}", url);

            let (sink, stream) = ws_stream.split();
            sink.send(Message::Text(subscribe.to_string()))
            .map_err(|e| eprintln!("sink send: {}", e))
            .and_then(|_| {
                println!("subscribe sent: {}", subscribe);

                let timer = Interval::new(Instant::now(), Duration::from_secs(1)).for_each(|_| {
                    println!("ttt");
                    Ok(())
                }).map_err(|_| ());

                let q_insert = r.db(DB).table(TBL.to_owned());
                let flow = stream.for_each(move |msg| {
                    let mut json: Value = serde_json::from_str(msg.to_text().unwrap()).unwrap();
                    transform_numbers(&mut json);

                    q_insert.insert(json).run::<ServerStatus>(conn)
                    .unwrap().wait().next();
                    // .map_err(|err| println!("{:?}", err) );
                    // handle.spawn(f2.into_future());
                    Ok(())
                }).map_err(|_| ());

                let f = flow.select(timer);
                f.map(|_| ()).map_err(|_| ())
            })
        })
    };