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
//! 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 {
/// A recorder that writes nothing.
pub fn disabled() -> Self {
Self {
enabled: false,
recording_dir: String::new(),
}
}
/// Read `IBAPI_RECORDING_DIR`. Unset or empty yields [`Self::disabled`].
pub fn from_env() -> Self {
match env::var("IBAPI_RECORDING_DIR") {
Ok(dir) if !dir.is_empty() => Self::recording_to(&dir),
_ => Self::disabled(),
}
}
/// Record into a fresh `<dir>/<timestamp>-<instance>` directory.
///
/// A diagnostic aid must never be the reason a connection fails: a
/// directory that cannot be created downgrades to [`Self::disabled`] with a
/// warning, the same policy the raw-frame tap applies in `super::raw_capture`.
pub fn recording_to(dir: &str) -> Self {
let format = format_description!("[year]-[month]-[day]-[hour]-[minute]");
let stamp = OffsetDateTime::now_utc().format(format).unwrap_or_else(|_| String::from("unknown"));
let instance_id = RECORDER_ID.fetch_add(1, Ordering::SeqCst);
let recording_dir = format!("{dir}/{stamp}-{instance_id}");
if let Err(err) = fs::create_dir_all(&recording_dir) {
warn!("message recording disabled: cannot create {recording_dir}: {err}");
return Self::disabled();
}
Self {
enabled: true,
recording_dir,
}
}
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;