futures-preview 0.3.0-alpha.9

use futures::channel::oneshot;
use futures::future::{ok, err};
use std::sync::mpsc;
use std::thread;

mod support;
use support::*;


#[test]
fn join1() {
    let (tx, rx) = mpsc::channel();
    ok::<i32, i32>(1).join(ok(2))
                           .map(move |v| tx.send(v).unwrap())
                           .forget();
    assert_eq!(rx.recv(), Ok((1, 2)));
    assert!(rx.recv().is_err());
}

#[test]
fn join2() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.join(p2).map(move |v| tx.send(v).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    c1.send(1).unwrap();
    assert!(rx.try_recv().is_err());
    c2.send(2).unwrap();
    assert_eq!(rx.recv(), Ok((1, 2)));
    assert!(rx.recv().is_err());
}

#[test]
fn join3() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.join(p2).map_err(move |_v| tx.send(1).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    drop(c1);
    assert_eq!(rx.recv(), Ok(1));
    assert!(rx.recv().is_err());
    drop(c2);
}

#[test]
fn join4() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.join(p2).map_err(move |v| tx.send(v).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    drop(c1);
    assert!(rx.recv().is_ok());
    drop(c2);
    assert!(rx.recv().is_err());
}

#[test]
fn join5() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (c3, p3) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.join(p2).join(p3).map(move |v| tx.send(v).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    c1.send(1).unwrap();
    assert!(rx.try_recv().is_err());
    c2.send(2).unwrap();
    assert!(rx.try_recv().is_err());
    c3.send(3).unwrap();
    assert_eq!(rx.recv(), Ok(((1, 2), 3)));
    assert!(rx.recv().is_err());
}

#[test]
fn select1() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.select(p2).map(move |v| tx.send(v).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    c1.send(1).unwrap();
    let (v, p2) = rx.recv().unwrap().into_inner();
    assert_eq!(v, 1);
    assert!(rx.recv().is_err());

    let (tx, rx) = mpsc::channel();
    p2.map(move |v| tx.send(v).unwrap()).forget();
    c2.send(2).unwrap();
    assert_eq!(rx.recv(), Ok(2));
    assert!(rx.recv().is_err());
}

#[test]
fn select2() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.select(p2).map_err(move |v| tx.send((1, v.into_inner().1)).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    drop(c1);
    let (v, p2) = rx.recv().unwrap();
    assert_eq!(v, 1);
    assert!(rx.recv().is_err());

    let (tx, rx) = mpsc::channel();
    p2.map(move |v| tx.send(v).unwrap()).forget();
    c2.send(2).unwrap();
    assert_eq!(rx.recv(), Ok(2));
    assert!(rx.recv().is_err());
}

#[test]
fn select3() {
    let (c1, p1) = oneshot::channel::<i32>();
    let (c2, p2) = oneshot::channel::<i32>();
    let (tx, rx) = mpsc::channel();
    p1.select(p2).map_err(move |v| tx.send((1, v.into_inner().1)).unwrap()).forget();
    assert!(rx.try_recv().is_err());
    drop(c1);
    let (v, p2) = rx.recv().unwrap();
    assert_eq!(v, 1);
    assert!(rx.recv().is_err());

    let (tx, rx) = mpsc::channel();
    p2.map_err(move |_v| tx.send(2).unwrap()).forget();
    drop(c2);
    assert_eq!(rx.recv(), Ok(2));
    assert!(rx.recv().is_err());
}

#[test]
fn select4() {
    let (tx, rx) = mpsc::channel::<oneshot::Sender<i32>>();

    let t = thread::spawn(move || {
        for c in rx {
            c.send(1).unwrap();
        }
    });

    let (tx2, rx2) = mpsc::channel();
    for _ in 0..10000 {
        let (c1, p1) = oneshot::channel::<i32>();
        let (c2, p2) = oneshot::channel::<i32>();

        let tx3 = tx2.clone();
        p1.select(p2).map(move |_| tx3.send(()).unwrap()).forget();
        tx.send(c1).unwrap();
        rx2.recv().unwrap();
        drop(c2);
    }
    drop(tx);

    t.join().unwrap();
}