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
//! The MessageRecorder is used to log interactions between the client and
//! the TWS server.
//! The record is enabled by setting the environment variable IBAPI_RECORDING_DIR
//! IBAPI_RECORDING_DIR is set to the path to store logs
//! e.g. set to /tmp/logs
//! /tmp/logs/0001-request.msg
//! /tmp/logs/0002-response.msg
use std::env;
use std::fs;
use std::sync::atomic::{AtomicUsize, Ordering};
use log::warn;
use time::macros::format_description;
use time::OffsetDateTime;
use super::ResponseMessage;
static RECORDING_SEQ: AtomicUsize = AtomicUsize::new(0);
static RECORDER_ID: AtomicUsize = AtomicUsize::new(0);
#[derive(Clone, Debug)]
pub(crate) struct MessageRecorder {
enabled: bool,
recording_dir: String,
}
impl MessageRecorder {
pub fn new(enabled: bool, recording_dir: String) -> Self {
Self { enabled, recording_dir }
}
pub fn from_env() -> Self {
match env::var("IBAPI_RECORDING_DIR") {
Ok(dir) => {
if dir.is_empty() {
MessageRecorder {
enabled: false,
recording_dir: String::from(""),
}
} else {
let format = format_description!("[year]-[month]-[day]-[hour]-[minute]");
let now = OffsetDateTime::now_utc();
let instance_id = RECORDER_ID.fetch_add(1, Ordering::SeqCst);
let recording_dir = format!("{}/{}-{}", dir, now.format(&format).unwrap(), instance_id);
// A diagnostic aid must never be the reason a connection
// fails. This used to `unwrap`, so pointing
// `IBAPI_RECORDING_DIR` at an unwritable path panicked
// during connect — the same policy the raw-frame tap
// applies in `super::raw_capture`.
if let Err(err) = fs::create_dir_all(&recording_dir) {
warn!("message recording disabled: cannot create {recording_dir}: {err}");
return MessageRecorder::new(false, String::from(""));
}
MessageRecorder::new(true, recording_dir)
}
}
_ => MessageRecorder {
enabled: false,
recording_dir: String::from(""),
},
}
}
pub fn record_request(&self, data: &[u8]) {
if !self.enabled {
return;
}
let record_id = RECORDING_SEQ.fetch_add(1, Ordering::SeqCst);
if let Err(err) = fs::write(self.request_file(record_id), data) {
warn!("failed to record request: {err}");
}
}
pub fn record_response(&self, message: &ResponseMessage) {
if !self.enabled {
return;
}
let record_id = RECORDING_SEQ.fetch_add(1, Ordering::SeqCst);
if let Err(err) = fs::write(self.response_file(record_id), Self::render(message)) {
warn!("failed to record response: {err}");
}
}
/// A protobuf response is recorded as its wire frame — the 4-byte big-endian
/// message id followed by the payload — which is what `record_request`
/// already writes for outbound messages, and what a replay would need.
///
/// Text responses keep their pipe-delimited rendering.
///
/// This used to be `message.encode()` for both, which joins the parsed text
/// fields. A protobuf frame has none, so every recorded response since the
/// transition to protobuf-only was the bare message id and nothing else.
///
/// The id comes from [`ResponseMessage::message_id`], not from the resolved
/// kind. They agree for every recognized id — `IncomingMessages::from` maps
/// a value to the variant with that discriminant — but an *un*recognized id
/// resolves to `NotValid`, whose discriminant is `-1`. Recording
/// `message_type() as i32` therefore fabricated an id for exactly the frames
/// worth replaying: an operator capturing a desync burst with
/// `IBAPI_RECORDING_DIR` got `-1` where the offending id should be.
fn render(message: &ResponseMessage) -> Vec<u8> {
match (message.raw_bytes(), message.message_id()) {
(Some(payload), Some(id)) => crate::messages::encode_protobuf_message(id, payload),
_ => message.encode().replace('\0', "|").into_bytes(),
}
}
fn request_file(&self, record_id: usize) -> String {
format!("{}/{:04}-request.msg", self.recording_dir, record_id)
}
fn response_file(&self, record_id: usize) -> String {
format!("{}/{:04}-response.msg", self.recording_dir, record_id)
}
}
#[cfg(test)]
#[path = "recorder_tests.rs"]
mod tests;