ig_client/application/
streaming_convert.rs

1/******************************************************************************
2   Author: Joaquín Béjar García
3   Email: jb@taunais.com
4   Date: 20/10/25
5******************************************************************************/
6
7//! Adapters from a Lightstreamer update to the presentation-layer DTOs.
8//!
9//! The presentation DTOs stay transport-agnostic: they expose pure
10//! `from_fields(..)` constructors that take plain field maps. This module is the
11//! only place that depends on `lightstreamer_rs`; it reads an [`ItemUpdate`] and
12//! feeds the extracted metadata / field maps into those pure constructors.
13//!
14//! # The seam
15//!
16//! [`ItemUpdate`] is the streaming crate's own type: it borrows from the
17//! subscription schema, has no public constructor, and models a field value as
18//! [`FieldValue`] rather than a string. [`StreamingUpdate`] is this crate's
19//! owned, constructible mirror of it, and every conversion below goes through
20//! it. That keeps the DTO parsers testable without a live session, and keeps
21//! the null-versus-empty distinction explicit at exactly one place:
22//! `FieldValue::Null` becomes `None`, `FieldValue::Text(s)` becomes
23//! `Some(s)` — including `Some("")` for a field the server deliberately sent
24//! empty.
25//!
26//! Both a `Result`-returning function (for callers that want to observe parse
27//! failures) and the `From` conversions (which degrade to a default on failure,
28//! preserving the previous streaming behaviour) are provided per type.
29
30use crate::error::AppError;
31use crate::presentation::account::AccountData;
32use crate::presentation::chart::ChartData;
33use crate::presentation::market::{MarketFields, PresentationMarketData};
34use crate::presentation::price::PriceData;
35use crate::presentation::trade::TradeData;
36use lightstreamer_rs::{FieldValue, ItemUpdate};
37use pretty_simple_display::{DebugPretty, DisplaySimple};
38use serde::{Deserialize, Serialize};
39use std::collections::HashMap;
40
41/// One Lightstreamer item update, owned and free of borrowed schema state.
42///
43/// This is the boundary type between `lightstreamer-rs` and this crate's
44/// presentation DTOs. It is deliberately constructible by hand so that the
45/// field parsers can be exercised from captured IG payloads without a live
46/// session — the streaming crate's own [`ItemUpdate`] cannot be built outside
47/// that crate.
48///
49/// A `None` field value means the server said the field has *no* value
50/// (`FieldValue::Null`); `Some("")` means it sent an empty one. The two are
51/// different answers and are kept apart.
52#[derive(DebugPretty, DisplaySimple, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
53pub struct StreamingUpdate {
54    /// The item name the subscription declared for this position, when it
55    /// declared one. `None` for a server-resolved item group, where the
56    /// protocol never transmits names.
57    pub item_name: Option<String>,
58    /// The item's 1-based position within the subscription.
59    pub item_pos: usize,
60    /// Whether this update carries snapshot content rather than a live change.
61    pub is_snapshot: bool,
62    /// The complete state of the item after this update, keyed by field name.
63    pub fields: HashMap<String, Option<String>>,
64    /// Only the fields this update actually carried a value for.
65    pub changed_fields: HashMap<String, Option<String>>,
66}
67
68/// Converts a [`FieldValue`] into the `Option<String>` the pure field parsers
69/// consume: null becomes absent, text (empty or not) becomes present.
70#[must_use]
71#[inline]
72fn field_value_as_option(value: FieldValue<'_>) -> Option<String> {
73    match value {
74        FieldValue::Null => None,
75        FieldValue::Text(text) => Some(text.to_owned()),
76    }
77}
78
79impl From<&ItemUpdate> for StreamingUpdate {
80    fn from(update: &ItemUpdate) -> Self {
81        let fields = update
82            .fields()
83            .map(|field| {
84                (
85                    field.name().to_owned(),
86                    field_value_as_option(field.value()),
87                )
88            })
89            .collect();
90        let changed_fields = update
91            .changed_fields()
92            .map(|field| {
93                (
94                    field.name().to_owned(),
95                    field_value_as_option(field.value()),
96                )
97            })
98            .collect();
99
100        Self {
101            item_name: update.declared_item_name().map(str::to_owned),
102            item_pos: update.item_index(),
103            is_snapshot: update.is_snapshot(),
104            fields,
105            changed_fields,
106        }
107    }
108}
109
110/// Converts a streaming update into a [`PriceData`].
111///
112/// # Errors
113/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
114/// non-numeric price or an unknown dealing flag). The message names the
115/// offending field and value.
116#[must_use = "the parse result must be handled"]
117pub fn price_data_from_item_update(update: &StreamingUpdate) -> Result<PriceData, AppError> {
118    PriceData::from_fields(
119        update.item_name.as_deref(),
120        update.item_pos,
121        update.is_snapshot,
122        &update.fields,
123        &update.changed_fields,
124    )
125    .map_err(AppError::Deserialization)
126}
127
128impl From<&StreamingUpdate> for PriceData {
129    fn from(update: &StreamingUpdate) -> Self {
130        price_data_from_item_update(update).unwrap_or_else(|e| {
131            tracing::warn!(error = %e, "failed to convert streaming update to PriceData, returning default");
132            PriceData::default()
133        })
134    }
135}
136
137impl From<&ItemUpdate> for PriceData {
138    fn from(update: &ItemUpdate) -> Self {
139        Self::from(&StreamingUpdate::from(update))
140    }
141}
142
143/// Converts a streaming update into a [`PresentationMarketData`].
144///
145/// # Errors
146/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. an
147/// unknown market state or an invalid `MARKET_DELAY` value).
148#[must_use = "the parse result must be handled"]
149pub fn market_data_from_item_update(
150    update: &StreamingUpdate,
151) -> Result<PresentationMarketData, AppError> {
152    PresentationMarketData::from_fields(
153        update.item_name.as_deref(),
154        update.item_pos,
155        update.is_snapshot,
156        &update.fields,
157        &update.changed_fields,
158    )
159    .map_err(AppError::Deserialization)
160}
161
162impl From<&StreamingUpdate> for PresentationMarketData {
163    fn from(update: &StreamingUpdate) -> Self {
164        market_data_from_item_update(update).unwrap_or_else(|_| PresentationMarketData {
165            item_name: String::new(),
166            item_pos: 0,
167            fields: MarketFields::default(),
168            changed_fields: MarketFields::default(),
169            is_snapshot: false,
170        })
171    }
172}
173
174impl From<&ItemUpdate> for PresentationMarketData {
175    fn from(update: &ItemUpdate) -> Self {
176        Self::from(&StreamingUpdate::from(update))
177    }
178}
179
180/// Converts a streaming update into a [`ChartData`].
181///
182/// # Errors
183/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
184/// non-numeric candle value).
185#[must_use = "the parse result must be handled"]
186pub fn chart_data_from_item_update(update: &StreamingUpdate) -> Result<ChartData, AppError> {
187    ChartData::from_fields(
188        update.item_name.as_deref(),
189        update.item_pos,
190        update.is_snapshot,
191        &update.fields,
192        &update.changed_fields,
193    )
194    .map_err(AppError::Deserialization)
195}
196
197impl From<&StreamingUpdate> for ChartData {
198    fn from(update: &StreamingUpdate) -> Self {
199        chart_data_from_item_update(update).unwrap_or_default()
200    }
201}
202
203impl From<&ItemUpdate> for ChartData {
204    fn from(update: &ItemUpdate) -> Self {
205        Self::from(&StreamingUpdate::from(update))
206    }
207}
208
209/// Converts a streaming update into a [`TradeData`].
210///
211/// # Errors
212/// Returns [`AppError::Deserialization`] when the embedded OPU / WOU JSON
213/// payload fails to parse.
214#[must_use = "the parse result must be handled"]
215pub fn trade_data_from_item_update(update: &StreamingUpdate) -> Result<TradeData, AppError> {
216    TradeData::from_fields(
217        update.item_name.as_deref(),
218        update.item_pos,
219        update.is_snapshot,
220        &update.fields,
221        &update.changed_fields,
222    )
223    .map_err(AppError::Deserialization)
224}
225
226impl From<&StreamingUpdate> for TradeData {
227    fn from(update: &StreamingUpdate) -> Self {
228        trade_data_from_item_update(update).unwrap_or_default()
229    }
230}
231
232impl From<&ItemUpdate> for TradeData {
233    fn from(update: &ItemUpdate) -> Self {
234        Self::from(&StreamingUpdate::from(update))
235    }
236}
237
238/// Converts a streaming update into an [`AccountData`].
239///
240/// # Errors
241/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
242/// non-numeric P&L or margin value).
243#[must_use = "the parse result must be handled"]
244pub fn account_data_from_item_update(update: &StreamingUpdate) -> Result<AccountData, AppError> {
245    AccountData::from_fields(
246        update.item_name.as_deref(),
247        update.item_pos,
248        update.is_snapshot,
249        &update.fields,
250        &update.changed_fields,
251    )
252    .map_err(AppError::Deserialization)
253}
254
255impl From<&StreamingUpdate> for AccountData {
256    fn from(update: &StreamingUpdate) -> Self {
257        account_data_from_item_update(update).unwrap_or_else(|_| AccountData::default())
258    }
259}
260
261impl From<&ItemUpdate> for AccountData {
262    fn from(update: &ItemUpdate) -> Self {
263        Self::from(&StreamingUpdate::from(update))
264    }
265}
266
267#[cfg(test)]
268mod tests {
269    use super::*;
270    use crate::presentation::order::{Direction, OrderType, Status, TimeInForce};
271
272    /// Builds an update carrying a single TRADE field, mirroring how the
273    /// Lightstreamer TRADE subscription delivers OPU / WOU / CONFIRMS payloads.
274    ///
275    /// `item_name`, `item_pos` and `is_snapshot` are set to distinctive,
276    /// obviously-synthetic values so tests can assert whether they survive the
277    /// parse (they are dropped on the silent-default path — see the malformed
278    /// test).
279    fn trade_item_update(field_key: &str, field_value: &str) -> StreamingUpdate {
280        let mut fields = HashMap::new();
281        fields.insert(field_key.to_string(), Some(field_value.to_string()));
282        StreamingUpdate {
283            item_name: Some("TRADE:SYNTHETIC".to_string()),
284            item_pos: 2,
285            fields,
286            changed_fields: HashMap::new(),
287            is_snapshot: true,
288        }
289    }
290
291    // Realistic, sanitized OPU payload. Field names mirror IG's Open Position
292    // Update stream (camelCase); the deal ids are obviously synthetic and carry
293    // no secret material.
294    const SANITIZED_OPU_JSON: &str = r#"{
295        "dealReference": "REF_SYNTHETIC_0001",
296        "dealId": "DIAAAAF00SYNTH01",
297        "dealIdOrigin": "DIAAAAF00SYNTH01",
298        "direction": "BUY",
299        "epic": "IX.D.DAX.DAILY.IP",
300        "status": "OPEN",
301        "dealStatus": "ACCEPTED",
302        "level": "18000.5",
303        "size": "1.0",
304        "currency": "EUR",
305        "timestamp": "2024-01-15T10:30:00.000",
306        "channel": "PublicRestOTC",
307        "expiry": "DFB"
308    }"#;
309
310    // Realistic, sanitized WOU payload. Field names mirror IG's Working Order
311    // Update stream (camelCase); the deal ids are obviously synthetic.
312    const SANITIZED_WOU_JSON: &str = r#"{
313        "dealReference": "REF_SYNTHETIC_0002",
314        "dealId": "DIAAAAF00SYNTH02",
315        "direction": "SELL",
316        "epic": "CS.D.EURUSD.CFD.IP",
317        "status": "OPEN",
318        "dealStatus": "ACCEPTED",
319        "level": "1.1000",
320        "size": "10000",
321        "currency": "USD",
322        "timestamp": "2024-01-15T10:31:00.000",
323        "channel": "PublicRestOTC",
324        "expiry": "-",
325        "stopDistance": "10.5",
326        "limitDistance": "20.0",
327        "guaranteedStop": false,
328        "orderType": "LIMIT",
329        "timeInForce": "GOOD_TILL_CANCELLED",
330        "goodTillDate": ""
331    }"#;
332
333    #[test]
334    fn test_trade_data_from_item_update_parses_open_position_update() {
335        let item = trade_item_update("OPU", SANITIZED_OPU_JSON);
336
337        let result = trade_data_from_item_update(&item);
338        assert!(result.is_ok(), "OPU payload should parse: {result:?}");
339        let data = result.expect("OPU parse checked ok above");
340
341        // Update metadata is preserved on the success path.
342        assert_eq!(data.item_name, "TRADE:SYNTHETIC");
343        assert_eq!(data.item_pos, 2);
344        assert!(data.is_snapshot);
345
346        let opu = data.fields.opu.expect("OPU should be populated");
347        assert_eq!(opu.deal_reference.as_deref(), Some("REF_SYNTHETIC_0001"));
348        assert_eq!(opu.deal_id.as_deref(), Some("DIAAAAF00SYNTH01"));
349        assert_eq!(opu.deal_id_origin.as_deref(), Some("DIAAAAF00SYNTH01"));
350        assert_eq!(opu.direction, Some(Direction::Buy));
351        assert_eq!(opu.epic.as_deref(), Some("IX.D.DAX.DAILY.IP"));
352        assert_eq!(opu.status, Some(Status::Open));
353        assert_eq!(opu.deal_status, Some(Status::Accepted));
354        assert_eq!(opu.level, Some(18000.5));
355        assert_eq!(opu.size, Some(1.0));
356        assert_eq!(opu.currency.as_deref(), Some("EUR"));
357        assert_eq!(opu.expiry.as_deref(), Some("DFB"));
358
359        // No WOU or CONFIRMS present in this update.
360        assert!(data.fields.wou.is_none());
361        assert!(data.fields.confirms.is_none());
362    }
363
364    #[test]
365    fn test_trade_data_from_item_update_parses_working_order_update() {
366        let item = trade_item_update("WOU", SANITIZED_WOU_JSON);
367
368        let result = trade_data_from_item_update(&item);
369        assert!(result.is_ok(), "WOU payload should parse: {result:?}");
370        let data = result.expect("WOU parse checked ok above");
371
372        let wou = data.fields.wou.expect("WOU should be populated");
373        assert_eq!(wou.deal_reference.as_deref(), Some("REF_SYNTHETIC_0002"));
374        assert_eq!(wou.deal_id.as_deref(), Some("DIAAAAF00SYNTH02"));
375        assert_eq!(wou.direction, Some(Direction::Sell));
376        assert_eq!(wou.epic.as_deref(), Some("CS.D.EURUSD.CFD.IP"));
377        assert_eq!(wou.status, Some(Status::Open));
378        assert_eq!(wou.deal_status, Some(Status::Accepted));
379        assert_eq!(wou.level, Some(1.1000));
380        assert_eq!(wou.size, Some(10000.0));
381        assert_eq!(wou.currency.as_deref(), Some("USD"));
382        assert_eq!(wou.stop_distance, Some(10.5));
383        assert_eq!(wou.limit_distance, Some(20.0));
384        assert_eq!(wou.guaranteed_stop, Some(false));
385        assert_eq!(wou.order_type, Some(OrderType::Limit));
386        assert_eq!(wou.time_in_force, Some(TimeInForce::GoodTillCancelled));
387        // Empty string good-till-date degrades to None.
388        assert!(wou.good_till_date.is_none());
389
390        // No OPU or CONFIRMS present in this update.
391        assert!(data.fields.opu.is_none());
392        assert!(data.fields.confirms.is_none());
393    }
394
395    #[test]
396    fn test_trade_data_from_item_update_malformed_opu_is_error_on_direct_path() {
397        let item = trade_item_update("OPU", "{ this is not valid json ");
398
399        // The direct parse path surfaces the failure as an Err.
400        let result = trade_data_from_item_update(&item);
401        assert!(
402            result.is_err(),
403            "malformed OPU JSON must surface an error on the direct parse path"
404        );
405    }
406
407    #[test]
408    fn test_trade_data_from_impl_swallows_malformed_opu_into_default() {
409        // KNOWN GAP (pinned, do not "fix" here): `From<&ItemUpdate>` funnels any
410        // parse failure through `.unwrap_or_default()`, so a malformed TRADE
411        // payload is silently swallowed into `TradeData::default()` instead of
412        // being surfaced. This test PINS that current behavior so a future change
413        // to make the swallow deliberate (e.g. logging, or propagating the error)
414        // is a conscious, reviewed decision rather than an accidental regression.
415        //
416        // TODO(streaming): decide whether `From<&ItemUpdate>` should log the
417        // dropped payload at WARN or expose the parse error instead of defaulting.
418        let item = trade_item_update("OPU", "{ this is not valid json ");
419
420        let data = TradeData::from(&item);
421
422        // Everything degrades to the Default, including the update metadata that
423        // was present on the source ItemUpdate (item_name / item_pos / snapshot).
424        assert!(
425            data.item_name.is_empty(),
426            "silent-default path discards item_name"
427        );
428        assert_eq!(data.item_pos, 0, "silent-default path discards item_pos");
429        assert!(
430            !data.is_snapshot,
431            "silent-default path discards is_snapshot"
432        );
433        assert!(data.fields.opu.is_none());
434        assert!(data.fields.wou.is_none());
435        assert!(data.fields.confirms.is_none());
436    }
437
438    #[test]
439    fn test_price_data_from_item_update_defaults_on_unknown_dealing_flag() {
440        // An unknown dealing flag makes the pure parser fail; the `From` impl
441        // degrades to `PriceData::default()` while the direct fn surfaces the Err.
442        let mut fields = HashMap::new();
443        fields.insert("DLG_FLAG".to_string(), Some("NOT_A_FLAG".to_string()));
444        let item = StreamingUpdate {
445            item_name: Some("PRICE:CS.D.EURUSD.MINI.IP".to_string()),
446            item_pos: 1,
447            fields,
448            changed_fields: HashMap::new(),
449            is_snapshot: true,
450        };
451
452        assert!(price_data_from_item_update(&item).is_err());
453        // The `From` impl degrades to the default: metadata from the source
454        // update is discarded (matching the pre-refactor silent-default path).
455        let data = PriceData::from(&item);
456        assert!(data.item_name.is_empty());
457        assert_eq!(data.item_pos, 0);
458        assert!(!data.is_snapshot);
459        assert!(data.fields.dealing_flag.is_none());
460    }
461
462    // --- The seam type itself ----------------------------------------------
463
464    #[test]
465    fn test_field_value_null_and_empty_text_are_distinct() {
466        // The whole reason `StreamingUpdate` stores `Option<String>` rather
467        // than `String`: TLCP distinguishes "no value" from "the empty string",
468        // and the seam must not flatten one into the other.
469        assert_eq!(field_value_as_option(FieldValue::Null), None);
470        assert_eq!(
471            field_value_as_option(FieldValue::Text("")),
472            Some(String::new())
473        );
474        assert_eq!(
475            field_value_as_option(FieldValue::Text("1.2345")),
476            Some("1.2345".to_string())
477        );
478    }
479
480    #[test]
481    fn test_streaming_update_serde_round_trip() {
482        let mut fields = HashMap::new();
483        fields.insert("BID".to_string(), Some("18000.5".to_string()));
484        // A null field and an empty-text field must both survive the round trip
485        // as themselves.
486        fields.insert("OFFER".to_string(), None);
487        fields.insert("MARKET_STATE".to_string(), Some(String::new()));
488
489        let mut changed_fields = HashMap::new();
490        changed_fields.insert("BID".to_string(), Some("18000.5".to_string()));
491
492        let update = StreamingUpdate {
493            item_name: Some("MARKET:IX.D.DAX.DAILY.IP".to_string()),
494            item_pos: 3,
495            is_snapshot: true,
496            fields,
497            changed_fields,
498        };
499
500        let json = serde_json::to_string(&update).expect("StreamingUpdate should serialize");
501        let decoded: StreamingUpdate =
502            serde_json::from_str(&json).expect("StreamingUpdate should deserialize");
503
504        assert_eq!(decoded, update);
505        assert_eq!(decoded.fields.get("OFFER"), Some(&None));
506        assert_eq!(
507            decoded.fields.get("MARKET_STATE"),
508            Some(&Some(String::new()))
509        );
510    }
511}
ig_client/application/streaming_convert.rs

ig_client/application/
streaming_convert.rs
