Skip to main content

nautilus_common/messages/data/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Data specific messages such as subscriptions and requests.
17
18use std::{any::Any, sync::Arc};
19
20use nautilus_core::{Params, UUID4, UnixNanos};
21use nautilus_model::{
22    data::{BarType, NautilusDataType},
23    identifiers::{ClientId, InstrumentId, Venue},
24};
25use serde::{Deserialize, Serialize};
26
27use crate::cache::Cache;
28
29pub mod request;
30pub mod response;
31pub mod subscribe;
32pub mod unsubscribe;
33
34mod identifiers;
35
36/// Params key used to flag a book subscription as targeting a parent symbol.
37///
38/// When the boolean value is `true`, the subscription fans out across all
39/// instruments that resolve from the parent components (see
40/// [`InstrumentId::parse_parent_components`]). When absent or `false`, the
41/// subscription is routed to the concrete instrument id only.
42///
43/// [`InstrumentId::parse_parent_components`]: nautilus_model::identifiers::InstrumentId::parse_parent_components
44pub const PARAMS_IS_PARENT: &str = "is_parent";
45
46// Re-exports
47pub use identifiers::{
48    IdentifierCountError, IntoBarTypes, IntoInstrumentIds, QuoteRouting, ensure_routable_bar_types,
49    ensure_routable_instrument_ids, ensure_subscription_routes, quote_routing,
50};
51pub use request::{
52    RequestBars, RequestBookDeltas, RequestBookDepth, RequestBookSnapshot, RequestCustomData,
53    RequestFundingRates, RequestInstrument, RequestInstruments, RequestJoin,
54    RequestOptionChainReferencePrice, RequestQuotes, RequestTrades,
55};
56pub use response::{
57    BarsResponse, BookDeltasResponse, BookDepthResponse, BookResponse, CustomDataResponse,
58    FundingRatesResponse, InstrumentResponse, InstrumentsResponse,
59    OptionChainReferencePriceResponse, QuotesResponse, TradesResponse,
60};
61pub use subscribe::{
62    SubscribeBars, SubscribeBookDeltas, SubscribeBookDepth, SubscribeBookSnapshots,
63    SubscribeCustomData, SubscribeFundingRates, SubscribeIndexPrices, SubscribeInstrument,
64    SubscribeInstrumentClose, SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
65    SubscribeOptionChain, SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades,
66    SubscriptionError,
67};
68pub use unsubscribe::{
69    UnsubscribeBars, UnsubscribeBookDeltas, UnsubscribeBookDepth, UnsubscribeBookSnapshots,
70    UnsubscribeCustomData, UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
71    UnsubscribeInstrumentClose, UnsubscribeInstrumentStatus, UnsubscribeInstruments,
72    UnsubscribeMarkPrices, UnsubscribeOptionChain, UnsubscribeOptionGreeks, UnsubscribeQuotes,
73    UnsubscribeTrades,
74};
75
76#[cfg(feature = "defi")]
77use crate::messages::defi::{DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand};
78
79#[non_exhaustive]
80#[derive(Clone, Debug, PartialEq)]
81pub enum DataCommand {
82    Request(RequestCommand),
83    Subscribe(SubscribeCommand),
84    Unsubscribe(UnsubscribeCommand),
85    #[cfg(feature = "defi")]
86    DefiRequest(DefiRequestCommand),
87    #[cfg(feature = "defi")]
88    DefiSubscribe(DefiSubscribeCommand),
89    #[cfg(feature = "defi")]
90    DefiUnsubscribe(DefiUnsubscribeCommand),
91}
92
93impl DataCommand {
94    /// Converts the command to a dyn Any trait object for messaging.
95    pub fn as_any(&self) -> &dyn Any {
96        self
97    }
98
99    /// Converts a subscribe variant into its matching unsubscribe variant.
100    ///
101    /// Returns `None` for request and unsubscribe variants.
102    pub(crate) fn into_unsubscribe(self, command_id: UUID4, ts_init: UnixNanos) -> Option<Self> {
103        match self {
104            Self::Subscribe(command) => {
105                let correlation_id = matches!(
106                    command,
107                    SubscribeCommand::BookDeltas(_)
108                        | SubscribeCommand::BookDepth(_)
109                        | SubscribeCommand::BookSnapshots(_)
110                )
111                .then(|| command.command_id());
112                Some(Self::Unsubscribe(command.into_unsubscribe(
113                    command_id,
114                    ts_init,
115                    correlation_id,
116                )))
117            }
118            #[cfg(feature = "defi")]
119            Self::DefiSubscribe(command) => Some(Self::DefiUnsubscribe(
120                command.into_unsubscribe(command_id, ts_init),
121            )),
122            _ => None,
123        }
124    }
125}
126
127#[derive(Clone, Debug, Serialize, Deserialize)]
128pub enum SubscribeCommand {
129    Data(SubscribeCustomData),
130    Instrument(SubscribeInstrument),
131    Instruments(SubscribeInstruments),
132    BookDeltas(SubscribeBookDeltas),
133    BookDepth(SubscribeBookDepth),
134    BookSnapshots(SubscribeBookSnapshots),
135    Quotes(SubscribeQuotes),
136    Trades(SubscribeTrades),
137    Bars(SubscribeBars),
138    MarkPrices(SubscribeMarkPrices),
139    IndexPrices(SubscribeIndexPrices),
140    FundingRates(SubscribeFundingRates),
141    InstrumentStatus(SubscribeInstrumentStatus),
142    InstrumentClose(SubscribeInstrumentClose),
143    OptionGreeks(SubscribeOptionGreeks),
144    OptionChain(SubscribeOptionChain),
145}
146
147impl PartialEq for SubscribeCommand {
148    fn eq(&self, other: &Self) -> bool {
149        self.command_id() == other.command_id()
150    }
151}
152
153impl SubscribeCommand {
154    /// Validates list routing before subscription ownership is retained.
155    ///
156    /// # Errors
157    ///
158    /// Returns an error if an identifier or bar list is invalid or trade routes are mixed.
159    /// For non-external clients, also rejects mixed quote routes and unsupported lists with
160    /// multiple IDs containing synthetic members.
161    pub fn validate_routing(&self, cache: &Cache) -> anyhow::Result<()> {
162        if let Some(bar_types) = self.bar_types() {
163            ensure_routable_bar_types(bar_types, self.client_id().is_some())?;
164        }
165
166        let Some(ids) = self.instrument_ids() else {
167            return Ok(());
168        };
169
170        ensure_routable_instrument_ids(ids, self.client_id().is_some())?;
171
172        match self {
173            Self::Quotes(command) => {
174                quote_routing(ids, self.client_id(), command.params.as_ref(), cache)?;
175            }
176            Self::Trades(_) => {
177                anyhow::ensure!(
178                    ids.iter()
179                        .all(|id| id.is_synthetic() == ids[0].is_synthetic()),
180                    "trade identifier lists cannot mix client and synthetic routes"
181                );
182            }
183            Self::Instrument(_)
184            | Self::BookDeltas(_)
185            | Self::BookDepth(_)
186            | Self::BookSnapshots(_)
187            | Self::InstrumentStatus(_)
188            | Self::InstrumentClose(_)
189            | Self::OptionGreeks(_) => {
190                ensure_subscription_routes(ids, self.client_id(), cache)?;
191            }
192            _ => {}
193        }
194
195        Ok(())
196    }
197
198    /// Returns the instrument IDs for an instrument subscription, if applicable.
199    #[must_use]
200    pub fn instrument_ids(&self) -> Option<&[InstrumentId]> {
201        match self {
202            Self::Instrument(command) => Some(&command.instrument_ids),
203            Self::BookDeltas(command) => Some(&command.instrument_ids),
204            Self::BookDepth(command) => Some(&command.instrument_ids),
205            Self::BookSnapshots(command) => Some(&command.instrument_ids),
206            Self::Quotes(command) => Some(&command.instrument_ids),
207            Self::Trades(command) => Some(&command.instrument_ids),
208            Self::MarkPrices(command) => Some(&command.instrument_ids),
209            Self::IndexPrices(command) => Some(&command.instrument_ids),
210            Self::FundingRates(command) => Some(&command.instrument_ids),
211            Self::InstrumentStatus(command) => Some(&command.instrument_ids),
212            Self::InstrumentClose(command) => Some(&command.instrument_ids),
213            Self::OptionGreeks(command) => Some(&command.instrument_ids),
214            _ => None,
215        }
216    }
217
218    /// Projects an instrument subscription onto the supplied IDs, preserving its metadata.
219    ///
220    /// # Panics
221    ///
222    /// Panics if an instrument subscription receives an empty identifier list.
223    #[must_use]
224    pub fn with_instrument_ids(&self, instrument_ids: impl IntoInstrumentIds) -> Option<Self> {
225        match self {
226            Self::Instrument(command) => Some(Self::Instrument(
227                command.with_instrument_ids(instrument_ids),
228            )),
229            Self::BookDeltas(command) => Some(Self::BookDeltas(
230                command.with_instrument_ids(instrument_ids),
231            )),
232            Self::BookDepth(command) => {
233                Some(Self::BookDepth(command.with_instrument_ids(instrument_ids)))
234            }
235            Self::BookSnapshots(command) => Some(Self::BookSnapshots(
236                command.with_instrument_ids(instrument_ids),
237            )),
238            Self::Quotes(command) => {
239                Some(Self::Quotes(command.with_instrument_ids(instrument_ids)))
240            }
241            Self::Trades(command) => {
242                Some(Self::Trades(command.with_instrument_ids(instrument_ids)))
243            }
244            Self::MarkPrices(command) => Some(Self::MarkPrices(
245                command.with_instrument_ids(instrument_ids),
246            )),
247            Self::IndexPrices(command) => Some(Self::IndexPrices(
248                command.with_instrument_ids(instrument_ids),
249            )),
250            Self::FundingRates(command) => Some(Self::FundingRates(
251                command.with_instrument_ids(instrument_ids),
252            )),
253            Self::InstrumentStatus(command) => Some(Self::InstrumentStatus(
254                command.with_instrument_ids(instrument_ids),
255            )),
256            Self::InstrumentClose(command) => Some(Self::InstrumentClose(
257                command.with_instrument_ids(instrument_ids),
258            )),
259            Self::OptionGreeks(command) => Some(Self::OptionGreeks(
260                command.with_instrument_ids(instrument_ids),
261            )),
262            _ => None,
263        }
264    }
265
266    /// Returns the bar types for a bar subscription, if applicable.
267    #[must_use]
268    pub fn bar_types(&self) -> Option<&[BarType]> {
269        match self {
270            Self::Bars(command) => Some(&command.bar_types),
271            _ => None,
272        }
273    }
274
275    /// Projects a bar subscription onto the supplied types, preserving its metadata.
276    ///
277    /// # Panics
278    ///
279    /// Panics if a bar subscription receives an empty list.
280    #[must_use]
281    pub fn with_bar_types(&self, bar_types: impl IntoBarTypes) -> Option<Self> {
282        match self {
283            Self::Bars(command) => Some(Self::Bars(command.with_bar_types(bar_types))),
284            _ => None,
285        }
286    }
287
288    /// Converts the command to a dyn Any trait object for messaging.
289    pub fn as_any(&self) -> &dyn Any {
290        self
291    }
292
293    /// Converts this subscribe command into its matching unsubscribe command.
294    ///
295    /// Preserves the subscribed data identity and client route while replacing the command ID and
296    /// initialization timestamp. It also preserves parameters and sets the supplied correlation ID
297    /// when the matching unsubscribe command supports those fields.
298    #[must_use]
299    pub fn into_unsubscribe(
300        self,
301        command_id: UUID4,
302        ts_init: UnixNanos,
303        correlation_id: Option<UUID4>,
304    ) -> UnsubscribeCommand {
305        match self {
306            Self::Data(cmd) => UnsubscribeCommand::Data(UnsubscribeCustomData::new(
307                cmd.client_id,
308                cmd.venue,
309                cmd.data_type,
310                command_id,
311                ts_init,
312                correlation_id,
313                cmd.params,
314            )),
315            Self::Instrument(cmd) => UnsubscribeCommand::Instrument(UnsubscribeInstrument::new(
316                cmd.instrument_ids,
317                cmd.client_id,
318                cmd.venue,
319                command_id,
320                ts_init,
321                correlation_id,
322                cmd.params,
323            )),
324            Self::Instruments(cmd) => UnsubscribeCommand::Instruments(UnsubscribeInstruments::new(
325                cmd.client_id,
326                cmd.venue,
327                command_id,
328                ts_init,
329                correlation_id,
330                cmd.params,
331            )),
332            Self::BookDeltas(cmd) => UnsubscribeCommand::BookDeltas(UnsubscribeBookDeltas::new(
333                cmd.instrument_ids,
334                cmd.client_id,
335                cmd.venue,
336                command_id,
337                ts_init,
338                correlation_id,
339                cmd.params,
340            )),
341            Self::BookDepth(cmd) => UnsubscribeCommand::BookDepth(UnsubscribeBookDepth::new(
342                cmd.instrument_ids,
343                cmd.client_id,
344                cmd.venue,
345                command_id,
346                ts_init,
347                correlation_id,
348                cmd.params,
349            )),
350            Self::BookSnapshots(cmd) => {
351                UnsubscribeCommand::BookSnapshots(UnsubscribeBookSnapshots::new(
352                    cmd.instrument_ids,
353                    cmd.interval_ms,
354                    cmd.client_id,
355                    cmd.venue,
356                    command_id,
357                    ts_init,
358                    correlation_id,
359                    cmd.params,
360                ))
361            }
362            Self::Quotes(cmd) => UnsubscribeCommand::Quotes(UnsubscribeQuotes::new(
363                cmd.instrument_ids,
364                cmd.client_id,
365                cmd.venue,
366                command_id,
367                ts_init,
368                correlation_id,
369                cmd.params,
370            )),
371            Self::Trades(cmd) => UnsubscribeCommand::Trades(UnsubscribeTrades::new(
372                cmd.instrument_ids,
373                cmd.client_id,
374                cmd.venue,
375                command_id,
376                ts_init,
377                correlation_id,
378                cmd.params,
379            )),
380            Self::Bars(cmd) => UnsubscribeCommand::Bars(UnsubscribeBars::new(
381                cmd.bar_types,
382                cmd.client_id,
383                cmd.venue,
384                command_id,
385                ts_init,
386                correlation_id,
387                cmd.params,
388            )),
389            Self::MarkPrices(cmd) => UnsubscribeCommand::MarkPrices(UnsubscribeMarkPrices::new(
390                cmd.instrument_ids,
391                cmd.client_id,
392                cmd.venue,
393                command_id,
394                ts_init,
395                correlation_id,
396                cmd.params,
397            )),
398            Self::IndexPrices(cmd) => UnsubscribeCommand::IndexPrices(UnsubscribeIndexPrices::new(
399                cmd.instrument_ids,
400                cmd.client_id,
401                cmd.venue,
402                command_id,
403                ts_init,
404                correlation_id,
405                cmd.params,
406            )),
407            Self::FundingRates(cmd) => {
408                UnsubscribeCommand::FundingRates(UnsubscribeFundingRates::new(
409                    cmd.instrument_ids,
410                    cmd.client_id,
411                    cmd.venue,
412                    command_id,
413                    ts_init,
414                    correlation_id,
415                    cmd.params,
416                ))
417            }
418            Self::InstrumentStatus(cmd) => {
419                UnsubscribeCommand::InstrumentStatus(UnsubscribeInstrumentStatus::new(
420                    cmd.instrument_ids,
421                    cmd.client_id,
422                    cmd.venue,
423                    command_id,
424                    ts_init,
425                    correlation_id,
426                    cmd.params,
427                ))
428            }
429            Self::InstrumentClose(cmd) => {
430                UnsubscribeCommand::InstrumentClose(UnsubscribeInstrumentClose::new(
431                    cmd.instrument_ids,
432                    cmd.client_id,
433                    cmd.venue,
434                    command_id,
435                    ts_init,
436                    correlation_id,
437                    cmd.params,
438                ))
439            }
440            Self::OptionGreeks(cmd) => {
441                UnsubscribeCommand::OptionGreeks(UnsubscribeOptionGreeks::new(
442                    cmd.instrument_ids,
443                    cmd.client_id,
444                    cmd.venue,
445                    command_id,
446                    ts_init,
447                    correlation_id,
448                    cmd.params,
449                ))
450            }
451            Self::OptionChain(cmd) => {
452                let mut unsubscribe = UnsubscribeOptionChain::new(
453                    cmd.series_id,
454                    command_id,
455                    ts_init,
456                    cmd.client_id,
457                    cmd.venue,
458                );
459                unsubscribe.params = cmd.params;
460                UnsubscribeCommand::OptionChain(unsubscribe)
461            }
462        }
463    }
464
465    pub fn command_id(&self) -> UUID4 {
466        match self {
467            Self::Data(cmd) => cmd.command_id,
468            Self::Instrument(cmd) => cmd.command_id,
469            Self::Instruments(cmd) => cmd.command_id,
470            Self::BookDeltas(cmd) => cmd.command_id,
471            Self::BookDepth(cmd) => cmd.command_id,
472            Self::BookSnapshots(cmd) => cmd.command_id,
473            Self::Quotes(cmd) => cmd.command_id,
474            Self::Trades(cmd) => cmd.command_id,
475            Self::Bars(cmd) => cmd.command_id,
476            Self::MarkPrices(cmd) => cmd.command_id,
477            Self::IndexPrices(cmd) => cmd.command_id,
478            Self::FundingRates(cmd) => cmd.command_id,
479            Self::InstrumentStatus(cmd) => cmd.command_id,
480            Self::InstrumentClose(cmd) => cmd.command_id,
481            Self::OptionGreeks(cmd) => cmd.command_id,
482            Self::OptionChain(cmd) => cmd.command_id,
483        }
484    }
485
486    pub fn client_id(&self) -> Option<&ClientId> {
487        match self {
488            Self::Data(cmd) => cmd.client_id.as_ref(),
489            Self::Instrument(cmd) => cmd.client_id.as_ref(),
490            Self::Instruments(cmd) => cmd.client_id.as_ref(),
491            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
492            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
493            Self::BookSnapshots(cmd) => cmd.client_id.as_ref(),
494            Self::Quotes(cmd) => cmd.client_id.as_ref(),
495            Self::Trades(cmd) => cmd.client_id.as_ref(),
496            Self::MarkPrices(cmd) => cmd.client_id.as_ref(),
497            Self::IndexPrices(cmd) => cmd.client_id.as_ref(),
498            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
499            Self::Bars(cmd) => cmd.client_id.as_ref(),
500            Self::InstrumentStatus(cmd) => cmd.client_id.as_ref(),
501            Self::InstrumentClose(cmd) => cmd.client_id.as_ref(),
502            Self::OptionGreeks(cmd) => cmd.client_id.as_ref(),
503            Self::OptionChain(cmd) => cmd.client_id.as_ref(),
504        }
505    }
506
507    pub fn venue(&self) -> Option<&Venue> {
508        match self {
509            Self::Data(cmd) => cmd.venue.as_ref(),
510            Self::Instrument(cmd) => cmd.venue.as_ref(),
511            Self::Instruments(cmd) => Some(&cmd.venue),
512            Self::BookDeltas(cmd) => cmd.venue.as_ref(),
513            Self::BookDepth(cmd) => cmd.venue.as_ref(),
514            Self::BookSnapshots(cmd) => cmd.venue.as_ref(),
515            Self::Quotes(cmd) => cmd.venue.as_ref(),
516            Self::Trades(cmd) => cmd.venue.as_ref(),
517            Self::MarkPrices(cmd) => cmd.venue.as_ref(),
518            Self::IndexPrices(cmd) => cmd.venue.as_ref(),
519            Self::FundingRates(cmd) => cmd.venue.as_ref(),
520            Self::Bars(cmd) => cmd.venue.as_ref(),
521            Self::InstrumentStatus(cmd) => cmd.venue.as_ref(),
522            Self::InstrumentClose(cmd) => cmd.venue.as_ref(),
523            Self::OptionGreeks(cmd) => cmd.venue.as_ref(),
524            Self::OptionChain(cmd) => cmd.venue.as_ref(),
525        }
526    }
527
528    pub fn ts_init(&self) -> UnixNanos {
529        match self {
530            Self::Data(cmd) => cmd.ts_init,
531            Self::Instrument(cmd) => cmd.ts_init,
532            Self::Instruments(cmd) => cmd.ts_init,
533            Self::BookDeltas(cmd) => cmd.ts_init,
534            Self::BookDepth(cmd) => cmd.ts_init,
535            Self::BookSnapshots(cmd) => cmd.ts_init,
536            Self::Quotes(cmd) => cmd.ts_init,
537            Self::Trades(cmd) => cmd.ts_init,
538            Self::MarkPrices(cmd) => cmd.ts_init,
539            Self::IndexPrices(cmd) => cmd.ts_init,
540            Self::FundingRates(cmd) => cmd.ts_init,
541            Self::Bars(cmd) => cmd.ts_init,
542            Self::InstrumentStatus(cmd) => cmd.ts_init,
543            Self::InstrumentClose(cmd) => cmd.ts_init,
544            Self::OptionGreeks(cmd) => cmd.ts_init,
545            Self::OptionChain(cmd) => cmd.ts_init,
546        }
547    }
548
549    pub fn correlation_id(&self) -> Option<UUID4> {
550        match self {
551            Self::Data(cmd) => cmd.correlation_id,
552            Self::Instrument(cmd) => cmd.correlation_id,
553            Self::Instruments(cmd) => cmd.correlation_id,
554            Self::BookDeltas(cmd) => cmd.correlation_id,
555            Self::BookDepth(cmd) => cmd.correlation_id,
556            Self::BookSnapshots(cmd) => cmd.correlation_id,
557            Self::Quotes(cmd) => cmd.correlation_id,
558            Self::Trades(cmd) => cmd.correlation_id,
559            Self::MarkPrices(cmd) => cmd.correlation_id,
560            Self::IndexPrices(cmd) => cmd.correlation_id,
561            Self::FundingRates(cmd) => cmd.correlation_id,
562            Self::Bars(cmd) => cmd.correlation_id,
563            Self::InstrumentStatus(cmd) => cmd.correlation_id,
564            Self::InstrumentClose(cmd) => cmd.correlation_id,
565            Self::OptionGreeks(cmd) => cmd.correlation_id,
566            Self::OptionChain(cmd) => cmd.correlation_id,
567        }
568    }
569
570    pub fn params(&self) -> Option<&Params> {
571        match self {
572            Self::Data(cmd) => cmd.params.as_ref(),
573            Self::Instrument(cmd) => cmd.params.as_ref(),
574            Self::Instruments(cmd) => cmd.params.as_ref(),
575            Self::BookDeltas(cmd) => cmd.params.as_ref(),
576            Self::BookDepth(cmd) => cmd.params.as_ref(),
577            Self::BookSnapshots(cmd) => cmd.params.as_ref(),
578            Self::Quotes(cmd) => cmd.params.as_ref(),
579            Self::Trades(cmd) => cmd.params.as_ref(),
580            Self::Bars(cmd) => cmd.params.as_ref(),
581            Self::MarkPrices(cmd) => cmd.params.as_ref(),
582            Self::IndexPrices(cmd) => cmd.params.as_ref(),
583            Self::FundingRates(cmd) => cmd.params.as_ref(),
584            Self::InstrumentStatus(cmd) => cmd.params.as_ref(),
585            Self::InstrumentClose(cmd) => cmd.params.as_ref(),
586            Self::OptionGreeks(cmd) => cmd.params.as_ref(),
587            Self::OptionChain(cmd) => cmd.params.as_ref(),
588        }
589    }
590}
591
592#[derive(Clone, Debug, Serialize, Deserialize)]
593pub enum UnsubscribeCommand {
594    Data(UnsubscribeCustomData),
595    Instrument(UnsubscribeInstrument),
596    Instruments(UnsubscribeInstruments),
597    BookDeltas(UnsubscribeBookDeltas),
598    BookDepth(UnsubscribeBookDepth),
599    BookSnapshots(UnsubscribeBookSnapshots),
600    Quotes(UnsubscribeQuotes),
601    Trades(UnsubscribeTrades),
602    Bars(UnsubscribeBars),
603    MarkPrices(UnsubscribeMarkPrices),
604    IndexPrices(UnsubscribeIndexPrices),
605    FundingRates(UnsubscribeFundingRates),
606    InstrumentStatus(UnsubscribeInstrumentStatus),
607    InstrumentClose(UnsubscribeInstrumentClose),
608    OptionGreeks(UnsubscribeOptionGreeks),
609    OptionChain(UnsubscribeOptionChain),
610}
611
612impl PartialEq for UnsubscribeCommand {
613    fn eq(&self, other: &Self) -> bool {
614        self.command_id() == other.command_id()
615    }
616}
617
618impl UnsubscribeCommand {
619    /// Returns the instrument IDs for an instrument subscription, if applicable.
620    #[must_use]
621    pub fn instrument_ids(&self) -> Option<&[InstrumentId]> {
622        match self {
623            Self::Instrument(command) => Some(&command.instrument_ids),
624            Self::BookDeltas(command) => Some(&command.instrument_ids),
625            Self::BookDepth(command) => Some(&command.instrument_ids),
626            Self::BookSnapshots(command) => Some(&command.instrument_ids),
627            Self::Quotes(command) => Some(&command.instrument_ids),
628            Self::Trades(command) => Some(&command.instrument_ids),
629            Self::MarkPrices(command) => Some(&command.instrument_ids),
630            Self::IndexPrices(command) => Some(&command.instrument_ids),
631            Self::FundingRates(command) => Some(&command.instrument_ids),
632            Self::InstrumentStatus(command) => Some(&command.instrument_ids),
633            Self::InstrumentClose(command) => Some(&command.instrument_ids),
634            Self::OptionGreeks(command) => Some(&command.instrument_ids),
635            _ => None,
636        }
637    }
638
639    /// Projects an instrument subscription onto the supplied IDs, preserving its metadata.
640    ///
641    /// # Panics
642    ///
643    /// Panics if an instrument subscription receives an empty identifier list.
644    #[must_use]
645    pub fn with_instrument_ids(&self, instrument_ids: impl IntoInstrumentIds) -> Option<Self> {
646        match self {
647            Self::Instrument(command) => Some(Self::Instrument(
648                command.with_instrument_ids(instrument_ids),
649            )),
650            Self::BookDeltas(command) => Some(Self::BookDeltas(
651                command.with_instrument_ids(instrument_ids),
652            )),
653            Self::BookDepth(command) => {
654                Some(Self::BookDepth(command.with_instrument_ids(instrument_ids)))
655            }
656            Self::BookSnapshots(command) => Some(Self::BookSnapshots(
657                command.with_instrument_ids(instrument_ids),
658            )),
659            Self::Quotes(command) => {
660                Some(Self::Quotes(command.with_instrument_ids(instrument_ids)))
661            }
662            Self::Trades(command) => {
663                Some(Self::Trades(command.with_instrument_ids(instrument_ids)))
664            }
665            Self::MarkPrices(command) => Some(Self::MarkPrices(
666                command.with_instrument_ids(instrument_ids),
667            )),
668            Self::IndexPrices(command) => Some(Self::IndexPrices(
669                command.with_instrument_ids(instrument_ids),
670            )),
671            Self::FundingRates(command) => Some(Self::FundingRates(
672                command.with_instrument_ids(instrument_ids),
673            )),
674            Self::InstrumentStatus(command) => Some(Self::InstrumentStatus(
675                command.with_instrument_ids(instrument_ids),
676            )),
677            Self::InstrumentClose(command) => Some(Self::InstrumentClose(
678                command.with_instrument_ids(instrument_ids),
679            )),
680            Self::OptionGreeks(command) => Some(Self::OptionGreeks(
681                command.with_instrument_ids(instrument_ids),
682            )),
683            _ => None,
684        }
685    }
686
687    /// Returns the bar types for a bar subscription, if applicable.
688    #[must_use]
689    pub fn bar_types(&self) -> Option<&[BarType]> {
690        match self {
691            Self::Bars(command) => Some(&command.bar_types),
692            _ => None,
693        }
694    }
695
696    /// Projects a bar subscription onto the supplied types, preserving its metadata.
697    ///
698    /// # Panics
699    ///
700    /// Panics if a bar subscription receives an empty list.
701    #[must_use]
702    pub fn with_bar_types(&self, bar_types: impl IntoBarTypes) -> Option<Self> {
703        match self {
704            Self::Bars(command) => Some(Self::Bars(command.with_bar_types(bar_types))),
705            _ => None,
706        }
707    }
708
709    /// Groups instrument releases with matching routing and command metadata.
710    ///
711    /// Bar releases with matching metadata merge into one command, with duplicate types removed.
712    /// Other commands remain separate.
713    ///
714    /// # Panics
715    ///
716    /// Panics if an instrument or bar release group has no members.
717    #[must_use]
718    pub fn group(commands: Vec<Self>) -> Vec<Self> {
719        let mut groups: Vec<(Self, Vec<InstrumentId>)> = Vec::new();
720
721        for command in commands {
722            let Some(ids) = command.instrument_ids() else {
723                if groups
724                    .iter_mut()
725                    .any(|(batch, _)| batch.merge_bars(&command))
726                {
727                    continue;
728                }
729
730                groups.push((command, Vec::new()));
731                continue;
732            };
733
734            if let Some((_, grouped_ids)) = groups.iter_mut().find(|(batch, _)| {
735                batch.command_id() == command.command_id()
736                    && batch.ts_init() == command.ts_init()
737                    && batch.correlation_id() == command.correlation_id()
738                    && batch.client_id() == command.client_id()
739                    && batch.venue() == command.venue()
740                    && match (batch, &command) {
741                        (Self::Instrument(batch), Self::Instrument(member)) => {
742                            batch.params == member.params
743                        }
744                        (Self::BookDeltas(batch), Self::BookDeltas(member)) => {
745                            batch.params == member.params
746                        }
747                        (Self::BookDepth(batch), Self::BookDepth(member)) => {
748                            batch.params == member.params
749                        }
750                        (Self::BookSnapshots(batch), Self::BookSnapshots(member)) => {
751                            batch.params == member.params && batch.interval_ms == member.interval_ms
752                        }
753                        (Self::Quotes(batch), Self::Quotes(member)) => {
754                            batch.params == member.params
755                        }
756                        (Self::Trades(batch), Self::Trades(member)) => {
757                            batch.params == member.params
758                        }
759                        (Self::MarkPrices(batch), Self::MarkPrices(member)) => {
760                            batch.params == member.params
761                        }
762                        (Self::IndexPrices(batch), Self::IndexPrices(member)) => {
763                            batch.params == member.params
764                        }
765                        (Self::FundingRates(batch), Self::FundingRates(member)) => {
766                            batch.params == member.params
767                        }
768                        (Self::InstrumentStatus(batch), Self::InstrumentStatus(member)) => {
769                            batch.params == member.params
770                        }
771                        (Self::InstrumentClose(batch), Self::InstrumentClose(member)) => {
772                            batch.params == member.params
773                        }
774                        (Self::OptionGreeks(batch), Self::OptionGreeks(member)) => {
775                            batch.params == member.params
776                        }
777                        _ => false,
778                    }
779            }) {
780                grouped_ids.extend_from_slice(ids);
781                continue;
782            }
783
784            let ids = ids.to_vec();
785            groups.push((command, ids));
786        }
787
788        groups
789            .into_iter()
790            .map(|(mut command, ids)| {
791                if let Self::Bars(batch) = &mut command {
792                    batch.bar_types = std::mem::take(&mut batch.bar_types)
793                        .into_bar_types()
794                        .expect("merged bar releases are nonempty");
795                    command
796                } else {
797                    command.with_instrument_ids(ids).unwrap_or(command)
798                }
799            })
800            .collect()
801    }
802
803    fn merge_bars(&mut self, other: &Self) -> bool {
804        let (Self::Bars(batch), Self::Bars(member)) = (self, other) else {
805            return false;
806        };
807
808        if batch.command_id != member.command_id
809            || batch.ts_init != member.ts_init
810            || batch.correlation_id != member.correlation_id
811            || batch.client_id != member.client_id
812            || batch.venue != member.venue
813            || batch.params != member.params
814        {
815            return false;
816        }
817
818        batch.bar_types.extend_from_slice(&member.bar_types);
819        true
820    }
821
822    /// Converts the command to a dyn Any trait object for messaging.
823    pub fn as_any(&self) -> &dyn Any {
824        self
825    }
826
827    pub fn command_id(&self) -> UUID4 {
828        match self {
829            Self::Data(cmd) => cmd.command_id,
830            Self::Instrument(cmd) => cmd.command_id,
831            Self::Instruments(cmd) => cmd.command_id,
832            Self::BookDeltas(cmd) => cmd.command_id,
833            Self::BookDepth(cmd) => cmd.command_id,
834            Self::BookSnapshots(cmd) => cmd.command_id,
835            Self::Quotes(cmd) => cmd.command_id,
836            Self::Trades(cmd) => cmd.command_id,
837            Self::Bars(cmd) => cmd.command_id,
838            Self::MarkPrices(cmd) => cmd.command_id,
839            Self::IndexPrices(cmd) => cmd.command_id,
840            Self::FundingRates(cmd) => cmd.command_id,
841            Self::InstrumentStatus(cmd) => cmd.command_id,
842            Self::InstrumentClose(cmd) => cmd.command_id,
843            Self::OptionGreeks(cmd) => cmd.command_id,
844            Self::OptionChain(cmd) => cmd.command_id,
845        }
846    }
847
848    pub fn client_id(&self) -> Option<&ClientId> {
849        match self {
850            Self::Data(cmd) => cmd.client_id.as_ref(),
851            Self::Instrument(cmd) => cmd.client_id.as_ref(),
852            Self::Instruments(cmd) => cmd.client_id.as_ref(),
853            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
854            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
855            Self::BookSnapshots(cmd) => cmd.client_id.as_ref(),
856            Self::Quotes(cmd) => cmd.client_id.as_ref(),
857            Self::Trades(cmd) => cmd.client_id.as_ref(),
858            Self::Bars(cmd) => cmd.client_id.as_ref(),
859            Self::MarkPrices(cmd) => cmd.client_id.as_ref(),
860            Self::IndexPrices(cmd) => cmd.client_id.as_ref(),
861            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
862            Self::InstrumentStatus(cmd) => cmd.client_id.as_ref(),
863            Self::InstrumentClose(cmd) => cmd.client_id.as_ref(),
864            Self::OptionGreeks(cmd) => cmd.client_id.as_ref(),
865            Self::OptionChain(cmd) => cmd.client_id.as_ref(),
866        }
867    }
868
869    pub fn venue(&self) -> Option<&Venue> {
870        match self {
871            Self::Data(cmd) => cmd.venue.as_ref(),
872            Self::Instrument(cmd) => cmd.venue.as_ref(),
873            Self::Instruments(cmd) => Some(&cmd.venue),
874            Self::BookDeltas(cmd) => cmd.venue.as_ref(),
875            Self::BookDepth(cmd) => cmd.venue.as_ref(),
876            Self::BookSnapshots(cmd) => cmd.venue.as_ref(),
877            Self::Quotes(cmd) => cmd.venue.as_ref(),
878            Self::Trades(cmd) => cmd.venue.as_ref(),
879            Self::Bars(cmd) => cmd.venue.as_ref(),
880            Self::MarkPrices(cmd) => cmd.venue.as_ref(),
881            Self::IndexPrices(cmd) => cmd.venue.as_ref(),
882            Self::FundingRates(cmd) => cmd.venue.as_ref(),
883            Self::InstrumentStatus(cmd) => cmd.venue.as_ref(),
884            Self::InstrumentClose(cmd) => cmd.venue.as_ref(),
885            Self::OptionGreeks(cmd) => cmd.venue.as_ref(),
886            Self::OptionChain(cmd) => cmd.venue.as_ref(),
887        }
888    }
889
890    pub fn ts_init(&self) -> UnixNanos {
891        match self {
892            Self::Data(cmd) => cmd.ts_init,
893            Self::Instrument(cmd) => cmd.ts_init,
894            Self::Instruments(cmd) => cmd.ts_init,
895            Self::BookDeltas(cmd) => cmd.ts_init,
896            Self::BookDepth(cmd) => cmd.ts_init,
897            Self::BookSnapshots(cmd) => cmd.ts_init,
898            Self::Quotes(cmd) => cmd.ts_init,
899            Self::Trades(cmd) => cmd.ts_init,
900            Self::MarkPrices(cmd) => cmd.ts_init,
901            Self::IndexPrices(cmd) => cmd.ts_init,
902            Self::FundingRates(cmd) => cmd.ts_init,
903            Self::Bars(cmd) => cmd.ts_init,
904            Self::InstrumentStatus(cmd) => cmd.ts_init,
905            Self::InstrumentClose(cmd) => cmd.ts_init,
906            Self::OptionGreeks(cmd) => cmd.ts_init,
907            Self::OptionChain(cmd) => cmd.ts_init,
908        }
909    }
910
911    pub fn correlation_id(&self) -> Option<UUID4> {
912        match self {
913            Self::Data(cmd) => cmd.correlation_id,
914            Self::Instrument(cmd) => cmd.correlation_id,
915            Self::Instruments(cmd) => cmd.correlation_id,
916            Self::BookDeltas(cmd) => cmd.correlation_id,
917            Self::BookDepth(cmd) => cmd.correlation_id,
918            Self::BookSnapshots(cmd) => cmd.correlation_id,
919            Self::Quotes(cmd) => cmd.correlation_id,
920            Self::Trades(cmd) => cmd.correlation_id,
921            Self::MarkPrices(cmd) => cmd.correlation_id,
922            Self::IndexPrices(cmd) => cmd.correlation_id,
923            Self::FundingRates(cmd) => cmd.correlation_id,
924            Self::Bars(cmd) => cmd.correlation_id,
925            Self::InstrumentStatus(cmd) => cmd.correlation_id,
926            Self::InstrumentClose(cmd) => cmd.correlation_id,
927            Self::OptionGreeks(cmd) => cmd.correlation_id,
928            Self::OptionChain(_) => None,
929        }
930    }
931}
932
933#[allow(
934    clippy::ref_option,
935    reason = "callers pass borrowed Option fields directly"
936)]
937fn check_client_id_or_venue(client_id: &Option<ClientId>, venue: &Option<Venue>) {
938    assert!(
939        client_id.is_some() || venue.is_some(),
940        "Both `client_id` and `venue` were None"
941    );
942}
943
944#[derive(Clone, Debug, Serialize, Deserialize)]
945pub enum RequestCommand {
946    Data(RequestCustomData),
947    Instrument(RequestInstrument),
948    Instruments(RequestInstruments),
949    BookSnapshot(RequestBookSnapshot),
950    BookDeltas(RequestBookDeltas),
951    BookDepth(RequestBookDepth),
952    Quotes(RequestQuotes),
953    Trades(RequestTrades),
954    FundingRates(RequestFundingRates),
955    OptionChainReferencePrice(RequestOptionChainReferencePrice),
956    Bars(RequestBars),
957    Join(RequestJoin),
958}
959
960impl PartialEq for RequestCommand {
961    fn eq(&self, other: &Self) -> bool {
962        self.request_id() == other.request_id()
963    }
964}
965
966impl RequestCommand {
967    /// Converts the command to a dyn Any trait object for messaging.
968    pub fn as_any(&self) -> &dyn Any {
969        self
970    }
971
972    pub fn request_id(&self) -> &UUID4 {
973        match self {
974            Self::Data(cmd) => &cmd.request_id,
975            Self::Instrument(cmd) => &cmd.request_id,
976            Self::Instruments(cmd) => &cmd.request_id,
977            Self::BookSnapshot(cmd) => &cmd.request_id,
978            Self::BookDeltas(cmd) => &cmd.request_id,
979            Self::BookDepth(cmd) => &cmd.request_id,
980            Self::Quotes(cmd) => &cmd.request_id,
981            Self::Trades(cmd) => &cmd.request_id,
982            Self::FundingRates(cmd) => &cmd.request_id,
983            Self::OptionChainReferencePrice(cmd) => &cmd.request_id,
984            Self::Bars(cmd) => &cmd.request_id,
985            Self::Join(cmd) => &cmd.request_id,
986        }
987    }
988
989    pub fn client_id(&self) -> Option<&ClientId> {
990        match self {
991            Self::Data(cmd) => Some(&cmd.client_id),
992            Self::Instrument(cmd) => cmd.client_id.as_ref(),
993            Self::Instruments(cmd) => cmd.client_id.as_ref(),
994            Self::BookSnapshot(cmd) => cmd.client_id.as_ref(),
995            Self::BookDeltas(cmd) => cmd.client_id.as_ref(),
996            Self::BookDepth(cmd) => cmd.client_id.as_ref(),
997            Self::Quotes(cmd) => cmd.client_id.as_ref(),
998            Self::Trades(cmd) => cmd.client_id.as_ref(),
999            Self::FundingRates(cmd) => cmd.client_id.as_ref(),
1000            Self::OptionChainReferencePrice(cmd) => cmd.client_id.as_ref(),
1001            Self::Bars(cmd) => cmd.client_id.as_ref(),
1002            Self::Join(_) => None,
1003        }
1004    }
1005
1006    pub fn venue(&self) -> Option<&Venue> {
1007        match self {
1008            Self::Data(_) => None,
1009            Self::Instrument(cmd) => first_venue(&cmd.instrument_ids),
1010            Self::Instruments(cmd) => cmd.venue.as_ref(),
1011            Self::BookSnapshot(cmd) => first_venue(&cmd.instrument_ids),
1012            Self::BookDeltas(cmd) => first_venue(&cmd.instrument_ids),
1013            Self::BookDepth(cmd) => first_venue(&cmd.instrument_ids),
1014            Self::Quotes(cmd) => first_venue(&cmd.instrument_ids),
1015            Self::Trades(cmd) => first_venue(&cmd.instrument_ids),
1016            Self::FundingRates(cmd) => first_venue(&cmd.instrument_ids),
1017            Self::OptionChainReferencePrice(cmd) => Some(&cmd.series_id.venue),
1018            // TODO: Extract the below somewhere
1019            Self::Bars(cmd) => match cmd.bar_types.first()? {
1020                BarType::Standard { instrument_id, .. } => Some(&instrument_id.venue),
1021                BarType::Composite { instrument_id, .. } => Some(&instrument_id.venue),
1022            },
1023            Self::Join(_) => None,
1024        }
1025    }
1026
1027    pub fn ts_init(&self) -> UnixNanos {
1028        match self {
1029            Self::Data(cmd) => cmd.ts_init,
1030            Self::Instrument(cmd) => cmd.ts_init,
1031            Self::Instruments(cmd) => cmd.ts_init,
1032            Self::BookSnapshot(cmd) => cmd.ts_init,
1033            Self::BookDeltas(cmd) => cmd.ts_init,
1034            Self::BookDepth(cmd) => cmd.ts_init,
1035            Self::Quotes(cmd) => cmd.ts_init,
1036            Self::Trades(cmd) => cmd.ts_init,
1037            Self::FundingRates(cmd) => cmd.ts_init,
1038            Self::OptionChainReferencePrice(cmd) => cmd.ts_init,
1039            Self::Bars(cmd) => cmd.ts_init,
1040            Self::Join(cmd) => cmd.ts_init,
1041        }
1042    }
1043}
1044
1045fn first_venue(ids: &[InstrumentId]) -> Option<&Venue> {
1046    ids.first().map(|id| &id.venue)
1047}
1048
1049#[derive(Clone, Debug)]
1050pub enum DataResponse {
1051    Data(CustomDataResponse),
1052    Instrument(Box<InstrumentResponse>),
1053    Instruments(InstrumentsResponse),
1054    Book(BookResponse),
1055    BookDeltas(BookDeltasResponse),
1056    BookDepth(BookDepthResponse),
1057    Quotes(QuotesResponse),
1058    Trades(TradesResponse),
1059    FundingRates(FundingRatesResponse),
1060    OptionChainReferencePrice(OptionChainReferencePriceResponse),
1061    Bars(BarsResponse),
1062}
1063
1064/// A historical payload and the completion state of its root request.
1065///
1066/// A terminal envelope can omit its payload, including when a request fails before returning data.
1067/// `sub_correlation_id` identifies a source child without changing the root correlation ID.
1068#[derive(Clone, Debug, Serialize)]
1069pub struct DataResponseEnvelope {
1070    correlation_id: UUID4,
1071    response: Option<DataResponse>,
1072    completed: bool,
1073    sub_correlation_id: Option<UUID4>,
1074    failure_reason: Option<String>,
1075}
1076
1077impl DataResponseEnvelope {
1078    /// Creates an envelope containing a payload.
1079    #[must_use]
1080    pub fn new(response: DataResponse, completed: bool, sub_correlation_id: Option<UUID4>) -> Self {
1081        Self {
1082            correlation_id: *response.correlation_id(),
1083            response: Some(response),
1084            completed,
1085            sub_correlation_id,
1086            failure_reason: None,
1087        }
1088    }
1089
1090    /// Completes a request without a payload. A reason marks the request as failed.
1091    #[must_use]
1092    pub fn terminal(correlation_id: UUID4, failure_reason: Option<String>) -> Self {
1093        Self {
1094            correlation_id,
1095            response: None,
1096            completed: true,
1097            sub_correlation_id: None,
1098            failure_reason,
1099        }
1100    }
1101
1102    /// Sets the root request ID while preserving the payload correlation ID.
1103    #[must_use]
1104    pub fn with_correlation_id(mut self, correlation_id: UUID4) -> Self {
1105        self.correlation_id = correlation_id;
1106        self
1107    }
1108
1109    /// Completes a failed request while preserving the payload returned by successful work.
1110    #[must_use]
1111    pub fn with_failure(mut self, reason: String) -> Self {
1112        self.completed = true;
1113        self.failure_reason = Some(reason);
1114        self
1115    }
1116
1117    #[must_use]
1118    pub const fn correlation_id(&self) -> &UUID4 {
1119        &self.correlation_id
1120    }
1121
1122    #[must_use]
1123    pub const fn response(&self) -> Option<&DataResponse> {
1124        self.response.as_ref()
1125    }
1126
1127    /// Returns the payload for in-place metadata updates or bounds trimming.
1128    pub const fn response_mut(&mut self) -> Option<&mut DataResponse> {
1129        self.response.as_mut()
1130    }
1131
1132    #[must_use]
1133    pub const fn completed(&self) -> bool {
1134        self.completed
1135    }
1136
1137    #[must_use]
1138    pub const fn sub_correlation_id(&self) -> Option<UUID4> {
1139        self.sub_correlation_id
1140    }
1141
1142    #[must_use]
1143    pub fn failure_reason(&self) -> Option<&str> {
1144        self.failure_reason.as_deref()
1145    }
1146}
1147
1148impl<'de> Deserialize<'de> for DataResponseEnvelope {
1149    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
1150        #[derive(Deserialize)]
1151        struct Wire {
1152            correlation_id: UUID4,
1153            response: Option<DataResponse>,
1154            completed: bool,
1155            sub_correlation_id: Option<UUID4>,
1156            failure_reason: Option<String>,
1157        }
1158
1159        let wire = Wire::deserialize(deserializer)?;
1160        let completed = wire.completed || wire.failure_reason.is_some();
1161
1162        if wire.response.is_none() && !completed {
1163            return Err(serde::de::Error::custom(
1164                "a payload-free response must complete its request",
1165            ));
1166        }
1167
1168        Ok(Self {
1169            correlation_id: wire.correlation_id,
1170            response: wire.response,
1171            completed,
1172            sub_correlation_id: wire.sub_correlation_id,
1173            failure_reason: wire.failure_reason,
1174        })
1175    }
1176}
1177
1178/// An adapter request failure, optionally carrying data collected before the failure.
1179#[derive(Clone, Debug)]
1180pub struct DataRequestFailed {
1181    pub request_id: UUID4,
1182    pub reason: String,
1183    pub response: Option<DataResponse>,
1184}
1185
1186impl DataResponse {
1187    /// Returns the responding client.
1188    #[must_use]
1189    pub fn client_id(&self) -> &ClientId {
1190        match self {
1191            Self::Data(r) => &r.client_id,
1192            Self::Instrument(r) => &r.client_id,
1193            Self::Instruments(r) => &r.client_id,
1194            Self::Book(r) => &r.client_id,
1195            Self::BookDeltas(r) => &r.client_id,
1196            Self::BookDepth(r) => &r.client_id,
1197            Self::Quotes(r) => &r.client_id,
1198            Self::Trades(r) => &r.client_id,
1199            Self::FundingRates(r) => &r.client_id,
1200            Self::OptionChainReferencePrice(r) => &r.client_id,
1201            Self::Bars(r) => &r.client_id,
1202        }
1203    }
1204
1205    /// Returns response parameters.
1206    #[must_use]
1207    pub fn params(&self) -> Option<&Params> {
1208        match self {
1209            Self::Data(r) => r.params.as_ref(),
1210            Self::Instrument(r) => r.params.as_ref(),
1211            Self::Instruments(r) => r.params.as_ref(),
1212            Self::Book(r) => r.params.as_ref(),
1213            Self::BookDeltas(r) => r.params.as_ref(),
1214            Self::BookDepth(r) => r.params.as_ref(),
1215            Self::Quotes(r) => r.params.as_ref(),
1216            Self::Trades(r) => r.params.as_ref(),
1217            Self::FundingRates(r) => r.params.as_ref(),
1218            Self::OptionChainReferencePrice(r) => r.params.as_ref(),
1219            Self::Bars(r) => r.params.as_ref(),
1220        }
1221    }
1222
1223    /// Returns the inclusive start bound.
1224    #[must_use]
1225    pub fn start(&self) -> Option<UnixNanos> {
1226        match self {
1227            Self::Data(r) => r.start,
1228            Self::Instrument(r) => r.start,
1229            Self::Instruments(r) => r.start,
1230            Self::Book(r) => r.start,
1231            Self::BookDeltas(r) => r.start,
1232            Self::BookDepth(r) => r.start,
1233            Self::Quotes(r) => r.start,
1234            Self::Trades(r) => r.start,
1235            Self::FundingRates(r) => r.start,
1236            Self::OptionChainReferencePrice(_) => None,
1237            Self::Bars(r) => r.start,
1238        }
1239    }
1240
1241    /// Returns the inclusive end bound.
1242    #[must_use]
1243    pub fn end(&self) -> Option<UnixNanos> {
1244        match self {
1245            Self::Data(r) => r.end,
1246            Self::Instrument(r) => r.end,
1247            Self::Instruments(r) => r.end,
1248            Self::Book(r) => r.end,
1249            Self::BookDeltas(r) => r.end,
1250            Self::BookDepth(r) => r.end,
1251            Self::Quotes(r) => r.end,
1252            Self::Trades(r) => r.end,
1253            Self::FundingRates(r) => r.end,
1254            Self::OptionChainReferencePrice(_) => None,
1255            Self::Bars(r) => r.end,
1256        }
1257    }
1258
1259    /// Returns the response creation timestamp.
1260    #[must_use]
1261    pub fn ts_init(&self) -> UnixNanos {
1262        match self {
1263            Self::Data(r) => r.ts_init,
1264            Self::Instrument(r) => r.ts_init,
1265            Self::Instruments(r) => r.ts_init,
1266            Self::Book(r) => r.ts_init,
1267            Self::BookDeltas(r) => r.ts_init,
1268            Self::BookDepth(r) => r.ts_init,
1269            Self::Quotes(r) => r.ts_init,
1270            Self::Trades(r) => r.ts_init,
1271            Self::FundingRates(r) => r.ts_init,
1272            Self::OptionChainReferencePrice(r) => r.ts_init,
1273            Self::Bars(r) => r.ts_init,
1274        }
1275    }
1276
1277    /// Returns the payload's catalog data type.
1278    #[must_use]
1279    pub fn data_kind(&self) -> NautilusDataType {
1280        match self {
1281            Self::Data(r) => NautilusDataType::Custom {
1282                type_name: r.data_type.type_name().to_string(),
1283            },
1284            Self::Instrument(_) | Self::Instruments(_) => NautilusDataType::Instrument,
1285            Self::Book(_) | Self::BookDeltas(_) => NautilusDataType::OrderBookDelta,
1286            Self::BookDepth(_) => NautilusDataType::OrderBookDepth,
1287            Self::Quotes(_) => NautilusDataType::QuoteTick,
1288            Self::Trades(_) => NautilusDataType::TradeTick,
1289            Self::FundingRates(_) => NautilusDataType::FundingRateUpdate,
1290            Self::Bars(_) => NautilusDataType::Bar,
1291            Self::OptionChainReferencePrice(_) => NautilusDataType::Custom {
1292                type_name: "OptionChainReferencePrice".to_string(),
1293            },
1294        }
1295    }
1296
1297    /// Converts the command to a dyn Any trait object for messaging.
1298    pub fn as_any(&self) -> &dyn Any {
1299        self
1300    }
1301
1302    pub fn correlation_id(&self) -> &UUID4 {
1303        match self {
1304            Self::Data(resp) => &resp.correlation_id,
1305            Self::Instrument(resp) => &resp.correlation_id,
1306            Self::Instruments(resp) => &resp.correlation_id,
1307            Self::Book(resp) => &resp.correlation_id,
1308            Self::BookDeltas(resp) => &resp.correlation_id,
1309            Self::BookDepth(resp) => &resp.correlation_id,
1310            Self::Quotes(resp) => &resp.correlation_id,
1311            Self::Trades(resp) => &resp.correlation_id,
1312            Self::FundingRates(resp) => &resp.correlation_id,
1313            Self::OptionChainReferencePrice(resp) => &resp.correlation_id,
1314            Self::Bars(resp) => &resp.correlation_id,
1315        }
1316    }
1317
1318    /// Returns a short variant name for compact logging.
1319    #[must_use]
1320    pub fn kind(&self) -> &'static str {
1321        match self {
1322            Self::Data(_) => "Data",
1323            Self::Instrument(_) => "Instrument",
1324            Self::Instruments(_) => "Instruments",
1325            Self::Book(_) => "Book",
1326            Self::BookDeltas(_) => "BookDeltas",
1327            Self::BookDepth(_) => "BookDepth",
1328            Self::Quotes(_) => "Quotes",
1329            Self::Trades(_) => "Trades",
1330            Self::FundingRates(_) => "FundingRates",
1331            Self::OptionChainReferencePrice(_) => "OptionChainReferencePrice",
1332            Self::Bars(_) => "Bars",
1333        }
1334    }
1335
1336    /// Returns the number of records carried by the response, where defined.
1337    ///
1338    /// Returns `None` for singular or opaque variants (`Data`, `Instrument`, `Book`)
1339    /// where a record count is not meaningful.
1340    #[must_use]
1341    pub fn record_count(&self) -> Option<usize> {
1342        match self {
1343            Self::Data(_) | Self::Instrument(_) | Self::Book(_) => None,
1344            Self::Instruments(resp) => Some(resp.data.len()),
1345            Self::BookDeltas(resp) => Some(resp.data.len()),
1346            Self::BookDepth(resp) => Some(resp.data.len()),
1347            Self::Quotes(resp) => Some(resp.data.len()),
1348            Self::Trades(resp) => Some(resp.data.len()),
1349            Self::FundingRates(resp) => Some(resp.data.len()),
1350            Self::OptionChainReferencePrice(_) => None,
1351            Self::Bars(resp) => Some(resp.data.len()),
1352        }
1353    }
1354
1355    /// Trims payloads to the inclusive `[start, end]` window on `ts_init`.
1356    ///
1357    /// Applies to variants whose payload elements implement `HasTsInit`
1358    /// (`BookDeltas`, `BookDepth`, `Quotes`, `Trades`, `FundingRates`,
1359    /// `Bars`, `Instruments`). Other variants are untouched: singular payloads
1360    /// (`Instrument`, `Book`),
1361    /// `OptionChainReferencePrice` (singular), and the opaque custom
1362    /// `Data` variant.
1363    pub fn trim_to_bounds(&mut self) {
1364        match self {
1365            Self::Quotes(r) => response::trim_batch_to_bounds(&mut r.data, r.start, r.end),
1366            Self::Trades(r) => response::trim_batch_to_bounds(&mut r.data, r.start, r.end),
1367            Self::FundingRates(r) => {
1368                response::trim_batch_to_bounds(&mut r.data, r.start, r.end);
1369            }
1370            Self::Bars(r) => response::trim_batch_to_bounds(&mut r.data, r.start, r.end),
1371            Self::Instruments(r) => response::trim_data_to_bounds(&mut r.data, r.start, r.end),
1372            Self::BookDeltas(r) => {
1373                response::trim_batch_to_bounds(&mut r.data, r.start, r.end);
1374            }
1375            Self::BookDepth(r) => response::trim_batch_to_bounds(&mut r.data, r.start, r.end),
1376            Self::Data(_)
1377            | Self::Instrument(_)
1378            | Self::Book(_)
1379            | Self::OptionChainReferencePrice(_) => {}
1380        }
1381    }
1382}
1383
1384pub type Payload = Arc<dyn Any + Send + Sync>;
1385
1386/// Returns `true` when `params` carries the [`PARAMS_IS_PARENT`] flag set to `true`.
1387///
1388/// Absent or non-boolean values resolve to `false`, keeping the default subscription
1389/// path concrete (exact topic).
1390#[must_use]
1391pub fn is_parent_subscription(params: Option<&Params>) -> bool {
1392    params
1393        .and_then(|p| p.get_bool(PARAMS_IS_PARENT))
1394        .unwrap_or(false)
1395}
1396
1397#[cfg(test)]
1398mod tests {
1399    use rstest::rstest;
1400
1401    use super::*;
1402
1403    #[rstest::rstest]
1404    fn failure_envelopes_complete_the_request() {
1405        let response = QuotesResponse::new(
1406            UUID4::new(),
1407            ClientId::from("SOURCE"),
1408            InstrumentId::from("AUD/USD.SIM"),
1409            Vec::new(),
1410            Some(UnixNanos::from(3)),
1411            Some(UnixNanos::from(17)),
1412            UnixNanos::from(29),
1413            None,
1414        );
1415        let source = UUID4::new();
1416        let intermediate =
1417            DataResponseEnvelope::new(DataResponse::Quotes(response), false, Some(source));
1418        let mut expected = serde_json::to_value(&intermediate).unwrap();
1419        expected["completed"] = serde_json::json!(true);
1420        expected["failure_reason"] = serde_json::json!("Missing dataset");
1421        let failure = intermediate.with_failure("Missing dataset".to_owned());
1422        assert_eq!(serde_json::to_value(failure).unwrap(), expected);
1423    }
1424
1425    #[rstest::rstest]
1426    fn deserialized_failure_envelopes_complete_the_request() {
1427        let expected =
1428            DataResponseEnvelope::terminal(UUID4::new(), Some("Missing dataset".to_owned()));
1429        let expected_value = serde_json::to_value(&expected).unwrap();
1430        let mut wire = expected_value.clone();
1431        wire["completed"] = serde_json::json!(false);
1432        let actual: DataResponseEnvelope = serde_json::from_value(wire).unwrap();
1433        assert_eq!(serde_json::to_value(actual).unwrap(), expected_value);
1434    }
1435
1436    macro_rules! identifier_message_tests {
1437        ($projection_test:ident, $release_test:ident, $serde_test:ident, $subscribe:ident, $unsubscribe:ident, $variant:ident) => {
1438            #[rstest]
1439            #[case::subscribe(true)]
1440            #[case::unsubscribe(false)]
1441            fn $projection_test(#[case] subscribe: bool) {
1442                let first = InstrumentId::from("ESM4.GLBX");
1443                let second = InstrumentId::from("NQM4.GLBX");
1444                let command_id = UUID4::new();
1445                let correlation_id = UUID4::new();
1446                let params: Params =
1447                    serde_json::from_value(serde_json::json!({"price_precision": 3})).unwrap();
1448
1449                let command = $subscribe::new(
1450                    [first, second],
1451                    Some(ClientId::from("QUOTE-CLIENT")),
1452                    Some(first.venue),
1453                    command_id,
1454                    UnixNanos::from(37),
1455                    Some(correlation_id),
1456                    Some(params),
1457                );
1458                let original = serde_json::to_value(&command).unwrap();
1459                let (before, after, dispatch, full_id, dispatch_id) = if subscribe {
1460                    let dispatch = command.with_instrument_ids(second);
1461                    (
1462                        original,
1463                        serde_json::to_value(&command).unwrap(),
1464                        serde_json::to_value(&dispatch).unwrap(),
1465                        command.single_instrument_id(),
1466                        dispatch.single_instrument_id(),
1467                    )
1468                } else {
1469                    let command = SubscribeCommand::$variant(command)
1470                        .into_unsubscribe(command_id, UnixNanos::from(37), Some(correlation_id));
1471
1472                    let UnsubscribeCommand::$variant(command) = command else {
1473                        unreachable!()
1474                    };
1475
1476                    let before = serde_json::to_value(&command).unwrap();
1477                    let dispatch = command.with_instrument_ids(second);
1478                    (
1479                        before,
1480                        serde_json::to_value(&command).unwrap(),
1481                        serde_json::to_value(&dispatch).unwrap(),
1482                        command.single_instrument_id(),
1483                        dispatch.single_instrument_id(),
1484                    )
1485                };
1486
1487                let mut expected = before.clone();
1488                expected["instrument_ids"] = serde_json::to_value([second]).unwrap();
1489                assert_eq!(after, before);
1490                assert_eq!(dispatch, expected);
1491                assert_eq!(
1492                    full_id.unwrap_err().to_string(),
1493                    "subscription requires one instrument ID on this adapter path, received 2"
1494                );
1495                assert_eq!(dispatch_id.unwrap(), second);
1496            }
1497
1498            #[rstest]
1499            fn $release_test(
1500                #[values(
1501                    "client_id",
1502                    "venue",
1503                    "params",
1504                    "command_id",
1505                    "ts_init",
1506                    "correlation_id"
1507                )]
1508                field: &str,
1509            ) {
1510                let first = InstrumentId::from("ESM4.GLBX");
1511                let second = InstrumentId::from("NQM4.GLBX");
1512                let third = InstrumentId::from("YMM4.GLBX");
1513                let params: Params =
1514                    serde_json::from_value(serde_json::json!({"route": 59})).unwrap();
1515
1516                let first_command = $unsubscribe::new(
1517                    first,
1518                    Some(ClientId::from("FIRST")),
1519                    Some(first.venue),
1520                    UUID4::new(),
1521                    UnixNanos::from(41),
1522                    Some(UUID4::new()),
1523                    Some(params),
1524                );
1525
1526                let second_command = first_command.with_instrument_ids(third);
1527
1528                let other_route = match field {
1529                    "client_id" => $unsubscribe {
1530                        client_id: Some(ClientId::from("SECOND")),
1531                        ..second_command
1532                    },
1533                    "venue" => $unsubscribe {
1534                        venue: Some(Venue::from("XCBO")),
1535                        ..second_command
1536                    },
1537                    "params" => $unsubscribe {
1538                        params: Some(
1539                            serde_json::from_value(serde_json::json!({"route": 97})).unwrap(),
1540                        ),
1541                        ..second_command
1542                    },
1543                    "command_id" => $unsubscribe {
1544                        command_id: UUID4::new(),
1545                        ..second_command
1546                    },
1547                    "ts_init" => $unsubscribe {
1548                        ts_init: UnixNanos::from(43),
1549                        ..second_command
1550                    },
1551                    "correlation_id" => $unsubscribe {
1552                        correlation_id: Some(UUID4::new()),
1553                        ..second_command
1554                    },
1555                    _ => unreachable!(),
1556                };
1557
1558                let commands = vec![
1559                    UnsubscribeCommand::$variant(first_command.clone()),
1560                    UnsubscribeCommand::$variant(other_route.clone()),
1561                    UnsubscribeCommand::$variant(
1562                        first_command.with_instrument_ids(second),
1563                    ),
1564                ];
1565                let before = serde_json::to_value(&commands).unwrap();
1566                let grouped = UnsubscribeCommand::group(commands.clone());
1567                let expected = vec![
1568                    UnsubscribeCommand::$variant(
1569                        first_command.with_instrument_ids([first, second]),
1570                    ),
1571                    UnsubscribeCommand::$variant(other_route),
1572                ];
1573                assert_eq!(serde_json::to_value(commands).unwrap(), before);
1574                assert_eq!(
1575                    serde_json::to_value(grouped).unwrap(),
1576                    serde_json::to_value(expected).unwrap()
1577                );
1578            }
1579
1580            #[rstest]
1581            fn $serde_test() {
1582                let first = InstrumentId::from("ESM4.GLBX");
1583                let second = InstrumentId::from("NQM4.GLBX");
1584
1585                let command = $subscribe::new(
1586                    [second, first, second],
1587                    None,
1588                    Some(first.venue),
1589                    UUID4::new(),
1590                    UnixNanos::from(53),
1591                    Some(UUID4::new()),
1592                    None,
1593                );
1594                let mut serialized = serde_json::to_value(&command).unwrap();
1595                serialized["instrument_ids"] =
1596                    serde_json::to_value([second, first, second]).unwrap();
1597                let subscribe: $subscribe =
1598                    serde_json::from_value(serialized.clone()).unwrap();
1599                let unsubscribe: $unsubscribe =
1600                    serde_json::from_value(serialized.clone()).unwrap();
1601                assert_eq!(
1602                    serde_json::to_value(subscribe).unwrap(),
1603                    serde_json::to_value(&command).unwrap()
1604                );
1605                assert_eq!(
1606                    serde_json::to_value(unsubscribe).unwrap(),
1607                    serde_json::to_value(&command).unwrap()
1608                );
1609                serialized["instrument_ids"] = serde_json::json!([]);
1610                assert!(
1611                    serde_json::from_value::<$subscribe>(serialized.clone())
1612                        .unwrap_err()
1613                        .to_string()
1614                        .contains("instrument IDs cannot be empty")
1615                );
1616                assert!(
1617                    serde_json::from_value::<$unsubscribe>(serialized)
1618                        .unwrap_err()
1619                        .to_string()
1620                        .contains("instrument IDs cannot be empty")
1621                );
1622            }
1623        };
1624    }
1625
1626    identifier_message_tests!(
1627        quote_dispatch_preserves_original_intent,
1628        quote_release_grouping_preserves_inputs_and_retained_routes,
1629        quote_message_deserialization_normalizes_lists_and_rejects_empty,
1630        SubscribeQuotes,
1631        UnsubscribeQuotes,
1632        Quotes
1633    );
1634    identifier_message_tests!(
1635        trade_dispatch_preserves_original_intent,
1636        trade_release_grouping_preserves_inputs_and_retained_routes,
1637        trade_message_deserialization_normalizes_lists_and_rejects_empty,
1638        SubscribeTrades,
1639        UnsubscribeTrades,
1640        Trades
1641    );
1642    identifier_message_tests!(
1643        instrument_dispatch_preserves_original_intent,
1644        instrument_release_grouping_preserves_inputs_and_retained_routes,
1645        instrument_message_deserialization_normalizes_lists_and_rejects_empty,
1646        SubscribeInstrument,
1647        UnsubscribeInstrument,
1648        Instrument
1649    );
1650    identifier_message_tests!(
1651        mark_prices_dispatch_preserves_original_intent,
1652        mark_prices_release_grouping_preserves_inputs_and_retained_routes,
1653        mark_prices_message_deserialization_normalizes_lists_and_rejects_empty,
1654        SubscribeMarkPrices,
1655        UnsubscribeMarkPrices,
1656        MarkPrices
1657    );
1658    identifier_message_tests!(
1659        index_prices_dispatch_preserves_original_intent,
1660        index_prices_release_grouping_preserves_inputs_and_retained_routes,
1661        index_prices_message_deserialization_normalizes_lists_and_rejects_empty,
1662        SubscribeIndexPrices,
1663        UnsubscribeIndexPrices,
1664        IndexPrices
1665    );
1666    identifier_message_tests!(
1667        funding_rates_dispatch_preserves_original_intent,
1668        funding_rates_release_grouping_preserves_inputs_and_retained_routes,
1669        funding_rates_message_deserialization_normalizes_lists_and_rejects_empty,
1670        SubscribeFundingRates,
1671        UnsubscribeFundingRates,
1672        FundingRates
1673    );
1674    identifier_message_tests!(
1675        instrument_status_dispatch_preserves_original_intent,
1676        instrument_status_release_grouping_preserves_inputs_and_retained_routes,
1677        instrument_status_message_deserialization_normalizes_lists_and_rejects_empty,
1678        SubscribeInstrumentStatus,
1679        UnsubscribeInstrumentStatus,
1680        InstrumentStatus
1681    );
1682    identifier_message_tests!(
1683        instrument_close_dispatch_preserves_original_intent,
1684        instrument_close_release_grouping_preserves_inputs_and_retained_routes,
1685        instrument_close_message_deserialization_normalizes_lists_and_rejects_empty,
1686        SubscribeInstrumentClose,
1687        UnsubscribeInstrumentClose,
1688        InstrumentClose
1689    );
1690    identifier_message_tests!(
1691        option_greeks_dispatch_preserves_original_intent,
1692        option_greeks_release_grouping_preserves_inputs_and_retained_routes,
1693        option_greeks_message_deserialization_normalizes_lists_and_rejects_empty,
1694        SubscribeOptionGreeks,
1695        UnsubscribeOptionGreeks,
1696        OptionGreeks
1697    );
1698
1699    #[rstest]
1700    #[case("BookDeltas")]
1701    #[case("BookDepth")]
1702    #[case("BookSnapshots")]
1703    fn test_book_identifier_lists_preserve_settings(#[case] family: &str) {
1704        let first = InstrumentId::from("ESM4.GLBX");
1705        let second = InstrumentId::from("NQM4.GLBX");
1706        let mut fields = serde_json::json!({
1707            "instrument_ids": [first, second, first], "book_type": "L2_MBP", "depth": 7,
1708            "client_id": "BOOKS", "venue": "GLBX", "command_id": UUID4::new(),
1709            "ts_init": 43, "correlation_id": UUID4::new(), "params": {"route": 59}
1710        });
1711
1712        if family == "BookSnapshots" {
1713            fields["interval_ms"] = serde_json::json!(13);
1714        } else {
1715            fields["managed"] = serde_json::json!(true);
1716        }
1717
1718        let command: SubscribeCommand =
1719            serde_json::from_value(serde_json::json!({family: fields})).unwrap();
1720        fields["instrument_ids"] = serde_json::json!([first, second]);
1721        let before = serde_json::to_value(&command).unwrap();
1722        let projected = command.with_instrument_ids(second).unwrap();
1723        let mut expected = fields.clone();
1724        expected["instrument_ids"] = serde_json::json!([second]);
1725        let inverse = command.clone().into_unsubscribe(
1726            command.command_id(),
1727            command.ts_init(),
1728            command.correlation_id(),
1729        );
1730        let mut expected_inverse = fields.clone();
1731        for field in ["book_type", "depth", "managed"] {
1732            expected_inverse.as_object_mut().unwrap().remove(field);
1733        }
1734
1735        let mut empty = fields.clone();
1736        empty["instrument_ids"] = serde_json::json!([]);
1737        let error = serde_json::from_value::<SubscribeCommand>(serde_json::json!({family: empty}))
1738            .unwrap_err();
1739        let releases = [first, second].map(|id| inverse.with_instrument_ids(id).unwrap());
1740        let grouped = UnsubscribeCommand::group(releases.to_vec());
1741        assert_eq!(command.instrument_ids(), Some([first, second].as_slice()));
1742        assert_eq!(before, serde_json::json!({family: fields}));
1743        assert_eq!(serde_json::to_value(&command).unwrap(), before);
1744        assert_eq!(
1745            serde_json::to_value(projected).unwrap(),
1746            serde_json::json!({family: expected})
1747        );
1748        assert_eq!(
1749            serde_json::to_value(&inverse).unwrap(),
1750            serde_json::json!({family: expected_inverse})
1751        );
1752        assert_eq!(
1753            serde_json::to_value(grouped).unwrap(),
1754            serde_json::json!([{family: expected_inverse}])
1755        );
1756        assert!(error.to_string().contains("instrument IDs cannot be empty"));
1757    }
1758
1759    #[rstest::rstest]
1760    #[case::subscribe(true)]
1761    #[case::unsubscribe(false)]
1762    fn bar_lists_serialize_normalize_and_project_metadata(#[case] subscribe: bool) {
1763        let first = BarType::from("AUD/USD.SIM-1-MINUTE-LAST-EXTERNAL");
1764        let second = BarType::from("AUD/USD.SIM-5-MINUTE-LAST-EXTERNAL");
1765
1766        let command = SubscribeBars::new(
1767            [second, first, second],
1768            Some(ClientId::from("BARS")),
1769            Some(Venue::from("SIM")),
1770            UUID4::new(),
1771            UnixNanos::from(37),
1772            Some(UUID4::new()),
1773            Some(serde_json::from_value(serde_json::json!({"route": 59})).unwrap()),
1774        );
1775
1776        let release = match SubscribeCommand::Bars(command.clone()).into_unsubscribe(
1777            UUID4::new(),
1778            UnixNanos::from(41),
1779            Some(UUID4::new()),
1780        ) {
1781            UnsubscribeCommand::Bars(command) => command,
1782            _ => unreachable!(),
1783        };
1784
1785        let before = if subscribe {
1786            serde_json::to_value(&command).unwrap()
1787        } else {
1788            serde_json::to_value(&release).unwrap()
1789        };
1790
1791        let projected = if subscribe {
1792            serde_json::to_value(command.with_bar_types(first)).unwrap()
1793        } else {
1794            serde_json::to_value(release.with_bar_types(first)).unwrap()
1795        };
1796
1797        let mut expected = before.clone();
1798        expected["bar_types"] = serde_json::to_value([first]).unwrap();
1799        assert_eq!(projected, expected);
1800        assert_eq!(
1801            before["bar_types"],
1802            serde_json::to_value([second, first]).unwrap()
1803        );
1804        assert!(before.get("bar_type").is_none());
1805        assert_eq!(
1806            command.single_bar_type().unwrap_err().to_string(),
1807            "subscription requires one bar type on this adapter path, received 2"
1808        );
1809        assert_eq!(
1810            release.with_bar_types(first).single_bar_type().unwrap(),
1811            first
1812        );
1813        let mut repeated = before.clone();
1814        repeated["bar_types"] = serde_json::to_value([second, first, second]).unwrap();
1815        let normalized = if subscribe {
1816            serde_json::to_value(serde_json::from_value::<SubscribeBars>(repeated).unwrap())
1817                .unwrap()
1818        } else {
1819            serde_json::to_value(serde_json::from_value::<UnsubscribeBars>(repeated).unwrap())
1820                .unwrap()
1821        };
1822
1823        assert_eq!(normalized, before);
1824        let mut empty = before;
1825        empty["bar_types"] = serde_json::json!([]);
1826
1827        if subscribe {
1828            assert!(
1829                serde_json::from_value::<SubscribeBars>(empty)
1830                    .unwrap_err()
1831                    .to_string()
1832                    .contains("bar types cannot be empty")
1833            );
1834        } else {
1835            assert!(
1836                serde_json::from_value::<UnsubscribeBars>(empty)
1837                    .unwrap_err()
1838                    .to_string()
1839                    .contains("bar types cannot be empty")
1840            );
1841        }
1842    }
1843
1844    #[rstest::rstest]
1845    fn bar_release_grouping_preserves_metadata_boundaries(
1846        #[values(
1847            "client_id",
1848            "venue",
1849            "params",
1850            "command_id",
1851            "ts_init",
1852            "correlation_id"
1853        )]
1854        field: &str,
1855    ) {
1856        let first = BarType::from("AUD/USD.SIM-1-MINUTE-LAST-EXTERNAL");
1857        let second = BarType::from("AUD/USD.SIM-5-MINUTE-LAST-EXTERNAL");
1858
1859        let command = UnsubscribeBars::new(
1860            first,
1861            Some(ClientId::from("FIRST")),
1862            Some(Venue::from("SIM")),
1863            UUID4::new(),
1864            UnixNanos::from(41),
1865            Some(UUID4::new()),
1866            Some(serde_json::from_value(serde_json::json!({"route": 59})).unwrap()),
1867        );
1868        let same = command.with_bar_types([second, first]);
1869        let mut other = serde_json::to_value(command.with_bar_types(second)).unwrap();
1870
1871        other[field] = match field {
1872            "client_id" => serde_json::to_value(ClientId::from("SECOND")).unwrap(),
1873            "venue" => serde_json::to_value(Venue::from("OTHER")).unwrap(),
1874            "params" => serde_json::json!({"route": 97}),
1875            "command_id" | "correlation_id" => serde_json::to_value(UUID4::new()).unwrap(),
1876            "ts_init" => serde_json::to_value(UnixNanos::from(43)).unwrap(),
1877            _ => unreachable!(),
1878        };
1879
1880        let other: UnsubscribeBars = serde_json::from_value(other).unwrap();
1881        let expected = vec![
1882            UnsubscribeCommand::Bars(command.with_bar_types([first, second])),
1883            UnsubscribeCommand::Bars(other.clone()),
1884        ];
1885        let grouped = UnsubscribeCommand::group(vec![
1886            UnsubscribeCommand::Bars(command),
1887            UnsubscribeCommand::Bars(same),
1888            UnsubscribeCommand::Bars(other),
1889        ]);
1890        assert_eq!(
1891            serde_json::to_value(grouped).unwrap(),
1892            serde_json::to_value(expected).unwrap()
1893        );
1894    }
1895}