hotfix 0.12.0

Buy-side FIX engine written in pure Rust
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

use crate::config::SessionConfig;
use crate::message::{Message, OutboundMessage};
use crate::session::admin_request::AdminRequest;
use crate::session::ctx::SessionCtx;
use crate::session::event::SessionEvent;
use crate::session::session_ref::{InternalSessionRef, OutboundRequest};
use crate::store::{MessageStore, Result as StoreResult};
use crate::transport::writer::{WriterMessage, WriterRef};
use chrono::{DateTime, Utc};
use hotfix_message::MessageBuilder;
use hotfix_message::dict::Dictionary;
use hotfix_message::message::Config as MessageConfig;
use tokio::sync::mpsc;

#[derive(Clone)]
pub(crate) struct FakeMessageStore {
    pub(crate) messages: Vec<Vec<u8>>,
    pub(crate) next_sender_seq: u64,
    pub(crate) next_target_seq: u64,
}

impl FakeMessageStore {
    pub(crate) fn new() -> Self {
        Self {
            messages: vec![],
            next_sender_seq: 1,
            next_target_seq: 1,
        }
    }
}

#[async_trait::async_trait]
impl MessageStore for FakeMessageStore {
    async fn add(&mut self, _: u64, msg: &[u8]) -> StoreResult<()> {
        self.messages.push(msg.to_vec());
        Ok(())
    }
    async fn get_slice(&self, _: usize, _: usize) -> StoreResult<Vec<Vec<u8>>> {
        Ok(self.messages.clone())
    }
    fn next_sender_seq_number(&self) -> u64 {
        self.next_sender_seq
    }
    fn next_target_seq_number(&self) -> u64 {
        self.next_target_seq
    }
    async fn increment_sender_seq_number(&mut self) -> StoreResult<()> {
        self.next_sender_seq += 1;
        Ok(())
    }
    async fn increment_target_seq_number(&mut self) -> StoreResult<()> {
        self.next_target_seq += 1;
        Ok(())
    }
    async fn set_target_seq_number(&mut self, seq: u64) -> StoreResult<()> {
        self.next_target_seq = seq;
        Ok(())
    }
    async fn reset(&mut self) -> StoreResult<()> {
        self.messages.clear();
        self.next_sender_seq = 1;
        self.next_target_seq = 1;
        Ok(())
    }
    fn creation_time(&self) -> DateTime<Utc> {
        Utc::now()
    }
}

pub(crate) fn create_writer() -> (WriterRef, mpsc::Receiver<WriterMessage>) {
    let (sender, receiver) = mpsc::channel(16);
    (WriterRef::new(sender), receiver)
}

pub(crate) fn create_test_ctx(store: FakeMessageStore) -> SessionCtx<(), FakeMessageStore> {
    let message_config = MessageConfig::default();
    let dictionary = Dictionary::fix44();
    let message_builder = MessageBuilder::new(dictionary, message_config).unwrap();
    SessionCtx {
        config: SessionConfig {
            begin_string: "FIX.4.4".to_string(),
            sender_comp_id: "SENDER".to_string(),
            target_comp_id: "TARGET".to_string(),
            data_dictionary_path: None,
            connection_host: "localhost".to_string(),
            connection_port: 9876,
            tls_config: None,
            heartbeat_interval: 30,
            logon_timeout: 10,
            logout_timeout: 2,
            reconnect_interval: 30,
            reset_on_logon: false,
            schedule: None,
        },
        store,
        application: (),
        message_builder,
        message_config,
    }
}

/// Extract the FIX message type (tag 35) from a raw FIX message bytes.
pub(crate) fn extract_msg_type(raw: &[u8]) -> Option<String> {
    let s = std::str::from_utf8(raw).ok()?;
    for field in s.split('\x01') {
        if let Some(value) = field.strip_prefix("35=") {
            return Some(value.to_string());
        }
    }
    None
}

/// Extract a string field value by tag number from raw FIX message bytes.
pub(crate) fn extract_field(raw: &[u8], tag: u32) -> Option<String> {
    let s = std::str::from_utf8(raw).ok()?;
    let prefix = format!("{tag}=");
    for field in s.split('\x01') {
        if let Some(value) = field.strip_prefix(&prefix) {
            return Some(value.to_string());
        }
    }
    None
}

#[derive(Clone)]
pub(crate) struct TestMessage;

impl OutboundMessage for TestMessage {
    fn write(&self, _msg: &mut Message) {}
    fn message_type(&self) -> &str {
        "TEST"
    }
}

pub(crate) fn create_test_session_ref() -> (
    InternalSessionRef<TestMessage>,
    mpsc::Receiver<SessionEvent>,
) {
    let (event_sender, event_receiver) = mpsc::channel::<SessionEvent>(100);
    let (outbound_message_sender, _outbound_receiver) =
        mpsc::channel::<OutboundRequest<TestMessage>>(10);
    let (admin_request_sender, _admin_receiver) = mpsc::channel::<AdminRequest>(10);

    let session_ref = InternalSessionRef {
        event_sender,
        outbound_message_sender,
        admin_request_sender,
    };

    (session_ref, event_receiver)
}