ibapi 5.1.0

An unofficial Rust client for the Interactive Brokers TWS API. Async (Tokio) and blocking clients for TWS and IB Gateway.
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223

use super::*;
use crate::messages::encode_raw_length;
use crate::transport::raw_capture::test_support;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;

/// Async twin of `transport::sync::tests::test_read_message_taps_raw_bytes_including_rejected_prefixes`.
///
/// Two claims in one: the reader rejects a length prefix that cannot describe a
/// frame, and the tap sees that prefix anyway. The second is the one worth a
/// test — a capture holding only frames the reader accepted would omit the byte
/// sequence in #891 that an operator
/// opens the capture to find.
#[tokio::test]
async fn test_read_framed_message_taps_raw_bytes_including_rejected_prefixes() {
    let dir = tempfile::TempDir::new().unwrap();
    let tap = RawFrameTap::capturing_to(dir.path());

    let good = encode_raw_length(&[0, 0, 0, 9, 42]);
    let bad_prefix = u32::MAX.to_be_bytes();

    let mut stream = std::io::Cursor::new([good.clone(), bad_prefix.to_vec()].concat());
    assert_eq!(read_framed_message(&mut stream, &tap).await.unwrap(), [0, 0, 0, 9, 42]);
    let err = read_framed_message(&mut stream, &tap).await.expect_err("prefix must be rejected");
    assert!(matches!(err, Error::InvalidFrame(_)), "got {err:?}");

    let mut expected = good;
    expected.extend_from_slice(&bad_prefix);
    assert_eq!(test_support::frames(dir.path()), expected);
}

/// Larger than loopback send + receive buffers, so a write to a peer that is
/// not reading stalls mid-frame.
const LARGE: usize = 32 * 1024 * 1024;

/// A connected socket, its peer, and the listener, kept for reconnects.
async fn socket_pair() -> (Arc<AsyncTcpSocket>, TcpStream, tokio::net::TcpListener) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap().to_string();
    let (socket, peer) = tokio::join!(AsyncTcpSocket::connect(&address, true), listener.accept());
    (Arc::new(socket.unwrap()), peer.unwrap().0, listener)
}

/// Write `bytes` with a caller that gives up after 100ms; the write must not
/// have completed by then.
async fn abandon_write(socket: &AsyncTcpSocket, bytes: &[u8]) {
    let written = tokio::time::timeout(Duration::from_millis(100), socket.write_all(bytes)).await;
    assert!(written.is_err(), "write completed; the peer should not be reading yet");
}

/// Wait until a pending write holds the writer (stalled on the full buffer).
async fn until_writer_held(socket: &AsyncTcpSocket) {
    tokio::time::timeout(Duration::from_secs(1), async {
        while socket.writer.try_lock().is_ok() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("large write never took the writer");
}

/// The peer sees EOF: our side released its socket.
pub(in crate::transport::r#async) async fn assert_peer_sees_eof(mut peer: TcpStream) {
    let eof = tokio::time::timeout(Duration::from_secs(1), peer.read_u8())
        .await
        .expect("TCP socket retained");
    assert_eq!(eof.unwrap_err().kind(), std::io::ErrorKind::UnexpectedEof);
}

/// Everything the peer receives until the connection closes.
fn read_to_end(mut peer: TcpStream) -> tokio::task::JoinHandle<Vec<u8>> {
    tokio::spawn(async move {
        let mut received = Vec::new();
        peer.read_to_end(&mut received).await.unwrap();
        received
    })
}

/// A write cut off mid-frame breaks the connection: a pending read fails, so
/// the dispatcher reconnects, and later writes are refused rather than
/// appended to the fragment.
#[tokio::test]
async fn test_write_cut_off_mid_frame_breaks_the_connection() {
    let (socket, peer, _listener) = socket_pair().await;
    let large = vec![0xAA; LARGE];

    let (read, ()) = tokio::join!(
        tokio::time::timeout(Duration::from_secs(1), socket.read_message()),
        abandon_write(&socket, &large),
    );
    let read = read.expect("pending read not woken");
    assert!(matches!(read, Err(Error::ConnectionReset)), "read: {read:?}");
    let written = socket.write_all(&[0xCC; 3]).await;
    assert!(matches!(written, Err(Error::ConnectionReset)), "write: {written:?}");

    let reader = read_to_end(peer);
    drop(socket);
    let received = reader.await.unwrap();
    assert!(!received.is_empty() && received.len() < LARGE, "no fragment: {} bytes", received.len());
    assert!(received.iter().all(|&b| b == 0xAA), "bytes followed the fragment");
}

/// A write abandoned while it waits for the writer sends nothing and leaves
/// the connection usable.
#[tokio::test]
async fn test_write_abandoned_before_it_starts_sends_nothing() {
    let (socket, peer, _listener) = socket_pair().await;
    let large = vec![0xAA; LARGE];

    let stalled = tokio::spawn({
        let socket = socket.clone();
        let large = large.clone();
        async move { socket.write_all(&large).await }
    });
    until_writer_held(&socket).await;
    abandon_write(&socket, &[0xBB; 3]).await;

    let reader = read_to_end(peer);
    stalled.await.unwrap().unwrap();
    socket.write_all(&[0xCC; 3]).await.unwrap();
    drop(socket);

    let received = reader.await.unwrap();
    assert!(received == [large, vec![0xCC; 3]].concat(), "received {} bytes", received.len());
}

/// `reconnect` clears a break: the new connection reads and writes normally.
#[tokio::test]
async fn test_reconnect_clears_a_broken_connection() {
    let (socket, _peer, listener) = socket_pair().await;
    abandon_write(&socket, &vec![0xAA; LARGE]).await;
    assert!(matches!(socket.write_all(&[0xCC; 3]).await, Err(Error::ConnectionReset)));

    let (reconnected, accepted) = tokio::join!(socket.reconnect(), listener.accept());
    reconnected.unwrap();
    let mut peer = accepted.unwrap().0;

    socket.write_all(&[0xCC; 3]).await.unwrap();
    let mut received = [0u8; 3];
    peer.read_exact(&mut received).await.unwrap();
    assert_eq!(received, [0xCC; 3]);

    peer.write_all(&encode_raw_length(&[0, 0, 0, 9, 42])).await.unwrap();
    assert_eq!(socket.read_message().await.unwrap(), [0, 0, 0, 9, 42]);
}

/// Terminal close interrupts a blocked read, releases TCP, and cannot reconnect.
#[tokio::test]
async fn test_physical_close_wakes_read_and_is_terminal() {
    let (socket, peer, _listener) = socket_pair().await;
    let reading = socket.clone();
    let read = tokio::spawn(async move { reading.read_message().await });
    tokio::task::yield_now().await;
    tokio::time::timeout(Duration::from_secs(1), socket.close())
        .await
        .expect("close blocked on read");
    let read = read.await.unwrap();
    assert!(matches!(read, Err(Error::Shutdown)), "read: {read:?}");
    assert_peer_sees_eof(peer).await;
    socket.close().await;
    assert!(matches!(socket.write_all(&[1]).await, Err(Error::Shutdown)));
    assert!(matches!(socket.read_message().await, Err(Error::Shutdown)));
    assert!(matches!(socket.reconnect().await, Err(Error::Shutdown)));
}

/// Close overrides a break: I/O fails with `Shutdown`, not `ConnectionReset`,
/// and reconnect refuses.
#[tokio::test]
async fn test_physical_close_overrides_a_broken_connection() {
    let (socket, _peer, _listener) = socket_pair().await;
    abandon_write(&socket, &vec![0xAA; LARGE]).await;
    assert!(matches!(socket.write_all(&[0xCC; 3]).await, Err(Error::ConnectionReset)));

    socket.close().await;
    let read = socket.read_message().await;
    assert!(matches!(read, Err(Error::Shutdown)), "read: {read:?}");
    let written = socket.write_all(&[0xCC; 3]).await;
    assert!(matches!(written, Err(Error::Shutdown)), "write: {written:?}");
    assert!(matches!(socket.reconnect().await, Err(Error::Shutdown)));
}

/// A sender queued before close cannot write when its lock becomes available.
#[tokio::test]
async fn test_physical_close_prevents_a_queued_write() {
    let (socket, peer, _listener) = socket_pair().await;
    let writer = socket.writer.lock().await;
    let sending = socket.clone();
    let write = tokio::spawn(async move { sending.write_all(&[0xAA; 4]).await });
    tokio::task::yield_now().await;
    let closing = socket.clone();
    let close = tokio::spawn(async move { closing.close().await });
    tokio::time::timeout(Duration::from_secs(1), async {
        while !socket.is_closed() {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("close never latched");
    drop(writer);
    tokio::time::timeout(Duration::from_secs(1), close).await.unwrap().unwrap();
    assert!(matches!(write.await.unwrap(), Err(Error::Shutdown)));
    assert_peer_sees_eof(peer).await;
}

/// Close wakes a write stalled on a full send buffer, rather than waiting for
/// the peer to drain it.
#[tokio::test]
async fn test_physical_close_wakes_a_backpressured_write() {
    let (socket, _peer, _listener) = socket_pair().await;
    let write = tokio::spawn({
        let socket = socket.clone();
        async move { socket.write_all(&vec![0xAA; LARGE]).await }
    });
    until_writer_held(&socket).await;

    tokio::time::timeout(Duration::from_secs(1), socket.close())
        .await
        .expect("close blocked on write");
    let written = write.await.unwrap();
    assert!(matches!(written, Err(Error::Shutdown)), "write: {written:?}");
    // The woken write was cut off mid-frame; that must not undo the close.
    assert!(matches!(socket.reconnect().await, Err(Error::Shutdown)));
}