netget/client/fix/
mod.rs

1//! FIX initiator over the acceptor's codec and session layer.
2pub mod actions;
3use crate::client::llm_budget::call_llm_for_client;
4use crate::llm::actions::client_trait::{Client, ClientActionResult};
5use crate::logging::emit::Log;
6use crate::protocol::{ConnectContext, Event};
7use crate::server::fix::codec::{self, Frame, Message};
8use crate::server::fix::session::{self, Session};
9use crate::server::fix::{actions as server_actions, comp_id_ok, dict, BEGIN_STRINGS, LOGOUT_WAIT};
10use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
11use crate::state::{AccessLogOwner, ClientStatus};
12use crate::utils::clock::Instant;
13pub use actions::FixClientProtocol;
14use anyhow::{bail, ensure, Context, Result};
15use serde_json::{json, Value};
16use std::net::SocketAddr;
17use std::time::Duration;
18use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
19use tokio::net::TcpStream;
20use tokio::sync::mpsc;
21
22pub const DEFAULT_TARGET: &str = "EXCHANGE";
23pub const DEFAULT_HEARTBEAT: Duration = Duration::from_secs(30);
24const LOGON_WAIT: Duration = Duration::from_secs(10);
25
26struct Wire {
27    r: ReadHalf<TcpStream>,
28    w: WriteHalf<TcpStream>,
29    buf: Vec<u8>,
30}
31
32impl Wire {
33    async fn write(&mut self, chunks: &[Vec<u8>]) -> Result<()> {
34        for c in chunks.iter().filter(|c| !c.is_empty()) {
35            self.w.write_all(c).await?;
36        }
37        Ok(())
38    }
39
40    async fn next(&mut self) -> Result<Option<Message>> {
41        loop {
42            match codec::frame(&self.buf) {
43                Frame::Message {
44                    message, consumed, ..
45                } => {
46                    self.buf.drain(..consumed);
47                    return Ok(Some(message));
48                }
49                Frame::Garbled { consumed, .. } => {
50                    self.buf.drain(..consumed);
51                }
52                Frame::Incomplete => {
53                    ensure!(
54                        self.buf.len() <= codec::MAX_BODY + 64,
55                        "unframed input over {} bytes",
56                        codec::MAX_BODY
57                    );
58                    let mut chunk = [0u8; 8192];
59                    let n = self.r.read(&mut chunk).await?;
60                    if n == 0 {
61                        return Ok(None);
62                    }
63                    self.buf.extend_from_slice(&chunk[..n]);
64                }
65            }
66        }
67    }
68}
69
70pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
71    let p = ctx.startup_params.as_ref();
72    let s = |k: &str| -> Result<Option<String>> {
73        Ok(p.map(|p| p.get_optional_string(k)).transpose()?.flatten())
74    };
75    let sender =
76        s("sender_comp_id")?.unwrap_or_else(|| crate::server::fix::DEFAULT_COMP_ID.to_owned());
77    let target = s("target_comp_id")?.unwrap_or_else(|| DEFAULT_TARGET.to_owned());
78    ensure!(
79        comp_id_ok(&sender) && comp_id_ok(&target),
80        "CompIDs are 1 to 64 printable characters without spaces"
81    );
82    let begin =
83        s("begin_string")?.unwrap_or_else(|| crate::server::fix::DEFAULT_BEGIN_STRING.to_owned());
84    ensure!(
85        BEGIN_STRINGS.contains(&begin.as_str()),
86        "begin_string is one of {BEGIN_STRINGS:?}"
87    );
88    let hb = p
89        .map(|p| p.get_optional_u64("heartbeat_secs"))
90        .transpose()?
91        .flatten()
92        .unwrap_or(DEFAULT_HEARTBEAT.as_secs());
93    ensure!(hb <= 3600, "heartbeat_secs is 0 to 3600");
94    let reset = p
95        .map(|p| p.get_optional_bool("reset_seq_num"))
96        .transpose()?
97        .flatten()
98        .unwrap_or(true);
99    let (username, password) = (s("username")?, s("password")?);
100    for v in [&username, &password].into_iter().flatten() {
101        ensure!(
102            !v.is_empty() && v.len() <= 256 && !crate::utils::sanitize::has_controls(&v),
103            "username and password are 1 to 256 printable characters"
104        );
105    }
106    let stream = tokio::time::timeout(LOGON_WAIT, TcpStream::connect(&ctx.remote_addr))
107        .await
108        .context("FIX connect timed out")??;
109    let local = stream.local_addr()?;
110    let (r, w) = tokio::io::split(stream);
111    let mut wire = Wire {
112        r,
113        w,
114        buf: Vec::new(),
115    };
116    let mut session = Session::new(
117        &begin,
118        &sender,
119        &target,
120        if hb == 0 {
121            Duration::from_secs(86_400)
122        } else {
123            Duration::from_secs(hb)
124        },
125    );
126    let mut logon = vec![(98, "0".to_owned()), (108, hb.to_string())];
127    if reset {
128        logon.push((141, "Y".into()));
129    }
130    if let Some(u) = &username {
131        logon.push((553, u.clone()));
132    }
133    if let Some(pw) = &password {
134        logon.push((554, pw.clone()));
135    }
136    let bytes = session.encode("A", &logon)?;
137    wire.write(&[bytes]).await?;
138    let answer = tokio::time::timeout(LOGON_WAIT, wire.next())
139        .await
140        .context("no answer to the Logon")??
141        .context("the acceptor closed the connection during logon")?;
142    match answer.msg_type() {
143        "A" => session.accept_logon(answer.seq().unwrap_or(1)),
144        "5" => bail!(
145            "logon refused: {}",
146            answer.get(58).unwrap_or("no reason given")
147        ),
148        other => bail!("the acceptor answered the Logon with MsgType {other}"),
149    }
150    ctx.state
151        .update_client_status(ctx.client_id, ClientStatus::Connected)
152        .await;
153    let external =
154        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
155    let (internal_tx, internal_rx) = mpsc::channel::<Value>(16);
156    let (event_tx, mut event_rx) = mpsc::channel::<Event>(64);
157    event_tx.try_send(Event::new(&actions::LOGGED_ON_EVENT, json!({"sender_comp_id": sender, "target_comp_id": target, "heartbeat_secs": answer.get(108).and_then(|v| v.parse::<u64>().ok()).unwrap_or(hb)})))?;
158    let events_ctx = ctx.clone();
159    let dispatcher = tokio::spawn(async move {
160        let protocol = FixClientProtocol;
161        while let Some(event) = event_rx.recv().await {
162            let instruction = events_ctx
163                .state
164                .get_instruction_for_client(events_ctx.client_id)
165                .await
166                .unwrap_or_default();
167            let memory = events_ctx
168                .state
169                .get_memory_for_client(events_ctx.client_id)
170                .await
171                .unwrap_or_default();
172            match call_llm_for_client(
173                &events_ctx.llm_client,
174                &events_ctx.state,
175                events_ctx.client_id.to_string(),
176                &instruction,
177                &memory,
178                Some(&event),
179                &protocol,
180                &events_ctx.status_tx,
181            )
182            .await
183            {
184                Ok(result) => {
185                    if let Some(memory) = result.memory_updates {
186                        events_ctx
187                            .state
188                            .set_memory_for_client(events_ctx.client_id, memory)
189                            .await;
190                    }
191                    for action in result.actions {
192                        if internal_tx.send(action).await.is_err() {
193                            return;
194                        }
195                    }
196                }
197                Err(e) => {
198                    Log::new(Some(&events_ctx.status_tx)).warn(format!("FIX client handler: {e}"))
199                }
200            }
201        }
202    });
203    ctx.state
204        .register_client_task(ctx.client_id, dispatcher)
205        .await;
206    let session_ctx = ctx.clone();
207    let task = tokio::spawn(async move {
208        let reason = match run(
209            &session_ctx,
210            &mut wire,
211            &mut session,
212            external,
213            internal_rx,
214            &event_tx,
215        )
216        .await
217        {
218            Ok(r) => r,
219            Err(e) => format!("{e:#}"),
220        };
221        let _ = event_tx
222            .send(Event::new(
223                &actions::LOGGED_OUT_EVENT,
224                json!({"reason": reason}),
225            ))
226            .await;
227        session_ctx
228            .state
229            .update_client_status(session_ctx.client_id, ClientStatus::Disconnected)
230            .await;
231        session_ctx
232            .state
233            .remove_client_handle(session_ctx.client_id)
234            .await;
235        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
236    });
237    ctx.state.register_client_task(ctx.client_id, task).await;
238    Ok(local)
239}
240
241/// Run the session until it ends; the reason.
242async fn run(
243    ctx: &ConnectContext,
244    wire: &mut Wire,
245    session: &mut Session,
246    mut external: mpsc::Receiver<ClientCommand>,
247    mut internal: mpsc::Receiver<Value>,
248    events: &mpsc::Sender<Event>,
249) -> Result<String> {
250    let mut tick = tokio::time::interval(Duration::from_secs(1));
251    let mut closing: Option<Instant> = None;
252    enum Wake {
253        Message(Option<Message>),
254        Tick,
255        Action(Value, Option<ClientCommand>),
256        Idle,
257    }
258    loop {
259        if closing.is_some_and(|t| t.elapsed() >= LOGOUT_WAIT) {
260            return Ok("no Logout answer from the acceptor".into());
261        }
262        let wake = tokio::select! {
263            m = wire.next() => Wake::Message(m?),
264            _ = tick.tick() => Wake::Tick,
265            c = external.recv(), if closing.is_none() => match c { Some(c) => Wake::Action(c.action.clone(), Some(c)), None => Wake::Idle },
266            a = internal.recv(), if closing.is_none() => match a { Some(a) => Wake::Action(a, None), None => Wake::Idle },
267        };
268        match wake {
269            Wake::Idle => {}
270            Wake::Message(None) => return Ok("the acceptor closed the connection".into()),
271            Wake::Tick => {
272                let (send, close) = session.tick();
273                wire.write(&send).await?;
274                if let Some(reason) = close {
275                    return Ok(reason);
276                }
277            }
278            Wake::Action(action, command) => {
279                let outcome = match FixClientProtocol.execute_action(action.clone()) {
280                    Err(e) => ClientSendOutcome::Rejected {
281                        error: e.to_string(),
282                    },
283                    Ok(ClientActionResult::Disconnect) => {
284                        let bytes = session.logout(None);
285                        let _ = wire.write(&[bytes]).await;
286                        if let Some(c) = command {
287                            crate::client::command_support::reply(
288                                c,
289                                Ok(ClientSendOutcome::Disconnected),
290                            );
291                        }
292                        return Ok("disconnected".into());
293                    }
294                    Ok(_) => {
295                        let bytes = match action["type"].as_str() {
296                            Some("fix_logout") => Ok(session.logout(action["text"].as_str())),
297                            _ => server_actions::app_msg_type(&action).and_then(|t| {
298                                session.encode(&t, &server_actions::body_fields(&action)?)
299                            }),
300                        };
301                        match bytes {
302                            Ok(b) => {
303                                let n = b.len();
304                                wire.write(&[b]).await?;
305                                ClientSendOutcome::Sent { bytes_sent: n }
306                            }
307                            Err(e) => ClientSendOutcome::Rejected {
308                                error: e.to_string(),
309                            },
310                        }
311                    }
312                };
313                if session.logout_sent {
314                    closing.get_or_insert_with(Instant::now);
315                }
316                if let Some(c) = command {
317                    ctx.state
318                        .record_access_log(
319                            AccessLogOwner::Client(ctx.client_id.as_u32()),
320                            "FIX",
321                            None,
322                            "injected_action",
323                            json!({"type": action["type"], "msg_type": action["msg_type"]}),
324                            vec![serde_json::to_value(&outcome).unwrap_or(Value::Null)],
325                        )
326                        .await;
327                    crate::client::command_support::reply(c, Ok(outcome));
328                }
329            }
330            Wake::Message(Some(m)) => {
331                let step = session.on_message(m);
332                wire.write(&step.send).await?;
333                if let Some(reason) = step.close {
334                    return Ok(reason);
335                }
336                if let Some(app) = step.app {
337                    let msg_type = app.msg_type().to_owned();
338                    let fields: Vec<Value> = session::body(&app).into_iter().map(|(tag, value)| json!({"tag": tag, "name": dict::field_name(tag), "value": value})).collect();
339                    let event = Event::new(
340                        &actions::MESSAGE_EVENT,
341                        json!({"msg_type": msg_type, "msg_type_name": dict::message_name(&msg_type), "seq": app.seq(), "fields": fields}),
342                    );
343                    if events.send(event).await.is_err() {
344                        return Ok("event consumer stopped".into());
345                    }
346                }
347            }
348        }
349    }
350}
netget/client/fix/mod.rs

netget/client/fix/
mod.rs
