ibapi 3.0.1

A Rust implementation of the Interactive Brokers TWS API, providing a reliable and user friendly interface for TWS and IB Gateway. Designed with a focus on simplicity and performance.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

//! Unit tests for the sync and async `NoticeStream` impls.

use crate::messages::Notice;

fn make_notice(code: i32, message: &str) -> Notice {
    Notice {
        code,
        message: message.into(),
        error_time: None,
        advanced_order_reject_json: String::new(),
    }
}

#[cfg(feature = "sync")]
mod sync_tests {
    use super::*;
    use crate::subscriptions::notice_stream::sync_impl::NoticeStream;
    use crossbeam::channel;
    use std::time::Duration;

    #[test]
    fn next_returns_pushed_notice() {
        let (sender, receiver) = channel::unbounded();
        let stream = NoticeStream::new(receiver);

        sender.send(make_notice(2104, "farm OK")).unwrap();
        let notice = stream.next().expect("pending notice not received");
        assert_eq!(notice.code, 2104);
        assert_eq!(notice.message, "farm OK");
    }

    #[test]
    fn try_next_is_non_blocking() {
        let (sender, receiver) = channel::unbounded();
        let stream = NoticeStream::new(receiver);

        assert!(stream.try_next().is_none(), "empty channel should yield None");

        sender.send(make_notice(1100, "lost")).unwrap();
        assert_eq!(stream.try_next().expect("notice").code, 1100);
    }

    #[test]
    fn next_timeout_returns_none_when_idle() {
        let (_sender, receiver) = channel::unbounded::<Notice>();
        let stream = NoticeStream::new(receiver);

        let start = std::time::Instant::now();
        assert!(stream.next_timeout(Duration::from_millis(50)).is_none());
        assert!(start.elapsed() >= Duration::from_millis(45));
    }

    #[test]
    fn next_returns_none_when_sender_dropped() {
        let (sender, receiver) = channel::unbounded::<Notice>();
        let stream = NoticeStream::new(receiver);
        drop(sender);
        assert!(stream.next().is_none());
    }

    #[test]
    fn iter_yields_buffered_notices_in_order() {
        let (sender, receiver) = channel::unbounded();
        let stream = NoticeStream::new(receiver);

        sender.send(make_notice(2104, "a")).unwrap();
        sender.send(make_notice(2107, "b")).unwrap();
        drop(sender);

        let codes: Vec<i32> = stream.iter().map(|n| n.code).collect();
        assert_eq!(codes, vec![2104, 2107]);
    }
}

#[cfg(feature = "async")]
mod async_tests {
    use super::*;
    use crate::subscriptions::notice_stream::async_impl::NoticeStream;
    use futures::StreamExt;
    use tokio::sync::broadcast;

    #[tokio::test]
    async fn next_returns_pushed_notice() {
        let (sender, receiver) = broadcast::channel(8);
        let mut stream = NoticeStream::new(receiver);

        sender.send(make_notice(2104, "farm OK")).unwrap();
        let notice = stream.next().await.expect("pending notice not received");
        assert_eq!(notice.code, 2104);
    }

    #[tokio::test]
    async fn next_returns_none_when_sender_dropped() {
        let (sender, receiver) = broadcast::channel::<Notice>(8);
        let mut stream = NoticeStream::new(receiver);
        drop(sender);
        assert!(stream.next().await.is_none());
    }

    #[tokio::test]
    async fn fan_out_to_two_subscribers() {
        let (sender, _) = broadcast::channel(8);
        let mut a = NoticeStream::new(sender.subscribe());
        let mut b = NoticeStream::new(sender.subscribe());

        sender.send(make_notice(1100, "lost")).unwrap();

        assert_eq!(a.next().await.unwrap().code, 1100);
        assert_eq!(b.next().await.unwrap().code, 1100);
    }

    #[tokio::test]
    async fn lag_is_skipped() {
        let (sender, receiver) = broadcast::channel(2);
        let mut stream = NoticeStream::new(receiver);

        // Overflow the channel; receiver lags.
        for code in 1..=4 {
            sender.send(make_notice(code, "")).unwrap();
        }
        // First recv lags; loop in `next` skips the lag and returns the most recent.
        let n = stream.next().await.expect("notice");
        assert!(n.code >= 3, "expected most recent post-lag notice, got {}", n.code);
    }

    #[tokio::test]
    async fn stream_adapter_collects() {
        let (sender, receiver) = broadcast::channel(8);
        let mut s = NoticeStream::new(receiver);

        for code in [2104, 2107, 1102] {
            sender.send(make_notice(code, "")).unwrap();
        }
        drop(sender);

        let collected: Vec<i32> = s.stream().map(|n| n.code).collect().await;
        assert_eq!(collected, vec![2104, 2107, 1102]);
    }
}