1use async_trait::async_trait;
8use bytes::Bytes;
9use futu_command_runtime::{
10 CommandAuthorizationEvidence, CommandRequest, CommandResponse, CommandTransport,
11};
12use futu_command_spec::trade::{TradeBackendChannel, trade_account_discovery_command};
13use futu_command_spec::{
14 AuthRequirement, BackendChannelKind, BrokerDiscoveryOperation, CommandSpecId,
15 ConnectionDiscoveryOperation, HeartbeatOperation, backend_extension_command,
16 broker_discovery_command, command_spec_by_id, connection_discovery_command, heartbeat_command,
17 indicator_command, market_event_command, qot_read_command, static_data_read_command,
18 system_read_command, user_cloud_command,
19};
20pub use futu_command_spec::{
21 BackendExtensionOperation, CMD_GET_BIZ_GROUP, CMD_GET_BOND_ANSWER_STATE,
22 CMD_GET_BOND_POSITION_LIST, CMD_GET_BOND_SINGLE_ASSET, CMD_GET_BOND_TOTAL_ASSET,
23 CMD_GET_BOND_TRADE_REMINDER, CMD_GET_CASH_DETAIL, CMD_GET_CASH_LOG, CMD_GET_CN_AH_MARGIN_INFO,
24 CMD_GET_HK_MARGIN_INFO, CMD_GET_US_MARGIN_INFO, CMD_PULL_ACCOUNT_FLAG,
25 IndicatorOperation as IndicatorCommandOperation, MarketEventOperation, QotReadOperation,
26 QotWriteOperation, StaticDataReadOperation, SystemReadOperation,
27 TradeAccountDiscoveryOperation, TradeAuthOperation, TradeQueryEnvironment, TradeQueryOperation,
28 UserCloudOperation,
29 trade::{CryptoTradeOperation, TradeWriteOperation},
30};
31use futu_core::error::{FutuError, Result as FutuResult};
32
33use crate::conn::{
34 BackendConn, BackendWriterAdmission, RequestTimeoutPolicy, WriterAdmissionObserver,
35};
36use crate::quote_sub::{
37 DispatchAccepted, NormalSubscriptionDispatchHooks, QotSubscriptionWriterAdmission,
38};
39
40#[path = "command_runtime_policy.rs"]
41mod policy;
42
43use policy::{BACKEND_COMMAND_TIMEOUT, BackendCommandPolicyPorts};
44
45#[cfg(test)]
46#[path = "command_runtime_tests.rs"]
47mod tests;
48
49#[cfg(test)]
50#[path = "command_runtime_crypto_tests.rs"]
51mod crypto_tests;
52
53#[cfg(test)]
54#[path = "command_runtime_ordinary_tests.rs"]
55mod ordinary_tests;
56
57#[cfg(test)]
58pub(crate) use crate::command_runtime_test_support::{
59 decode_internal_test_request, decode_internal_test_request_body, encode_internal_test_response,
60 execute_internal_test_command,
61};
62
63pub use futu_command_runtime::{
64 ChannelId, ChannelLifecycleState, ChannelRuntimeAction, CommandExecution,
65 CommandExecutionContext, CommandExecutionOutcome, CommandOutcome, CommandRuntime,
66 CommandRuntimeAction, CommandRuntimeDecision, CommandRuntimeError, decide_command_outcome,
67};
68
69#[derive(Clone)]
70pub(crate) struct SubscriptionDispatchContext {
71 pub(crate) quote_market_type: u8,
72 pub(crate) exact_normal_wire: Vec<u8>,
73 pub(crate) hooks: NormalSubscriptionDispatchHooks,
74 pub(crate) accepted: std::sync::Arc<parking_lot::Mutex<Option<DispatchAccepted>>>,
75}
76
77pub struct ChannelBoundBackendTransport<'a> {
78 backend: &'a BackendConn,
79 channel: BackendChannelKind,
80 request_timeout: std::time::Duration,
81 timeout_policy: RequestTimeoutPolicy,
82 subscription_dispatch: Option<SubscriptionDispatchContext>,
83}
84
85impl<'a> ChannelBoundBackendTransport<'a> {
86 #[must_use]
87 pub fn new(backend: &'a BackendConn, channel: BackendChannelKind) -> Self {
88 Self {
89 backend,
90 channel,
91 request_timeout: BACKEND_COMMAND_TIMEOUT,
92 timeout_policy: RequestTimeoutPolicy::Disconnect,
93 subscription_dispatch: None,
94 }
95 }
96
97 fn keep_connection_on_timeout(mut self) -> Self {
98 self.timeout_policy = RequestTimeoutPolicy::KeepConnection;
99 self
100 }
101
102 fn with_subscription_dispatch(mut self, context: SubscriptionDispatchContext) -> Self {
103 self.subscription_dispatch = Some(context);
104 self
105 }
106}
107
108#[async_trait]
109impl CommandTransport for ChannelBoundBackendTransport<'_> {
110 fn channel_kind(&self) -> Option<BackendChannelKind> {
111 Some(self.channel)
112 }
113
114 async fn execute(&self, request: CommandRequest) -> FutuResult<CommandResponse> {
115 let context = self.subscription_dispatch.as_ref();
116 let observer = match request.spec_id {
117 CommandSpecId::QotSubscriptionSet => context.map(|context| {
118 let quote_market_type = context.quote_market_type;
119 let exact_normal_wire = context.exact_normal_wire.clone();
120 let hooks = context.hooks.clone();
121 let admission_hooks = hooks.clone();
122 let accepted = std::sync::Arc::clone(&context.accepted);
123 WriterAdmissionObserver {
124 try_admit: std::sync::Arc::new(move |admission: BackendWriterAdmission| {
125 (admission_hooks.try_admit)(QotSubscriptionWriterAdmission {
126 connection_generation: admission.connection_generation,
127 serial_no: admission.serial_no,
128 })
129 }),
130 on_accepted: std::sync::Arc::new(move |admission: BackendWriterAdmission| {
131 let receipt = (hooks.on_accepted)(
132 QotSubscriptionWriterAdmission {
133 connection_generation: admission.connection_generation,
134 serial_no: admission.serial_no,
135 },
136 quote_market_type,
137 exact_normal_wire.clone(),
138 );
139 *accepted.lock() = Some(receipt);
140 }),
141 }
142 }),
143 _ if context.is_some() => {
144 return Err(FutuError::Codec(
145 "subscription dispatch context used by non-subscription command".into(),
146 ));
147 }
148 _ => None,
149 };
150 execute_on_backend(
151 self.backend,
152 request,
153 self.request_timeout,
154 self.timeout_policy,
155 observer,
156 )
157 .await
158 }
159}
160
161fn runtime_with_transport(
162 transport: ChannelBoundBackendTransport<'_>,
163 authorization: CommandAuthorizationEvidence,
164) -> CommandRuntime<ChannelBoundBackendTransport<'_>, BackendCommandPolicyPorts<'_>> {
165 let policy = BackendCommandPolicyPorts::new(transport.backend, authorization);
166 CommandRuntime::with_policy(transport, policy)
167}
168
169fn login_runtime_with_transport(
170 transport: ChannelBoundBackendTransport<'_>,
171) -> CommandRuntime<ChannelBoundBackendTransport<'_>, BackendCommandPolicyPorts<'_>> {
172 runtime_with_transport(transport, CommandAuthorizationEvidence::LoginSession)
173}
174
175fn authorization_for_command(
176 spec_id: CommandSpecId,
177 trade_cipher: Option<&[u8]>,
178) -> FutuResult<CommandAuthorizationEvidence> {
179 let spec = command_spec_by_id(spec_id)
180 .ok_or_else(|| FutuError::Codec(format!("unknown command spec id: {spec_id:?}")))?;
181 match spec.auth {
182 AuthRequirement::TradeUnlocked => trade_cipher
183 .ok_or_else(|| {
184 FutuError::Codec(
185 "command authorization evidence: trade cipher was not provided".to_owned(),
186 )
187 })
188 .and_then(|cipher| {
189 CommandAuthorizationEvidence::from_trade_cipher(cipher).map_err(|error| {
190 FutuError::Codec(format!("command authorization evidence: {error}"))
191 })
192 }),
193 AuthRequirement::None | AuthRequirement::Login => {
194 Ok(CommandAuthorizationEvidence::LoginSession)
195 }
196 }
197}
198
199pub async fn execute_trade_read(
200 backend: &BackendConn,
201 operation: TradeQueryOperation,
202 environment: TradeQueryEnvironment,
203 trade_cipher: Option<&[u8]>,
204 body: Bytes,
205) -> FutuResult<CommandResponse> {
206 let channel = match environment {
207 TradeQueryEnvironment::Real => BackendChannelKind::Broker,
208 TradeQueryEnvironment::Sim => BackendChannelKind::Platform,
209 };
210 let spec_id = CommandSpecId::TradeRead {
211 operation,
212 environment,
213 };
214 let runtime = runtime_with_transport(
215 ChannelBoundBackendTransport::new(backend, channel),
216 authorization_for_command(spec_id, trade_cipher)?,
217 );
218 let execution = runtime
219 .execute(CommandExecutionContext::new(spec_id, body))
220 .await
221 .map_err(|error| FutuError::Codec(format!("trade-read command spec error: {error}")))?;
222 let report = &execution.report;
223 tracing::trace!(
224 command = report.command_name,
225 cmd_id = report.cmd_id,
226 channel = ?report.channel,
227 outcome = ?report.outcome,
228 response_body_len = report.response_body_len,
229 "command runtime executed trade read"
230 );
231 execution.into_response()
232}
233
234pub async fn execute_backend_extension(
235 backend: &BackendConn,
236 operation: BackendExtensionOperation,
237 body: Bytes,
238) -> FutuResult<CommandResponse> {
239 let spec = backend_extension_command(operation);
240 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
241 backend,
242 spec.runtime.channel,
243 ));
244 let execution = runtime
245 .execute(CommandExecutionContext::new(
246 CommandSpecId::BackendExtension(operation),
247 body,
248 ))
249 .await
250 .map_err(|error| {
251 FutuError::Codec(format!("backend-extension command spec error: {error}"))
252 })?;
253 let report = &execution.report;
254 tracing::trace!(
255 command = report.command_name,
256 cmd_id = report.cmd_id,
257 channel = ?report.channel,
258 evidence_kind = ?report.command_evidence.kind,
259 outcome = ?report.outcome,
260 response_body_len = report.response_body_len,
261 "command runtime executed backend extension"
262 );
263 execution.into_response()
264}
265
266pub async fn execute_trade_auth(
267 backend: &BackendConn,
268 operation: TradeAuthOperation,
269 body: Bytes,
270) -> FutuResult<CommandResponse> {
271 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
272 backend,
273 BackendChannelKind::Broker,
274 ));
275 let execution = runtime
276 .execute(CommandExecutionContext::new(
277 CommandSpecId::TradeAuth(operation),
278 body,
279 ))
280 .await
281 .map_err(|error| FutuError::Codec(format!("trade-auth command spec error: {error}")))?;
282 let report = &execution.report;
283 tracing::trace!(
284 command = report.command_name,
285 cmd_id = report.cmd_id,
286 channel = ?report.channel,
287 outcome = ?report.outcome,
288 response_body_len = report.response_body_len,
289 "command runtime executed trade auth"
290 );
291 execution.into_response()
292}
293
294pub async fn execute_trade_account_discovery(
295 backend: &BackendConn,
296 operation: TradeAccountDiscoveryOperation,
297 body: Bytes,
298) -> FutuResult<CommandResponse> {
299 let channel = match trade_account_discovery_command(operation).channel {
300 TradeBackendChannel::Broker => BackendChannelKind::Broker,
301 TradeBackendChannel::Platform => BackendChannelKind::Platform,
302 };
303 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(backend, channel));
304 let execution = runtime
305 .execute(CommandExecutionContext::new(
306 CommandSpecId::TradeAccountDiscovery(operation),
307 body,
308 ))
309 .await
310 .map_err(|error| {
311 FutuError::Codec(format!(
312 "trade-account-discovery command spec error: {error}"
313 ))
314 })?;
315 let report = &execution.report;
316 tracing::trace!(
317 command = report.command_name,
318 cmd_id = report.cmd_id,
319 channel = ?report.channel,
320 outcome = ?report.outcome,
321 response_body_len = report.response_body_len,
322 "command runtime executed trade account discovery"
323 );
324 execution.into_response()
325}
326
327pub async fn execute_connection_discovery(
328 backend: &BackendConn,
329 operation: ConnectionDiscoveryOperation,
330 body: Bytes,
331) -> FutuResult<CommandResponse> {
332 let spec = connection_discovery_command(operation);
333 let runtime =
334 login_runtime_with_transport(ChannelBoundBackendTransport::new(backend, spec.channel));
335 let execution = runtime
336 .execute(CommandExecutionContext::new(
337 CommandSpecId::ConnectionDiscovery(operation),
338 body,
339 ))
340 .await
341 .map_err(|error| {
342 FutuError::Codec(format!("connection-discovery command spec error: {error}"))
343 })?;
344 let report = &execution.report;
345 tracing::trace!(
346 command = report.command_name,
347 cmd_id = report.cmd_id,
348 channel = ?report.channel,
349 outcome = ?report.outcome,
350 response_body_len = report.response_body_len,
351 "command runtime executed connection discovery"
352 );
353 execution.into_response()
354}
355
356pub async fn execute_broker_discovery(
357 backend: &BackendConn,
358 operation: BrokerDiscoveryOperation,
359 body: Bytes,
360) -> FutuResult<CommandResponse> {
361 if operation == BrokerDiscoveryOperation::ValidBrokerListChangedPush {
362 return Err(FutuError::Codec(
363 "CMD20177 is push-only and cannot be executed as a request".into(),
364 ));
365 }
366 let spec = broker_discovery_command(operation);
367 let mut transport = ChannelBoundBackendTransport::new(backend, spec.runtime.channel);
368 if operation == BrokerDiscoveryOperation::ValidBrokerList {
369 transport = transport.keep_connection_on_timeout();
370 }
371 let runtime = login_runtime_with_transport(transport);
372 let execution = runtime
373 .execute(CommandExecutionContext::new(
374 CommandSpecId::BrokerDiscovery(operation),
375 body,
376 ))
377 .await
378 .map_err(|error| {
379 FutuError::Codec(format!("broker-discovery command spec error: {error}"))
380 })?;
381 let report = &execution.report;
382 tracing::trace!(
383 command = report.command_name,
384 cmd_id = report.cmd_id,
385 channel = ?report.channel,
386 outcome = ?report.outcome,
387 response_body_len = report.response_body_len,
388 "command runtime executed broker discovery"
389 );
390 execution.into_response()
391}
392
393pub async fn execute_heartbeat(
394 backend: &BackendConn,
395 operation: HeartbeatOperation,
396 body: Bytes,
397) -> FutuResult<CommandResponse> {
398 let spec = heartbeat_command(operation);
399 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
400 backend,
401 spec.runtime.channel,
402 ));
403 let execution = runtime
404 .execute(CommandExecutionContext::new(
405 CommandSpecId::Heartbeat(operation),
406 body,
407 ))
408 .await
409 .map_err(|error| FutuError::Codec(format!("heartbeat command spec error: {error}")))?;
410 let report = &execution.report;
411 tracing::trace!(
412 command = report.command_name,
413 cmd_id = report.cmd_id,
414 channel = ?report.channel,
415 outcome = ?report.outcome,
416 response_body_len = report.response_body_len,
417 "command runtime executed heartbeat"
418 );
419 execution.into_response()
420}
421
422pub async fn execute_trade_write(
423 backend: &BackendConn,
424 operation: TradeWriteOperation,
425 environment: TradeQueryEnvironment,
426 trade_cipher: Option<&[u8]>,
427 body: Bytes,
428) -> FutuResult<CommandResponse> {
429 let channel = match environment {
430 TradeQueryEnvironment::Real => BackendChannelKind::Broker,
431 TradeQueryEnvironment::Sim => BackendChannelKind::Platform,
432 };
433 let spec_id = CommandSpecId::TradeWrite {
434 operation,
435 environment,
436 };
437 let runtime = runtime_with_transport(
438 ChannelBoundBackendTransport::new(backend, channel),
439 authorization_for_command(spec_id, trade_cipher)?,
440 );
441 let execution = runtime
442 .execute(CommandExecutionContext::new(spec_id, body))
443 .await
444 .map_err(|error| FutuError::Codec(format!("trade-write command spec error: {error}")))?;
445 let report = &execution.report;
446 tracing::trace!(
447 command = report.command_name,
448 cmd_id = report.cmd_id,
449 channel = ?report.channel,
450 outcome = ?report.outcome,
451 response_body_len = report.response_body_len,
452 "command runtime executed trade write"
453 );
454 execution.into_response()
455}
456
457pub async fn execute_crypto_trade_command(
458 backend: &BackendConn,
459 operation: CryptoTradeOperation,
460 trade_cipher: Option<&[u8]>,
461 body: Bytes,
462) -> FutuResult<CommandResponse> {
463 let spec_id = CommandSpecId::CryptoTrade(operation);
464 let runtime = runtime_with_transport(
465 ChannelBoundBackendTransport::new(backend, BackendChannelKind::Broker),
466 authorization_for_command(spec_id, trade_cipher)?,
467 );
468 let execution = runtime
469 .execute(CommandExecutionContext::new(spec_id, body))
470 .await
471 .map_err(|error| FutuError::Codec(format!("crypto-trade command spec error: {error}")))?;
472 let report = &execution.report;
473 tracing::trace!(
474 command = report.command_name,
475 cmd_id = report.cmd_id,
476 channel = ?report.channel,
477 outcome = ?report.outcome,
478 response_body_len = report.response_body_len,
479 "command runtime executed crypto trade command"
480 );
481 execution.into_response()
482}
483
484pub async fn execute_system_read(
485 backend: &BackendConn,
486 operation: SystemReadOperation,
487 body: Bytes,
488) -> FutuResult<CommandResponse> {
489 let spec = system_read_command(operation);
490 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
491 backend,
492 spec.runtime.channel,
493 ));
494 let execution = runtime
495 .execute(CommandExecutionContext::new(
496 CommandSpecId::SystemRead(operation),
497 body,
498 ))
499 .await
500 .map_err(|error| FutuError::Codec(format!("system-read command spec error: {error}")))?;
501 let report = &execution.report;
502 tracing::trace!(
503 command = report.command_name,
504 cmd_id = report.cmd_id,
505 channel = ?report.channel,
506 evidence_kind = ?report.command_evidence.kind,
507 outcome = ?report.outcome,
508 response_body_len = report.response_body_len,
509 "command runtime executed system read"
510 );
511 execution.into_response()
512}
513
514pub async fn execute_indicator_read(
515 backend: &BackendConn,
516 operation: IndicatorCommandOperation,
517 body: Bytes,
518) -> FutuResult<CommandResponse> {
519 let spec = indicator_command(operation);
520 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
521 backend,
522 spec.runtime.channel,
523 ));
524 let execution = runtime
525 .execute(CommandExecutionContext::new(
526 CommandSpecId::Indicator(operation),
527 body,
528 ))
529 .await
530 .map_err(|error| FutuError::Codec(format!("indicator command spec error: {error}")))?;
531 execution.into_response()
532}
533
534pub async fn execute_static_data_read(
535 backend: &BackendConn,
536 operation: StaticDataReadOperation,
537 body: Bytes,
538) -> FutuResult<CommandResponse> {
539 let spec = static_data_read_command(operation);
540 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
541 backend,
542 spec.runtime.channel,
543 ));
544 let execution = runtime
545 .execute(CommandExecutionContext::new(
546 CommandSpecId::StaticDataRead(operation),
547 body,
548 ))
549 .await
550 .map_err(|error| {
551 FutuError::Codec(format!("static-data-read command spec error: {error}"))
552 })?;
553 let report = &execution.report;
554 tracing::trace!(
555 command = report.command_name,
556 cmd_id = report.cmd_id,
557 channel = ?report.channel,
558 outcome = ?report.outcome,
559 response_body_len = report.response_body_len,
560 "command runtime executed static-data read"
561 );
562 execution.into_response()
563}
564
565pub async fn execute_user_cloud(
566 backend: &BackendConn,
567 operation: UserCloudOperation,
568 body: Bytes,
569) -> FutuResult<CommandResponse> {
570 if operation == UserCloudOperation::UpdatePush {
571 return Err(FutuError::Codec(
572 "user-cloud CMD20175 is push-only and cannot be executed as a request".to_string(),
573 ));
574 }
575 let spec = user_cloud_command(operation);
576 if spec.runtime.channel != BackendChannelKind::Platform {
577 return Err(FutuError::Codec(format!(
578 "user-cloud command {} is bound to {:?}",
579 spec.runtime.name, spec.runtime.channel
580 )));
581 }
582 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
583 backend,
584 spec.runtime.channel,
585 ));
586 let execution = runtime
587 .execute(CommandExecutionContext::new(
588 CommandSpecId::UserCloud(operation),
589 body,
590 ))
591 .await
592 .map_err(|error| FutuError::Codec(format!("user-cloud command spec error: {error}")))?;
593 let report = &execution.report;
594 tracing::trace!(
595 command = report.command_name,
596 cmd_id = report.cmd_id,
597 channel = ?report.channel,
598 outcome = ?report.outcome,
599 response_body_len = report.response_body_len,
600 "command runtime executed user cloud command"
601 );
602 execution.into_response()
603}
604
605pub async fn execute_market_event(
606 backend: &BackendConn,
607 operation: MarketEventOperation,
608 body: Bytes,
609 reserved: [u8; 10],
610) -> FutuResult<CommandResponse> {
611 if operation == MarketEventOperation::Push {
612 return Err(FutuError::Codec(
613 "market-event CMD6301 is push-only and cannot be executed as a request".to_string(),
614 ));
615 }
616 let spec = market_event_command(operation);
617 if spec.runtime.channel != BackendChannelKind::Qot {
618 return Err(FutuError::Codec(format!(
619 "market-event command {} is bound to {:?}",
620 spec.runtime.name, spec.runtime.channel
621 )));
622 }
623 execute_qot_command(
624 backend,
625 CommandSpecId::MarketEvent(operation),
626 body,
627 reserved,
628 )
629 .await
630}
631
632pub async fn execute_qot_plaintext(
633 backend: &BackendConn,
634 cmd_id: u16,
635 body: Bytes,
636 reserved: [u8; 10],
637) -> FutuResult<CommandResponse> {
638 execute_qot_command(backend, CommandSpecId::QotPlaintext(cmd_id), body, reserved).await
639}
640
641pub async fn execute_qot_read(
642 backend: &BackendConn,
643 operation: QotReadOperation,
644 body: Bytes,
645) -> FutuResult<CommandResponse> {
646 execute_qot_read_with_reserved(backend, operation, body, [0_u8; 10]).await
647}
648
649pub async fn execute_qot_read_with_reserved(
650 backend: &BackendConn,
651 operation: QotReadOperation,
652 body: Bytes,
653 reserved: [u8; 10],
654) -> FutuResult<CommandResponse> {
655 let spec = qot_read_command(operation);
656 if spec.runtime.channel != BackendChannelKind::Qot {
657 return Err(FutuError::Codec(format!(
658 "QOT read command {} is bound to {:?}",
659 spec.runtime.name, spec.runtime.channel
660 )));
661 }
662 execute_qot_command(backend, CommandSpecId::QotRead(operation), body, reserved).await
663}
664
665pub async fn execute_qot_subscription_set(
666 backend: &BackendConn,
667 body: Bytes,
668 reserved: [u8; 10],
669) -> FutuResult<CommandResponse> {
670 execute_qot_command(backend, CommandSpecId::QotSubscriptionSet, body, reserved).await
671}
672
673pub(crate) async fn execute_qot_subscription_set_with_dispatch(
674 backend: &BackendConn,
675 body: Bytes,
676 reserved: [u8; 10],
677 context: SubscriptionDispatchContext,
678) -> FutuResult<CommandResponse> {
679 let runtime = login_runtime_with_transport(
680 ChannelBoundBackendTransport::new(backend, BackendChannelKind::Qot)
681 .with_subscription_dispatch(context),
682 );
683 let execution = runtime
684 .execute(
685 CommandExecutionContext::new(CommandSpecId::QotSubscriptionSet, body)
686 .with_reserved(reserved),
687 )
688 .await
689 .map_err(|error| FutuError::Codec(format!("qot command spec error: {error}")))?;
690 execution.into_response()
691}
692
693pub async fn execute_qot_write(
694 backend: &BackendConn,
695 operation: QotWriteOperation,
696 body: Bytes,
697 reserved: [u8; 10],
698) -> FutuResult<CommandResponse> {
699 execute_qot_command(backend, CommandSpecId::QotWrite(operation), body, reserved).await
700}
701
702async fn execute_qot_command(
703 backend: &BackendConn,
704 spec_id: CommandSpecId,
705 body: Bytes,
706 reserved: [u8; 10],
707) -> FutuResult<CommandResponse> {
708 let runtime = login_runtime_with_transport(ChannelBoundBackendTransport::new(
709 backend,
710 BackendChannelKind::Qot,
711 ));
712 let execution = runtime
713 .execute(CommandExecutionContext::new(spec_id, body).with_reserved(reserved))
714 .await
715 .map_err(|error| FutuError::Codec(format!("qot command spec error: {error}")))?;
716 let report = &execution.report;
717 tracing::trace!(
718 command = report.command_name,
719 cmd_id = report.cmd_id,
720 channel = ?report.channel,
721 outcome = ?report.outcome,
722 response_body_len = report.response_body_len,
723 "command runtime executed QOT command"
724 );
725 execution.into_response()
726}
727
728async fn execute_on_backend(
729 backend: &BackendConn,
730 request: CommandRequest,
731 request_timeout: std::time::Duration,
732 timeout_policy: RequestTimeoutPolicy,
733 observer: Option<WriterAdmissionObserver>,
734) -> FutuResult<CommandResponse> {
735 let frame = backend
736 .request_with_reserved_timeout_policy_observed(
737 request.cmd_id,
738 request.body.to_vec(),
739 request.reserved,
740 request_timeout,
741 timeout_policy,
742 observer,
743 )
744 .await?;
745 Ok(CommandResponse {
746 request_serial_no: frame.header.serial_no,
747 cmd_id: frame.header.cmd_id,
748 body: frame.body,
749 ex_head: frame.ex_head,
750 })
751}