1/*
2 * SPDX-FileCopyrightText: 2025 Copyright (c) Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 */
13
14use std::{
15 sync::Arc,
16 time::{Duration, Instant},
17};
18
19use async_trait::async_trait;
20use cda_interfaces::{
21 DiagServiceError, DoipComParams, EcuAddresses, EcuConnectivityHandler, FunctionalTransport,
22 HashMap, HashMapExtensions, NetworkTopology, PhysicalTransport, RouteStatus, ServicePayload,
23 TransmissionParameters, TransportProbe, TransportResponse, VariantDetectionRequest,
24 VariantDetectionSender,
25 communication_control::{
26 GatewayLifecycle, TransportControl, TransportState, error::CommControlError,
27 },
28 dlt_ctx, pending_nrc_from_raw, uds_response_from_raw,
29 util::{self, tokio_ext},
30};
31use doip_definitions::{
32 header::ProtocolVersion,
33 payload::{
34 DiagnosticMessage, DiagnosticMessageNack, DoipPayload, GenericNack,
35 RoutingActivationRequest,
36 },
37};
38use futures::FutureExt;
39use tokio::{
40 sync::{Mutex, RwLock, broadcast, mpsc},
41 task::JoinHandle,
42};
43use tokio_util::sync::CancellationToken;
44
45pub mod config;
46pub mod error;
47pub use error::{ConnectionError, DoipGatewaySetupError};
48mod connection_receiver;
49mod connection_sender;
50mod connections;
51mod ecu_connection;
52pub mod socket;
53mod vir_vam;
54
55#[cfg(test)]
56pub(crate) mod test_helpers;
57
58use crate::{
59 config::DoipConfig,
60 connections::{EcuError, GatewayState},
61 socket::DoIPUdpSocket,
62};
63
64/// Timeout when suppressPosRspMsgIndicationBit is set.
65/// ECUs that reject a request usually respond quickly with an NRC and
66/// this timeout window should be able to capture those while
67/// not blocking for the full diagnostic timeout.
68///
69/// This constant is intentionally separate from `timeout_default` so it can be promoted to a
70/// configuration entry, or a `ComParam` in a future iteration without changing the
71/// surrounding logic.
72const SUPPRESS_POSITIVE_RESPONSE_TIMEOUT: Duration = Duration::from_millis(500);
73
74/// Events received on a `DoIP` connection.
75///
76/// `Msg` carries the raw UDS payload and its `DoIP` logical addresses. All other
77/// variants are ISO 13400-2 protocol-level events that must be handled before
78/// any UDS data is forwarded to the caller.
79#[derive(Debug, Clone)]
80enum DiagnosticResponse {
81 /// A UDS payload received from an ECU, carried verbatim before classification.
82 Msg {
83 source_address: u16,
84 target_address: u16,
85 data: Vec<u8>,
86 },
87 /// An acknowledgment of a previously sent UDS payload, carrying the original
88 /// request's logical address and data.
89 Ack((u16, Vec<u8>)),
90 /// A negative acknowledgment of a previously sent UDS payload, including
91 /// the original Nack message struct.
92 Nack(DiagnosticMessageNack),
93 /// A response to a `DoIP` alive-check request, indicating that the ECU is reachable.
94 AliveCheckResponse,
95 /// `TesterPresent` NRC - intercepted at the decoding layer so the receiver
96 /// can log-and-drop without routing it to a per-ECU channel.
97 TesterPresentNRC(u8),
98 /// Generic `DoIP` negative acknowledgment, carrying the `GenericNack` struct with the NACK code.
99 GenericNack(GenericNack), // todo #22 -> we need the address of the ECU that sent the nack
100}
101
102impl DiagnosticResponse {
103 fn matches_request(&self, request: &[u8]) -> bool {
104 let Some(sid) = request.first() else {
105 return false;
106 };
107
108 match self {
109 Self::Ack((_, previous)) => request.starts_with(previous),
110 // Positive and negative echoes both belong to the request: a
111 // final NRC can overtake the ACK, and ignoring it here consumes
112 // it from the receiver - the later response read then times out
113 // although the ECU answered.
114 Self::Msg { data, .. } => util::uds_response_matches_request_sid(*sid, data),
115 _ => false,
116 }
117 }
118}
119
120pub(crate) struct DoipGatewayState<T: EcuAddresses + DoipComParams> {
121 pub(crate) doip_connections: Arc<RwLock<Vec<Arc<DoipConnection>>>>,
122 pub(crate) logical_address_to_connection: Arc<RwLock<HashMap<u16, usize>>>,
123 pub(crate) ecus: Arc<HashMap<String, RwLock<T>>>,
124 /// `None` until the first successful `start()` binds it. No UDP socket is
125 /// created at construction time, only when an authorized activation runs.
126 pub(crate) socket: Arc<Mutex<Option<DoIPUdpSocket>>>,
127 pub(crate) netmask: u32,
128}
129
130/// Owns all connection tasks for one enabled lifecycle so they stop together.
131pub(crate) struct ConnectionTasks(Mutex<Vec<JoinHandle<()>>>);
132
133impl ConnectionTasks {
134 fn new() -> Self {
135 Self(Mutex::new(Vec::new()))
136 }
137
138 pub(crate) async fn push(&self, task: JoinHandle<()>) {
139 self.0.lock().await.push(task);
140 }
141
142 async fn shutdown(&self) {
143 let mut tasks = std::mem::take(&mut *self.0.lock().await);
144 for task in &tasks {
145 task.abort();
146 }
147 for task in tasks.drain(..) {
148 if let Err(error) = task.await
149 && error.is_panic()
150 {
151 tracing::error!(%error, "DoIP connection task panicked during shutdown");
152 }
153 }
154 }
155}
156
157async fn spawn_connection_task(
158 connection_tasks: &Arc<ConnectionTasks>,
159 name: &str,
160 task: impl Future<Output = ()> + Send + 'static,
161) {
162 let task = cda_interfaces::spawn_named!(name, task);
163 connection_tasks.push(task).await;
164}
165
166impl Drop for ConnectionTasks {
167 fn drop(&mut self) {
168 // Best effort cleanup, we cannot await the tasks in
169 // here. A user should call 'shutdown' instead of relying on this.
170 for task in self.0.get_mut().drain(..) {
171 tracing::warn!("DoIP connection tasks dropped without `shutdown`, aborting tasks");
172 task.abort();
173 }
174 }
175}
176
177impl<T: EcuAddresses + DoipComParams> Clone for DoipGatewayState<T> {
178 fn clone(&self) -> Self {
179 Self {
180 doip_connections: Arc::clone(&self.doip_connections),
181 logical_address_to_connection: Arc::clone(&self.logical_address_to_connection),
182 ecus: Arc::clone(&self.ecus),
183 socket: Arc::clone(&self.socket),
184 netmask: self.netmask,
185 }
186 }
187}
188
[docs] 189/// [[ dimpl~deferred-doip-transport, Deferred diagnostic transport activation, dimpl ]]
190pub struct DoipDiagGateway<T: EcuAddresses + DoipComParams> {
191 state: DoipGatewayState<T>,
192 config: DoipConfig,
193 variant_detection: VariantDetectionSender,
194 connectivity_handler: Arc<dyn EcuConnectivityHandler>,
195 lifecycle: Arc<GatewayLifecycle<DoipGatewayOperation>>,
196}
197
198/// Background work owned by the currently enabled lifecycle.
199#[derive(Default)]
200struct DoipGatewayOperation {
201 cancel: Option<CancellationToken>,
202 vam_listener: Option<JoinHandle<()>>,
203 connection_tasks: Option<Arc<ConnectionTasks>>,
204}
205
206impl Drop for DoipGatewayOperation {
207 fn drop(&mut self) {
208 if self.cancel.is_some() || self.vam_listener.is_some() || self.connection_tasks.is_some() {
209 tracing::warn!("DoIP gateway operation dropped without `shutdown`, aborting tasks");
210 }
211 self.take_and_abort();
212 }
213}
214
215impl DoipGatewayOperation {
216 fn take_and_abort(&mut self) -> (Option<JoinHandle<()>>, Option<Arc<ConnectionTasks>>) {
217 if let Some(cancel) = self.cancel.take() {
218 cancel.cancel();
219 }
220 let listener = self.vam_listener.take();
221 if let Some(listener) = &listener {
222 listener.abort();
223 }
224 (listener, self.connection_tasks.take())
225 }
226
227 async fn shutdown(&mut self) {
228 let (listener, connection_tasks) = self.take_and_abort();
229 if let Some(listener) = listener
230 && let Err(error) = listener.await
231 && !error.is_cancelled()
232 {
233 tracing::error!(%error, "DoIP VAM listener failed during shutdown");
234 }
235 if let Some(connection_tasks) = connection_tasks {
236 connection_tasks.shutdown().await;
237 }
238 }
239}
240
241/// A gateway discovered on the network during `DoIP` vehicle discovery.
242///
243/// Contains the IP address, ECU name, and logical address obtained from a
244/// `VehicleIdentificationResponse` or `EntityStatusResponse` message.
245#[derive(Debug)]
246pub(crate) struct DiscoveredGateway {
247 pub(crate) ip: String,
248 pub(crate) ecu_name: String,
249 pub(crate) logical_address: u16,
250 pub(crate) doip_protocol_version: ProtocolVersion,
251}
252
253/// Transport-level settings shared across all `DoIP` gateway connections.
254/// Built once from `DoipConfig` in `DoipDiagGateway::new`; cloned into each
255/// `GatewayDoipConfig` when a specific gateway is discovered.
256#[derive(Clone)]
257pub(crate) struct DoipTransportConfig {
258 /// IP address of the diagnostic tester interface.
259 pub(crate) tester_ip: String,
260 /// TCP port for `DoIP` communication.
261 pub(crate) port: u16,
262 /// TCP port for `DoIP` over TLS communication.
263 pub(crate) tls_port: u16,
264 /// Whether to send a `DiagnosticMessageAck` upon receiving a `DiagnosticMessage`.
265 pub(crate) send_diagnostic_message_ack: bool,
266 /// Timeout for sending a single `DoIP` message.
267 pub(crate) send_timeout: Duration,
268 /// Interval between alive-check requests on idle connections (0 = disabled).
269 pub(crate) alive_check_interval: Duration,
270}
271
272/// Per-gateway configuration combining the discovered gateway identity with the
273/// shared transport settings. Only constructed inside `handle_gateway_connection`
274/// once a real `ip` and `name` are known - never holds placeholder values.
275#[derive(Clone)]
276pub(crate) struct GatewayDoipConfig {
277 /// IP address of the gateway ECU.
278 pub(crate) gateway_ip: String,
279 /// Name of the gateway ECU.
280 pub(crate) name: String,
281 /// UDS address of the tester.
282 pub(crate) tester_address: [u8; 2],
283 /// The `DoIp` protocol version to use for this gateway connection.
284 /// Set from the protocol version field in the VAM.
285 pub(crate) protocol_version: ProtocolVersion,
286 /// Shared transport-level settings (tester IP, ports, socket config, timeouts).
287 pub(crate) transport: DoipTransportConfig,
288}
289
290/// ECU-lifecycle timeouts sourced from `EcuAddresses` / `DoipComParams` during gateway setup.
291#[derive(Clone, Copy)]
292pub(crate) struct EcuTimeouts {
293 pub(crate) routing_activation: Duration,
294 pub(crate) retry_delay: Duration,
295 pub(crate) connection: Duration,
296 pub(crate) max_retry_attempts: u32,
297}
298
299/// Parameters needed to (re-)establish a TCP connection to one `DoIP` gateway.
300/// Cloned into the reconnect task and passed to `ecu_connection` functions;
301/// does **not** carry one-time setup data such as the ECU list or variant-detection channel.
302#[derive(Clone)]
303pub(crate) struct GatewayConnectionConfig {
304 pub(crate) doip: GatewayDoipConfig,
305 pub(crate) routing_activation_request: RoutingActivationRequest,
306 pub(crate) ecu_timeouts: EcuTimeouts,
307}
308
309/// One-shot setup bundle consumed by `connection_handler` when bringing up a new gateway.
310/// Carries `GatewayConnectionConfig` plus the ECU list needed during initial channel creation
311/// and the `ConnectivityHandler` for propagating connection events.
312#[derive(Clone)]
313pub(crate) struct GatewaySetup {
314 pub(crate) connection: GatewayConnectionConfig,
315 pub(crate) ecus: Vec<u16>,
316 pub(crate) ecu_names: Vec<String>,
317 pub(crate) connectivity_handler: Arc<dyn EcuConnectivityHandler>,
318}
319
320struct DoipEcu {
321 sender: mpsc::Sender<DoipPayload>,
322 receiver: broadcast::Receiver<Result<DiagnosticResponse, EcuError>>,
323}
324
325struct DoipConnection {
326 ecus: HashMap<u16, Arc<Mutex<DoipEcu>>>,
327 ip: String,
328}
329
330impl<T: EcuAddresses + DoipComParams> DoipDiagGateway<T> {
331 /// Create a new `DoipDiagGateway` instance.
332 ///
333 /// Purely in-memory. No UDP socket is created or bound, no packet is sent
334 /// and no task is started, so this is safe to call in any `init_mode`.
335 ///
336 /// # Errors
337 /// Returns `String` if initialization fails, e.g. when the configured
338 /// tester address/subnet cannot be parsed into a netmask.
339 #[tracing::instrument(
340 skip(doip_config, ecus, variant_detection, connectivity_handler),
341 fields(
342 tester_ip = doip_config.tester_address,
343 gateway_port = doip_config.gateway_port,
344 ecu_count = ecus.len(),
345 dlt_context = dlt_ctx!("DOIP")
346 )
347 )]
348 pub async fn new(
349 doip_config: &DoipConfig,
350 ecus: Arc<HashMap<String, RwLock<T>>>,
351 variant_detection: VariantDetectionSender,
352 connectivity_handler: Arc<dyn EcuConnectivityHandler>,
353 ) -> Result<Self, DoipGatewaySetupError> {
354 Ok(Self {
355 state: DoipGatewayState {
356 doip_connections: Arc::new(RwLock::new(Vec::new())),
357 logical_address_to_connection: Arc::new(RwLock::new(HashMap::new())),
358 ecus,
359 socket: Arc::new(Mutex::new(None)),
360 netmask: create_netmask(&doip_config.tester_address, &doip_config.tester_subnet)?,
361 },
362 config: doip_config.clone(),
363 variant_detection,
364 connectivity_handler,
365 lifecycle: Arc::new(GatewayLifecycle::new(TransportState::Disabled)),
366 })
367 }
368
369 async fn start(
370 &self,
371 operation: &mut DoipGatewayOperation,
372 ) -> Result<(), DoipGatewaySetupError> {
373 let mut socket_guard = self.state.socket.lock().await;
374 if operation.cancel.is_some()
375 || operation.vam_listener.is_some()
376 || operation.connection_tasks.is_some()
377 || socket_guard.is_some()
378 {
379 return Err(DoipGatewaySetupError::InvalidState(
380 "DoIP communication is already running".to_string(),
381 ));
382 }
383
384 let config = &self.config;
385 let transport_config = DoipTransportConfig {
386 tester_ip: config.tester_address.clone(),
387 port: config.gateway_port,
388 tls_port: config.tls_port,
389 send_diagnostic_message_ack: config.send_diagnostic_message_ack,
390 send_timeout: Duration::from_millis(config.send_timeout_ms),
391 alive_check_interval: Duration::from_secs(config.alive_check_interval_secs),
392 };
393 let mask = self.state.netmask;
394 let cancel = CancellationToken::new();
395 let shutdown = cancel.child_token().cancelled_owned().shared();
396 let connection_tasks = Arc::new(ConnectionTasks::new());
397 operation.cancel = Some(cancel.clone());
398 operation.connection_tasks = Some(Arc::clone(&connection_tasks));
399
400 // TODO(persistence-init-mode): Implement as described in req~dt-ecu-list-persistence
401 let socket = socket_guard.insert(create_udp_vir_socket(
402 &config.tester_address,
403 config.gateway_port,
404 )?);
405 let gateways = vir_vam::get_vehicle_identification::<T, _>(
406 socket,
407 mask,
408 config.gateway_port,
409 &self.state.ecus,
410 shutdown.clone(),
411 )
412 .await
413 .map_err(|error| DoipGatewaySetupError::ResourceError(error.to_string()))?;
414 drop(socket_guard);
415
416 let mut gateway_ecu_map: HashMap<u16, Vec<u16>> = HashMap::new();
417 let mut gateway_ecu_name_map: HashMap<u16, Vec<String>> = HashMap::new();
418 for (ecu_name, ecu_lock) in self.state.ecus.iter() {
419 let ecu = ecu_lock.read().await;
420 gateway_ecu_map
421 .entry(ecu.logical_gateway_address())
422 .or_default()
423 .push(ecu.logical_address());
424 gateway_ecu_name_map
425 .entry(ecu.logical_gateway_address())
426 .or_default()
427 .push(ecu_name.clone());
428 }
429 for gateway in gateways {
430 let gateway_name = gateway.ecu_name.clone();
431 let gateway_ecu_names = gateway_ecu_name_map
432 .get(&gateway.logical_address)
433 .cloned()
434 .unwrap_or_default();
435 let variant_detection = self.variant_detection.clone();
436 let transport_config = transport_config.clone();
437 let doip_connections = Arc::clone(&self.state.doip_connections);
438 let logical_address_to_connection =
439 Arc::clone(&self.state.logical_address_to_connection);
440 let gateway_state = GatewayState {
441 doip_connections: Arc::clone(&doip_connections),
442 ecus: Arc::clone(&self.state.ecus),
443 gateway_ecu_map: gateway_ecu_map.clone(),
444 connection_tasks: Arc::clone(&connection_tasks),
445 };
446 let connectivity_handler = Arc::clone(&self.connectivity_handler);
447 spawn_connection_task(
448 &connection_tasks,
449 &format!("doip-gateway-connect-{gateway_name}"),
450 async move {
451 match connections::handle_gateway_connection::<T>(
452 gateway,
453 &transport_config,
454 &gateway_state,
455 connectivity_handler,
456 )
457 .await
458 {
459 Ok(logical_address) => {
460 let connection_index =
461 doip_connections.read().await.iter().position(|connection| {
462 connection.ecus.contains_key(&logical_address)
463 });
464 if let Some(connection_index) = connection_index {
465 logical_address_to_connection
466 .write()
467 .await
468 .insert(logical_address, connection_index);
469 }
470 if let Err(error) = variant_detection
471 .send(VariantDetectionRequest::new(gateway_ecu_names))
472 .await
473 {
474 tracing::warn!(
475 gateway = %gateway_name,
476 error = ?error,
477 "Failed to trigger variant detection after gateway connection"
478 );
479 }
480 }
481 Err(error) => {
482 tracing::warn!(
483 gateway = %gateway_name,
484 error = %error,
485 "Failed to establish initial gateway connection"
486 );
487 }
488 }
489 },
490 )
491 .await;
492 }
493
494 let listener = vir_vam::listen_for_vams(
495 transport_config,
496 mask,
497 self.state.clone(),
498 Arc::clone(&connection_tasks),
499 self.variant_detection.clone(),
500 Arc::clone(&self.connectivity_handler),
501 shutdown,
502 )
503 .await;
504 operation.vam_listener = Some(listener);
505 Ok(())
506 }
507
508 async fn stop(&self, operation: &mut DoipGatewayOperation) {
509 operation.shutdown().await;
510 self.state.socket.lock().await.take();
511 self.state.doip_connections.write().await.clear();
512 self.state
513 .logical_address_to_connection
514 .write()
515 .await
516 .clear();
517 }
518
519 async fn get_doip_connection(
520 &self,
521 logical_address: u16,
522 ) -> Result<Arc<DoipConnection>, DiagServiceError> {
523 let conn_idx = *self
524 .state
525 .logical_address_to_connection
526 .read()
527 .await
528 .get(&logical_address)
529 .ok_or_else(|| DiagServiceError::EcuOffline(format!("[{logical_address}]")))?;
530
531 let lock = self.state.doip_connections.read().await;
532 let conn = lock
533 .get(conn_idx)
534 .ok_or(DiagServiceError::ConnectionClosed(format!(
535 "Connection entry for address {logical_address} found, but it was already closed"
536 )))?;
537
538 Ok(Arc::clone(conn))
539 }
540
541 async fn get_ecu_mtx(
542 &self,
543 doip_conn: &DoipConnection,
544 message: &ServicePayload,
545 transmission_params: &TransmissionParameters,
546 ) -> Result<Arc<Mutex<DoipEcu>>, DiagServiceError> {
547 // first try looking up with the target address.
548 if let Some(ecu) = doip_conn.ecus.get(&message.target_address) {
549 return Ok(Arc::clone(ecu));
550 }
551
552 // if we cannot find the target address,
553 // the request might be sent on the functional address
554 // in that case, lookup the ecu name and check if the functional address
555 // matches the given address.
556 // this will be the case for tester present.
557 if let Some(ecu) = self
558 .state
559 .ecus
560 .get(&transmission_params.ecu_name.to_lowercase())
561 && ecu.read().await.logical_functional_address() == message.target_address
562 && let Some(gateway_ecu) = doip_conn.ecus.get(&transmission_params.gateway_address)
563 {
564 return Ok(Arc::clone(gateway_ecu));
565 }
566
567 Err(DiagServiceError::EcuOffline(
568 transmission_params.ecu_name.clone(),
569 ))
570 }
571}
572
573impl<T: EcuAddresses + DoipComParams> PhysicalTransport for DoipDiagGateway<T> {
574 #[tracing::instrument(skip_all,
575 fields(dlt_context = dlt_ctx!("DOIP"))
576 )]
577
578 async fn send(
579 &self,
580 transmission_params: TransmissionParameters,
581 message: ServicePayload,
582 response_sender: mpsc::Sender<Result<Option<TransportResponse>, DiagServiceError>>,
583 expect_uds_reply: bool,
584 ) -> Result<tokio::task::JoinHandle<()>, DiagServiceError> {
585 let start = Instant::now();
586
587 let doip_conn = self
588 .get_doip_connection(transmission_params.gateway_address)
589 .await?;
590 let ecu_mtx = self
591 .get_ecu_mtx(&doip_conn, &message, &transmission_params)
592 .await?;
593
594 let doip_message = DiagnosticMessage {
595 source_address: message.source_address.to_be_bytes(),
596 target_address: message.target_address.to_be_bytes(),
597 message: message.data,
598 };
599
600 let handle = cda_interfaces::spawn_named!(
601 &format!("ecu-data-receive-{}", transmission_params.ecu_name),
602 {
603 async move {
604 let mut ecu = ecu_mtx.lock().await;
605 let lock_acquired = start.elapsed();
606 tracing::debug!(
607 ecu_name = %transmission_params.ecu_name,
608 locked_after = ?lock_acquired,
609 message_data = %util::tracing::print_hex(&doip_message.message, 8),
610 "Sending Message to ECU"
611 );
612
613 // Clear any pending messages
614 tokio_ext::clear_pending_messages(&mut ecu.receiver);
615 let receiver_flushed = start.elapsed().saturating_sub(lock_acquired);
616
617 let mut resend_counter = 0;
618 if let Err(e) = send_with_retries(
619 &doip_message,
620 &ecu.sender,
621 &mut resend_counter,
622 transmission_params.repeat_request_count_transmission,
623 )
624 .await
625 {
626 // failed to send the message after exhausting retries.
627 // informing receiver and giving up.
628 try_send_transport_response(&response_sender, Err(e)).await;
629 return;
630 }
631
632 let received_event = match wait_for_ack_or_response_until_timeout(
633 &mut ecu.receiver,
634 &transmission_params.ecu_name,
635 &doip_message,
636 transmission_params.timeout_ack,
637 )
638 .await
639 {
640 Ok(first) => first,
641 Err(e) => {
642 try_send_transport_response(&response_sender, Err(e)).await;
643 return;
644 }
645 };
646
647 let send_and_ackd_after = start
648 .elapsed()
649 .saturating_sub(lock_acquired)
650 .saturating_sub(receiver_flushed);
651
652 if !expect_uds_reply {
653 try_send_transport_response(&response_sender, Ok(None)).await;
654 }
655
656 // Read ECU responses as long as the sender is open.
657 // We might get multiple responses for a single request,
658 // e.g. when the ECU is busy and sends NRC 0x78.
659 read_ecu_responses(
660 &mut ecu.receiver,
661 &transmission_params.ecu_name,
662 &response_sender,
663 received_event,
664 )
665 .await;
666
667 let rx_done = start
668 .elapsed()
669 .saturating_sub(lock_acquired)
670 .saturating_sub(send_and_ackd_after)
671 .saturating_sub(receiver_flushed);
672 tracing::debug!(
673 ecu_name = %transmission_params.ecu_name,
674 total_duration = ?start.elapsed(),
675 lock_duration = ?lock_acquired,
676 flush_duration = ?receiver_flushed,
677 send_ack_duration = ?send_and_ackd_after,
678 response_duration = ?rx_done,
679 "Handled DOIP request timing breakdown"
680 );
681 }
682 }
683 );
684
685 Ok(handle)
686 }
687
688 async fn ecu_online<E: EcuAddresses>(
689 &self,
690 ecu_name: &str,
691 ecu_db: &RwLock<E>,
692 ) -> Result<(), DiagServiceError> {
693 let ecu_lock = ecu_db.read().await;
694
695 let doip_conn = self
696 .get_doip_connection(ecu_lock.logical_gateway_address())
697 .await?;
698 doip_conn
699 .ecus
700 .get(&ecu_lock.logical_address())
701 .ok_or_else(|| DiagServiceError::EcuOffline(ecu_name.to_owned()))?;
702 Ok(())
703 }
704}
705
706impl<T: EcuAddresses + DoipComParams> FunctionalTransport for DoipDiagGateway<T> {
707 async fn send_functional(
708 &self,
709 transmission_params: TransmissionParameters,
710 message: ServicePayload,
711 expected_ecu_logical_addrs: HashMap<u16, String>,
712 timeout: Duration,
713 expect_positive_response: bool,
714 ) -> Result<HashMap<String, Result<ServicePayload, DiagServiceError>>, DiagServiceError> {
715 let doip_conn = self
716 .get_doip_connection(transmission_params.gateway_address)
717 .await?;
718
719 // Get the gateway ECU for sending the functional request
720 let gateway_ecu = doip_conn
721 .ecus
722 .get(&transmission_params.gateway_address)
723 .ok_or_else(|| DiagServiceError::EcuOffline("Gateway ECU not found".to_string()))?;
724
725 let doip_message = DiagnosticMessage {
726 source_address: message.source_address.to_be_bytes(),
727 target_address: message.target_address.to_be_bytes(),
728 message: message.data,
729 };
730
731 let mut result_map = HashMap::new();
732 let expected_count = expected_ecu_logical_addrs.len();
733
734 tracing::debug!(
735 gateway_address = %transmission_params.gateway_address,
736 expected_ecus = expected_count,
737 message_data = %util::tracing::print_hex(&doip_message.message, 8),
738 "Sending functional request to gateway"
739 );
740
741 // Send the functional request once
742 let mut ecu = gateway_ecu.lock().await;
743 let mut ecu_mtxs = expected_ecu_logical_addrs
744 .iter()
745 .filter_map(|(addr, name)| {
746 if *addr == transmission_params.gateway_address {
747 None
748 } else {
749 doip_conn
750 .ecus
751 .get(addr)
752 .cloned()
753 .map(|ecu| (name.clone(), ecu))
754 }
755 })
756 .collect::<Vec<_>>();
757
758 // Clear any pending messages
759 tokio_ext::clear_pending_messages(&mut ecu.receiver);
760
761 let mut resend_counter = 0;
762 send_with_retries(
763 &doip_message,
764 &ecu.sender,
765 &mut resend_counter,
766 transmission_params.repeat_request_count_transmission,
767 )
768 .await?;
769
770 drop(ecu); // release lock before waiting for responses
771 ecu_mtxs.push((
772 transmission_params.ecu_name.to_lowercase(),
773 Arc::clone(gateway_ecu),
774 ));
775
776 // Use a short window to capture any negative responses instead of the full timeout,
777 // as NRCs are usually sent immediately by the ECUs.
778 let response_timeout = if expect_positive_response {
779 timeout
780 } else {
781 SUPPRESS_POSITIVE_RESPONSE_TIMEOUT
782 };
783
784 let received_responses: Arc<Mutex<HashMap<String, Result<DiagnosticMessage, EcuError>>>> =
785 Arc::new(Mutex::new(HashMap::new()));
786 let mut futures = Vec::new();
787 for (name, ecu) in ecu_mtxs.drain(..) {
788 let received_responses = Arc::clone(&received_responses);
789 let fut = async move {
790 let mut lock = ecu.lock().await;
791 if let Some(response) = wait_for_ecu_response(&mut lock, response_timeout).await {
792 received_responses.lock().await.insert(name, response);
793 }
794 };
795 futures.push(fut);
796 }
797
798 futures::future::join_all(futures).await;
799
800 for (ecu_name, msg) in received_responses.lock().await.drain() {
801 if !result_map.contains_key(&ecu_name) {
802 match msg {
803 Ok(msg) => {
804 let source_addr = u16::from_be_bytes(msg.source_address);
805
806 let uds_response = ServicePayload {
807 data: msg.message,
808 source_address: source_addr,
809 target_address: u16::from_be_bytes(msg.target_address),
810 new_session: None,
811 new_security: None,
812 };
813
814 result_map.insert(ecu_name.clone(), Ok(uds_response));
815
816 tracing::debug!(
817 ecu_name = %ecu_name,
818 source_addr = source_addr,
819 "Received functional response"
820 );
821 }
822 Err(e) => {
823 tracing::debug!(
824 ecu_name = %ecu_name,
825 "Error receiving functional response: {e}"
826 );
827 result_map.insert(ecu_name.clone(), Err(e.into()));
828 }
829 }
830 }
831 }
832
833 // For ECUs that did not respond:
834 // Insert a TimeoutError for ECUs that did not response, if
835 // a positive response is expected. (suppress bit is not set)
836 for (logical_addr, ecu_name) in &expected_ecu_logical_addrs {
837 if !result_map.contains_key(ecu_name) && expect_positive_response {
838 result_map.insert(ecu_name.clone(), Err(DiagServiceError::Timeout));
839 tracing::debug!(
840 ecu_name = %ecu_name,
841 logical_addr = logical_addr,
842 "ECU did not respond to functional request"
843 );
844 }
845 }
846
847 Ok(result_map)
848 }
849}
850
851impl<T: EcuAddresses + DoipComParams> NetworkTopology for DoipDiagGateway<T> {
852 async fn get_gateway_network_address(&self, logical_address: u16) -> Option<String> {
853 self.state
854 .doip_connections
855 .read()
856 .await
857 .iter()
858 .find(|conn| conn.ecus.contains_key(&logical_address))
859 .map(|conn| conn.ip.clone())
860 }
861}
862
863impl<T: EcuAddresses + DoipComParams> TransportProbe for DoipDiagGateway<T> {
864 async fn route_status(&self, ecu_name: &str) -> RouteStatus {
865 let ecu_name = ecu_name.to_lowercase();
866 let Some(ecu_lock) = self.state.ecus.get(&ecu_name) else {
867 return RouteStatus::NotConfigured;
868 };
869 // Check if the gateway connection for this ECU is established
870 let gateway_addr = ecu_lock.read().await.logical_gateway_address();
871 if self.get_doip_connection(gateway_addr).await.is_ok() {
872 RouteStatus::Ready
873 } else {
874 // DoIP addressing is gateway-scoped, so there is
875 // no per-ECU probe to offer. Reconnecting is this transport's job
876 // (VAM listener, connection-reset task), not the router's.
877 RouteStatus::Unavailable
878 }
879 }
880
881 fn probe_ecu(&self, _ecu_name: &str) -> impl Future<Output = bool> {
882 std::future::ready(false)
883 }
884}
885/// Waits for the `DoIP` diagnostic-message acknowledgement from the gateway,
886/// with a deadline of `timeout`.
887#[allow(
888 clippy::needless_continue,
889 reason = "Explicit continue improves readability of wait logic"
890)]
891async fn wait_for_ack_or_response_until_timeout(
892 receiver: &mut broadcast::Receiver<Result<DiagnosticResponse, EcuError>>,
893 ecu_name: &str,
894 sent_message: &DiagnosticMessage,
895 timeout: Duration,
896) -> Result<Option<DiagnosticResponse>, DiagServiceError> {
897 async fn wait_for_ack_or_response(
898 receiver: &mut broadcast::Receiver<Result<DiagnosticResponse, EcuError>>,
899 ecu_name: &str,
900 sent_message: &DiagnosticMessage,
901 ) -> Result<Option<DiagnosticResponse>, DiagServiceError> {
902 loop {
903 let response = match receiver.recv().await {
904 Ok(Ok(response)) => response,
905 Ok(Err(e)) => {
906 tracing::error!(
907 ecu_name = %ecu_name,
908 error = %e,
909 "Error while waiting for ACK/NACK"
910 );
911
912 return Err(DiagServiceError::NoResponse(format!(
913 "Error while waiting for ACK/NACK, {e}"
914 )));
915 }
916 Err(_) => {
917 tracing::error!(
918 ecu_name = %ecu_name,
919 "ECU receiver unexpectedly closed while waiting for ACK/NACK"
920 );
921
922 return Err(DiagServiceError::NoResponse(
923 "ECU receiver unexpectedly closed".to_owned(),
924 ));
925 }
926 };
927
928 match &response {
929 DiagnosticResponse::Nack(nack) => {
930 tracing::warn!(
931 ecu_name = %ecu_name,
932 nack_code = ?nack.nack_code,
933 "Received NACK"
934 );
935 return Err(DiagServiceError::Nack(u8::from(nack.nack_code)));
936 }
937 DiagnosticResponse::GenericNack(nack) => {
938 tracing::warn!(
939 ecu_name = %ecu_name,
940 nack_code = ?nack.nack_code,
941 "Received generic NACK"
942 );
943 return Err(DiagServiceError::Nack(u8::from(nack.nack_code)));
944 }
945 _ => {}
946 }
947
948 if !response.matches_request(&sent_message.message) {
949 if let DiagnosticResponse::Ack((_, previous)) = &response {
950 tracing::warn!(
951 ecu_name = %ecu_name,
952 previous = %util::tracing::print_hex(previous, 8),
953 sent = %util::tracing::print_hex(&sent_message.message, 8),
954 "ACK previous message does not match sent message"
955 );
956 } else {
957 tracing::debug!(
958 ecu_name = %ecu_name,
959 ?response,
960 "Received response does not match sent message. Ignoring."
961 );
962 }
963
964 continue;
965 }
966
967 return match response {
968 DiagnosticResponse::Ack((_, _)) => {
969 tracing::debug!(ecu_name = %ecu_name, "Received ACK");
970 Ok(None)
971 }
972 response => {
973 tracing::debug!(
974 ecu_name = %ecu_name,
975 "Received diagnostic response before ACK, treating as implicit ACK"
976 );
977 Ok(Some(response))
978 }
979 };
980 }
981 }
982
983 if let Ok(result) = tokio::time::timeout(
984 timeout,
985 wait_for_ack_or_response(receiver, ecu_name, sent_message),
986 )
987 .await
988 {
989 result
990 } else {
991 tracing::warn!(
992 ecu_name = %ecu_name,
993 timeout = ?timeout,
994 "Timeout waiting for ACK/NACK from ECU"
995 );
996 Err(DiagServiceError::Timeout)
997 }
998}
999
1000/// Reads ECU responses from `receiver` and forwards them through
1001/// `response_sender` until either the sender is closed (caller no longer
1002/// interested) or the receiver is closed (connection dropped).
1003///
1004/// `received_message` carries an already-received response from the implicit-ACK path
1005/// (where the ECU skipped the ACK and sent the diagnostic reply directly).
1006/// Pass `None` after a normal explicit ACK.
1007async fn read_ecu_responses(
1008 receiver: &mut broadcast::Receiver<Result<DiagnosticResponse, EcuError>>,
1009 ecu_name: &str,
1010 response_sender: &mpsc::Sender<Result<Option<TransportResponse>, DiagServiceError>>,
1011 received_event: Option<DiagnosticResponse>,
1012) {
1013 if let Some(event) = received_event
1014 && !try_send_transport_response(response_sender, diagnostic_response_to_transport(event))
1015 .await
1016 {
1017 return;
1018 }
1019 loop {
1020 tokio::select! {
1021 // Using biased saves a bit of CPU time because tokio does not
1022 // have to generate a random number to select the branch.
1023 // Prioritizing the ECU receiver over the closed handler is fine
1024 // because it is unlikely that both fire at the exact same time.
1025 biased;
1026 res = receiver.recv() => {
1027 if let Ok(res) = res {
1028 match res {
1029 Ok(event) => {
1030 if !try_send_transport_response(
1031 response_sender,
1032 diagnostic_response_to_transport(event),
1033 )
1034 .await
1035 {
1036 break;
1037 }
1038 }
1039 Err(e) => {
1040 tracing::error!(
1041 ecu_name = %ecu_name,
1042 error = %e,
1043 "Error while waiting for response message"
1044 );
1045 if !try_send_transport_response(
1046 response_sender,
1047 Err(DiagServiceError::NoResponse(format!(
1048 "Error while waiting for message, {e}"
1049 ))),
1050 )
1051 .await
1052 {
1053 break;
1054 }
1055 }
1056 }
1057 } else {
1058 tracing::error!(
1059 ecu_name = %ecu_name,
1060 "ECU receiver unexpectedly closed while waiting for response"
1061 );
1062 try_send_transport_response(
1063 response_sender,
1064 Err(DiagServiceError::NoResponse(
1065 "ECU receiver unexpectedly closed".to_owned(),
1066 )),
1067 )
1068 .await;
1069 break;
1070 }
1071 }
1072 () = response_sender.closed() => {
1073 tracing::debug!(
1074 ecu_name = %ecu_name,
1075 "Response sender closed, aborting loop"
1076 );
1077 break;
1078 }
1079 }
1080 }
1081}
1082
1083/// Converts a [`DiagnosticResponse::Msg`] to a [`TransportResponse`] result for the
1084/// caller channel.
1085///
1086/// Pending NRCs are classified via [`pending_nrc_from_raw`]; all other frames
1087/// are classified as final via [`uds_response_from_raw`].
1088///
1089/// `TesterPresentNRC` is absorbed by [`connection_receiver`] before it is
1090/// broadcast to any per-ECU channel -- it never reaches this function.
1091/// All other [`DiagnosticResponse`] variants (Ack, Nack, etc.) must be
1092/// handled before calling this function.
1093fn diagnostic_response_to_transport(
1094 event: DiagnosticResponse,
1095) -> Result<Option<TransportResponse>, DiagServiceError> {
1096 match event {
1097 DiagnosticResponse::Msg {
1098 data,
1099 source_address,
1100 target_address,
1101 } => {
1102 if let Some(pending) = pending_nrc_from_raw(&data, source_address) {
1103 Ok(Some(TransportResponse::Pending(pending)))
1104 } else {
1105 Ok(Some(TransportResponse::UdsResponse(uds_response_from_raw(
1106 data,
1107 source_address,
1108 target_address,
1109 ))))
1110 }
1111 }
1112 _ => Err(DiagServiceError::BadPayload(
1113 "Unexpected DoIP event type in UDS response stream".to_owned(),
1114 )),
1115 }
1116}
1117
1118#[allow(
1119 clippy::needless_continue,
1120 reason = "Explicit continue improves readability of complex loop logic"
1121)]
1122async fn wait_for_ecu_response(
1123 ecu: &mut DoipEcu,
1124 timeout: Duration,
1125) -> Option<Result<DiagnosticMessage, EcuError>> {
1126 tokio::time::timeout(timeout, async {
1127 loop {
1128 match ecu.receiver.recv().await {
1129 Ok(Ok(DiagnosticResponse::Msg {
1130 source_address,
1131 target_address,
1132 data,
1133 })) => {
1134 return Some(Ok(DiagnosticMessage {
1135 source_address: source_address.to_be_bytes(),
1136 target_address: target_address.to_be_bytes(),
1137 message: data,
1138 }));
1139 }
1140 Ok(Ok(_ignore)) => {
1141 // Ignore other event types
1142 continue;
1143 }
1144 Ok(Err(e)) => {
1145 return Some(Err(e));
1146 }
1147 Err(_) => {
1148 // Receiver closed
1149 return None;
1150 }
1151 }
1152 }
1153 })
1154 .await
1155 .unwrap_or_default()
1156}
1157
1158fn create_netmask(tester_ip: &str, tester_subnet: &str) -> Result<u32, DoipGatewaySetupError> {
1159 let ip = tester_ip.parse::<std::net::Ipv4Addr>().map_err(|e| {
1160 DoipGatewaySetupError::InvalidAddress(format!(
1161 "DoipGateway: Failed to parse tester IP address: {e:?}"
1162 ))
1163 })?;
1164 let subnet = tester_subnet.parse::<std::net::Ipv4Addr>().map_err(|e| {
1165 DoipGatewaySetupError::InvalidAddress(format!(
1166 "DoipGateway: Failed to parse tester subnet mask: {e:?}"
1167 ))
1168 })?;
1169
1170 Ok(ip.to_bits() & subnet.to_bits())
1171}
1172
1173/// Creates a UDP socket for `DoIP` communication.
1174///
1175/// # Errors
1176/// Returns errors when:
1177/// * The provided `tester_ip` and `gateway_port` cannot be parsed into a valid `SocketAddr`,
1178/// resulting in a `DoipGatewaySetupError::InvalidAddress`.
1179/// * The underlying system call to create a new socket fails,
1180/// resulting in a `DoipGatewaySetupError::SocketCreationFailed`.
1181/// * Setting the `SO_REUSEADDR` socket option fails,
1182/// resulting in a `DoipGatewaySetupError::InvalidAddress`.
1183/// * On Unix-like systems, setting the `SO_REUSEPORT` socket option fails,
1184/// resulting in a `DoipGatewaySetupError::PortBindFailed`.
1185/// * Setting the `SO_BROADCAST` socket option fails,
1186/// resulting in a `DoipGatewaySetupError::SocketCreationFailed`.
1187/// * Binding the socket to the specified `tester_ip` and `gateway_port` fails,
1188/// resulting in a `DoipGatewaySetupError::SocketCreationFailed`.
1189/// * The `DoIPUdpSocket` constructor fails to create the DoIP-specific socket from the standard UDP socket,
1190/// resulting in a `DoipGatewaySetupError::SocketCreationFailed`.
1191pub fn create_udp_vir_socket(
1192 tester_ip: &str,
1193 gateway_port: u16,
1194) -> Result<DoIPUdpSocket, DoipGatewaySetupError> {
1195 let tester_ip = match tester_ip {
1196 "127.0.0.1" => "0.0.0.0",
1197 _ => tester_ip,
1198 };
1199 let broadcast_addr: std::net::SocketAddr =
1200 format!("{tester_ip}:{gateway_port}").parse().map_err(|e| {
1201 DoipGatewaySetupError::InvalidAddress(format!(
1202 "DoipGateway: Failed to create broadcast addr: {e:?}"
1203 ))
1204 })?;
1205
1206 let socket = socket2::Socket::new(
1207 socket2::Domain::IPV4,
1208 socket2::Type::DGRAM,
1209 Some(socket2::Protocol::UDP),
1210 )
1211 .map_err(|e| {
1212 DoipGatewaySetupError::SocketCreationFailed(format!(
1213 "DoipGateway: Failed to create socket: {e:?}"
1214 ))
1215 })?;
1216
1217 socket.set_reuse_address(true).map_err(|e| {
1218 DoipGatewaySetupError::InvalidAddress(format!(
1219 "DoipGateway: Failed to set reuse address: {e:?}"
1220 ))
1221 })?;
1222 #[cfg(target_family = "unix")]
1223 socket.set_reuse_port(true).map_err(|e| {
1224 DoipGatewaySetupError::PortBindFailed(format!(
1225 "DoipGateway: Failed to set reuse port: {e:?}"
1226 ))
1227 })?;
1228 socket.set_broadcast(true).map_err(|e| {
1229 DoipGatewaySetupError::SocketCreationFailed(format!(
1230 "DoipGateway: Failed to set broadcast flag on socket: {e:?}"
1231 ))
1232 })?;
1233 socket.set_nonblocking(true).map_err(|e| {
1234 DoipGatewaySetupError::InvalidConfiguration(format!(
1235 "DoipGateway: Failed to set non-blocking mode: {e:?}"
1236 ))
1237 })?;
1238
1239 socket.bind(&broadcast_addr.into()).map_err(|e| {
1240 DoipGatewaySetupError::SocketCreationFailed(format!(
1241 "DoipGateway: Failed to bind socket, ip {tester_ip}, port {gateway_port}: {e:?}"
1242 ))
1243 })?;
1244
1245 let std_sock: std::net::UdpSocket = socket.into();
1246 DoIPUdpSocket::new(std_sock, ProtocolVersion::DefaultValue).map_err(|e| {
1247 DoipGatewaySetupError::SocketCreationFailed(format!(
1248 "DoipGateway: Failed to create DoIP socket from std socket: {e:?}"
1249 ))
1250 })
1251}
1252
1253impl<T: EcuAddresses + DoipComParams> Clone for DoipDiagGateway<T> {
1254 fn clone(&self) -> Self {
1255 Self {
1256 state: self.state.clone(),
1257 config: self.config.clone(),
1258 variant_detection: self.variant_detection.clone(),
1259 connectivity_handler: Arc::clone(&self.connectivity_handler),
1260 lifecycle: Arc::clone(&self.lifecycle),
1261 }
1262 }
1263}
1264
1265#[async_trait]
1266impl<T: EcuAddresses + DoipComParams> cda_interfaces::Shutdown for DoipDiagGateway<T> {
1267 async fn shutdown(&self) {
1268 if let Err(error) = self.disable().await {
1269 tracing::error!(%error, "Failed to disable DoIP communication during shutdown");
1270 }
1271 }
1272}
1273
1274#[async_trait]
1275impl<T: EcuAddresses + DoipComParams> TransportControl for DoipDiagGateway<T> {
1276 async fn enable(&self) -> Result<(), CommControlError> {
1277 let mut operation = self.lifecycle.operation.lock().await;
1278 if self.lifecycle.coordinator.active().await {
1279 return Ok(());
1280 }
1281 // Clean up any residual resources, usually should be a no-op
1282 self.stop(&mut operation).await;
1283 self.lifecycle
1284 .coordinator
1285 .transition(TransportState::Enabling)
1286 .await;
1287 match self.start(&mut operation).await {
1288 Ok(()) => {
1289 self.lifecycle
1290 .coordinator
1291 .transition(TransportState::Enabled)
1292 .await;
1293 Ok(())
1294 }
1295 Err(error) => {
1296 self.stop(&mut operation).await;
1297 self.lifecycle
1298 .coordinator
1299 .transition(TransportState::Failed)
1300 .await;
1301 Err(CommControlError::InitFailed(error.to_string()))
1302 }
1303 }
1304 }
1305
1306 async fn disable(&self) -> Result<(), CommControlError> {
1307 let mut operation = self.lifecycle.operation.lock().await;
1308 self.lifecycle
1309 .coordinator
1310 .transition(TransportState::Disabling)
1311 .await;
1312 self.stop(&mut operation).await;
1313 self.lifecycle
1314 .coordinator
1315 .transition(TransportState::Disabled)
1316 .await;
1317 Ok(())
1318 }
1319
1320 async fn state(&self) -> TransportState {
1321 self.lifecycle.coordinator.state().await
1322 }
1323}
1324
1325async fn send_with_retries(
1326 msg: &DiagnosticMessage,
1327 sender: &mpsc::Sender<DoipPayload>,
1328 resend_counter: &mut u32,
1329 max_retries: u32,
1330) -> Result<(), DiagServiceError> {
1331 while let Err(e) = sender
1332 .send(DoipPayload::DiagnosticMessage(msg.clone()))
1333 .await
1334 {
1335 *resend_counter = resend_counter.saturating_add(1);
1336 if *resend_counter > max_retries {
1337 return Err(DiagServiceError::SendFailed(format!(
1338 "Failed to send message after {max_retries} attempts: {e:?}",
1339 )));
1340 }
1341 }
1342 Ok(())
1343}
1344
1345#[tracing::instrument(skip_all,
1346 fields(dlt_context = dlt_ctx!("DOIP"))
1347)]
1348async fn try_send_transport_response(
1349 response_sender: &mpsc::Sender<Result<Option<TransportResponse>, DiagServiceError>>,
1350 response: Result<Option<TransportResponse>, DiagServiceError>,
1351) -> bool {
1352 if let Err(err) = response_sender.send(response).await {
1353 tracing::error!(error = %err, "Failed to send response");
1354 return false;
1355 }
1356 true
1357}
1358
1359#[cfg(test)]
1360mod tests {
1361 use std::{net::UdpSocket, sync::Arc, time::Duration};
1362
1363 use cda_interfaces::{
1364 DiagServiceError, EcuConnectivityHandler, HashMap, HashMapExtensions, PendingNrc,
1365 PhysicalTransport, ServicePayload, TransmissionParameters, TransportResponse,
1366 UDS_ID_RESPONSE_BITMASK, VariantDetectionSender,
1367 communication_control::{GatewayLifecycle, TransportState},
1368 nrc, service_ids,
1369 };
1370 use doip_definitions::{
1371 header::ProtocolVersion,
1372 payload::{DiagnosticMessage, DoipPayload},
1373 };
1374 use tokio::sync::{Mutex, RwLock, broadcast, mpsc};
1375
1376 use crate::{
1377 ConnectionTasks, DiagnosticResponse, DoIPUdpSocket, DoipConfig, DoipConnection,
1378 DoipDiagGateway, DoipEcu, DoipGatewayState, read_ecu_responses, spawn_connection_task,
1379 test_helpers::TestEcu, wait_for_ack_or_response_until_timeout,
1380 };
1381
1382 const ECU_ADDR: u16 = 0x0E80;
1383 const GATEWAY_ADDR: u16 = 0x1234;
1384 const TESTER_ADDR: u16 = 0x0E00;
1385 const PAYLOAD: [u8; 2] = [0xF1, 0x90];
1386 const REQUEST_DATA: [u8; 3] = [service_ids::READ_DATA_BY_IDENTIFIER, PAYLOAD[0], PAYLOAD[1]];
1387 const RESPONSE_DATA: [u8; 4] = [
1388 service_ids::READ_DATA_BY_IDENTIFIER | UDS_ID_RESPONSE_BITMASK,
1389 PAYLOAD[0],
1390 PAYLOAD[1],
1391 0x01,
1392 ];
1393
1394 struct TestConnectivityHandler;
1395
1396 #[async_trait::async_trait]
1397 impl EcuConnectivityHandler for TestConnectivityHandler {
1398 async fn on_gateway_connected(&self, _ecu_names: &[String]) {}
1399
1400 async fn on_gateway_disconnected(&self, _ecu_names: &[String]) {}
1401 }
1402
1403 /// Builds a minimal `DoipDiagGateway` whose single ECU is backed by the
1404 /// caller-supplied broadcast receiver and mpsc sender.
1405 ///
1406 /// The mpsc `_doip_payload_rx` must be kept alive by the caller for the
1407 /// duration of the test; `send_with_retries` writes into it and will error
1408 /// if the receiver is dropped before it gets a chance to send.
1409 fn make_gateway(
1410 broadcast_rx: broadcast::Receiver<Result<DiagnosticResponse, crate::connections::EcuError>>,
1411 doip_payload_tx: mpsc::Sender<DoipPayload>,
1412 ) -> DoipDiagGateway<TestEcu> {
1413 let ecu = Arc::new(Mutex::new(DoipEcu {
1414 sender: doip_payload_tx,
1415 receiver: broadcast_rx,
1416 }));
1417
1418 let mut ecus = HashMap::new();
1419 ecus.insert(ECU_ADDR, ecu);
1420 let conn = Arc::new(DoipConnection {
1421 ecus,
1422 ip: "127.0.0.1".to_owned(),
1423 });
1424
1425 // DoIPUdpSocket is never touched during send()
1426 let std_sock = UdpSocket::bind("127.0.0.1:0").expect("bind test socket");
1427 std_sock
1428 .set_nonblocking(true)
1429 .expect("set nonblocking for test socket");
1430 let udp_socket = DoIPUdpSocket::new(std_sock, ProtocolVersion::Iso13400_2012)
1431 .expect("create test DoIPUdpSocket");
1432
1433 let mut addr_map: HashMap<u16, usize> = HashMap::new();
1434 addr_map.insert(GATEWAY_ADDR, 0);
1435
1436 let state = DoipGatewayState {
1437 doip_connections: Arc::new(RwLock::new(vec![conn])),
1438 logical_address_to_connection: Arc::new(RwLock::new(addr_map)),
1439 ecus: Arc::new(HashMap::new()),
1440 socket: Arc::new(Mutex::new(Some(udp_socket))),
1441 netmask: 0,
1442 };
1443
1444 DoipDiagGateway {
1445 state,
1446 config: DoipConfig::default(),
1447 variant_detection: VariantDetectionSender::new(mpsc::channel(1).0),
1448 connectivity_handler: Arc::new(TestConnectivityHandler),
1449 lifecycle: Arc::new(GatewayLifecycle::new(TransportState::Enabled)),
1450 }
1451 }
1452
1453 #[tokio::test]
1454 async fn registering_connection_task_does_not_wait_for_connection() {
1455 let tasks = Arc::new(ConnectionTasks::new());
1456
1457 tokio::time::timeout(
1458 Duration::from_millis(100),
1459 spawn_connection_task(&tasks, "pending-connection", std::future::pending()),
1460 )
1461 .await
1462 .expect("task registration must not await connection completion");
1463
1464 assert_eq!(tasks.0.lock().await.len(), 1);
1465 tasks.shutdown().await;
1466 }
1467
1468 /// Constructing a gateway must never create or bind a UDP socket,
1469 /// even though construction succeeds.
1470 /// The socket is created lazily inside `start()`, reached only through an
1471 /// authorized [`DoipDiagGateway::enable`]
[docs]1472 /// [[ test~deferred-doip-passive-construction, Gateway construction does not bind its discovery socket, test ]]
1473 #[tokio::test]
1474 async fn new_never_binds_the_doip_socket() {
1475 let gateway = DoipDiagGateway::<TestEcu>::new(
1476 &DoipConfig::default(),
1477 Arc::new(HashMap::new()),
1478 VariantDetectionSender::new(mpsc::channel(1).0),
1479 Arc::new(TestConnectivityHandler),
1480 )
1481 .await
1482 .unwrap();
1483
1484 assert!(
1485 gateway.state.socket.lock().await.is_none(),
1486 "DoipDiagGateway::new must not bind a socket"
1487 );
1488 assert_eq!(
1489 cda_interfaces::communication_control::TransportControl::state(&gateway).await,
1490 TransportState::Disabled
1491 );
1492 }
1493
1494 fn transmission_params() -> TransmissionParameters {
1495 TransmissionParameters {
1496 gateway_address: GATEWAY_ADDR,
1497 timeout_ack: Duration::from_secs(1),
1498 ecu_name: "test-ecu".to_owned(),
1499 repeat_request_count_transmission: 0,
1500 }
1501 }
1502
1503 fn service_payload() -> ServicePayload {
1504 ServicePayload {
1505 data: REQUEST_DATA.to_vec(),
1506 source_address: TESTER_ADDR,
1507 target_address: ECU_ADDR,
1508 new_session: None,
1509 new_security: None,
1510 }
1511 }
1512
1513 fn ecu_msg_response() -> DiagnosticResponse {
1514 DiagnosticResponse::Msg {
1515 source_address: ECU_ADDR,
1516 target_address: TESTER_ADDR,
1517 data: RESPONSE_DATA.to_vec(),
1518 }
1519 }
1520
1521 fn diag_msg() -> DiagnosticMessage {
1522 DiagnosticMessage {
1523 source_address: TESTER_ADDR.to_be_bytes(),
1524 target_address: ECU_ADDR.to_be_bytes(),
1525 message: REQUEST_DATA.to_vec(),
1526 }
1527 }
1528
1529 type BroadcastTx = broadcast::Sender<Result<DiagnosticResponse, crate::connections::EcuError>>;
1530
1531 fn make_broadcast_pair() -> (
1532 BroadcastTx,
1533 broadcast::Receiver<Result<DiagnosticResponse, crate::connections::EcuError>>,
1534 ) {
1535 broadcast::channel(16)
1536 }
1537
1538 type ResponseChannel = (
1539 mpsc::Sender<Result<Option<TransportResponse>, DiagServiceError>>,
1540 mpsc::Receiver<Result<Option<TransportResponse>, DiagServiceError>>,
1541 );
1542
1543 fn make_response_channel() -> ResponseChannel {
1544 mpsc::channel(1)
1545 }
1546
1547 type SendResult = Result<tokio::task::JoinHandle<()>, cda_interfaces::DiagServiceError>;
1548 type ResponseItem = Result<Option<TransportResponse>, cda_interfaces::DiagServiceError>;
1549
1550 /// Shared test harness.
1551 ///
1552 /// Owns every channel handle that must stay alive for the duration of a
1553 /// test:
1554 /// - `broadcast_tx` - injects simulated ECU responses.
1555 /// - `_doip_payload_rx` - keeps the mpsc sender alive so `send_with_retries`
1556 /// does not error out trying to write the outgoing `DoIP` frame.
1557 /// - `send_handle` - join handle for the background `send()` task.
1558 /// - `response_rx` - receives the UDS responses forwarded by `send()`.
1559 struct TestHarness {
1560 broadcast_tx: broadcast::Sender<Result<DiagnosticResponse, crate::connections::EcuError>>,
1561 _doip_payload_rx: mpsc::Receiver<DoipPayload>,
1562 send_handle: tokio::task::JoinHandle<SendResult>,
1563 response_rx: mpsc::Receiver<ResponseItem>,
1564 }
1565
1566 impl TestHarness {
1567 /// Creates the gateway, spawns the `send()` task, and advances it past
1568 /// `clear_pending_messages` to the point where it is blocked on
1569 /// `ecu.receiver.recv()` inside the `'ack_waiting` loop.
1570 ///
1571 /// Uses `current_thread` flavor (see test attributes) so that
1572 /// `yield_now()` is a reliable synchronization primitive: on a
1573 /// cooperative single-threaded executor each yield hands control to
1574 /// exactly the next scheduled task. The spawned task must pass through
1575 /// the following `.await` points before reaching `recv()`:
1576 ///
1577 /// 1. `ecu_mtx.lock().await` - acquires the ECU mutex
1578 /// 2. `send_with_retries(...).await` - writes the outgoing `DoIP` frame
1579 /// 3. `tokio::time::timeout(...)` - enters the ACK-wait future
1580 /// 4. `ecu.receiver.recv().await` - now blocked, ready for injection
1581 ///
1582 /// Crucially, `clear_pending_messages` runs between steps 1 and 2.
1583 /// We must yield past it before calling `broadcast_tx.send()`, otherwise
1584 /// the injected message would be drained and the task would block forever.
1585 /// One `yield_now()` per await point, plus one safety margin = 5 total.
1586 async fn new() -> Self {
1587 let (broadcast_tx, broadcast_rx) =
1588 broadcast::channel::<Result<DiagnosticResponse, crate::connections::EcuError>>(1);
1589 let (doip_payload_tx, doip_payload_rx) = mpsc::channel::<DoipPayload>(1);
1590
1591 let gateway = make_gateway(broadcast_rx, doip_payload_tx);
1592
1593 let (response_tx, response_rx) = mpsc::channel::<ResponseItem>(1);
1594 let send_handle = tokio::spawn({
1595 let gateway = gateway.clone();
1596 async move {
1597 gateway
1598 .send(transmission_params(), service_payload(), response_tx, true)
1599 .await
1600 }
1601 });
1602
1603 for _ in 0..5 {
1604 tokio::task::yield_now().await;
1605 }
1606
1607 Self {
1608 broadcast_tx,
1609 _doip_payload_rx: doip_payload_rx,
1610 send_handle,
1611 response_rx,
1612 }
1613 }
1614
1615 /// Waits for the next response from the background `send()` task.
1616 ///
1617 /// Uses `tokio::time::timeout` rather than `try_recv()`, so no
1618 /// yield counting is needed after injecting a broadcast message.
1619 async fn recv_response(&mut self) -> ResponseItem {
1620 tokio::time::timeout(Duration::from_millis(500), self.response_rx.recv())
1621 .await
1622 .expect("timed out waiting for response from send() task")
1623 .expect("response channel closed unexpectedly")
1624 }
1625 }
1626
1627 // current_thread: yield_now() in TestHarness::new() must be a reliable sync point.
1628 #[tokio::test(flavor = "current_thread")]
1629 async fn implicit_ack_busy_retry_then_final_response() {
1630 let mut harness = TestHarness::new().await;
1631
1632 // ECU sends BusyRepeatRequest (NRC 0x21) via the implicit-ACK path:
1633 // the DiagnosticMessage arrives before the ACK, so
1634 // wait_for_ack_or_response_until_timeout returns it directly as the
1635 // seed for read_ecu_responses instead of waiting for a separate ACK.
1636 harness
1637 .broadcast_tx
1638 .send(Ok(DiagnosticResponse::Msg {
1639 source_address: ECU_ADDR,
1640 target_address: TESTER_ADDR,
1641 data: vec![
1642 service_ids::NEGATIVE_RESPONSE,
1643 REQUEST_DATA[0],
1644 nrc::BUSY_REPEAT_REQUEST,
1645 ],
1646 }))
1647 .expect("Failed to sent busy repeat request");
1648 let first = harness.recv_response().await;
1649 assert!(
1650 matches!(
1651 first,
1652 Ok(Some(TransportResponse::Pending(
1653 PendingNrc::BusyRepeatRequest { .. }
1654 )))
1655 ),
1656 "expected BusyRepeatRequest, got {first:?}"
1657 );
1658
1659 // ECU sends the final Msg.
1660 harness
1661 .broadcast_tx
1662 .send(Ok(ecu_msg_response()))
1663 .expect("Failed to sent final message");
1664 let second = harness.recv_response().await;
1665 assert!(
1666 matches!(second, Ok(Some(TransportResponse::UdsResponse(_)))),
1667 "expected Message, got {second:?}"
1668 );
1669
1670 drop(harness.response_rx);
1671 tokio::time::timeout(Duration::from_secs(2), harness.send_handle)
1672 .await
1673 .expect("send task did not finish")
1674 .expect("send task panicked")
1675 .expect("send() returned an error");
1676 }
1677
1678 // current_thread: yield_now() in TestHarness::new() must be a reliable sync point.
1679 #[tokio::test(flavor = "current_thread")]
1680 async fn implicit_ack_single_response() {
1681 let mut harness = TestHarness::new().await;
1682
1683 // ECU sends a single Msg - no ACK (implicit ACK path).
1684 harness
1685 .broadcast_tx
1686 .send(Ok(ecu_msg_response()))
1687 .expect("Failed to sent message");
1688 let response = harness.recv_response().await;
1689 assert!(
1690 matches!(response, Ok(Some(TransportResponse::UdsResponse(_)))),
1691 "expected Message, got {response:?}"
1692 );
1693
1694 // No further messages; dropping the receiver unblocks read_ecu_responses.
1695 drop(harness.response_rx);
1696 tokio::time::timeout(Duration::from_secs(2), harness.send_handle)
1697 .await
1698 .expect("send task did not finish")
1699 .expect("send task panicked")
1700 .expect("send() returned an error");
1701 }
1702 #[tokio::test]
1703 async fn wait_for_ack_or_response_ok_none_on_explicit_ack() {
1704 let (tx, mut rx) = make_broadcast_pair();
1705 let msg = diag_msg();
1706 tx.send(Ok(DiagnosticResponse::Ack((ECU_ADDR, vec![]))))
1707 .expect("Failed to send ack");
1708
1709 let result = wait_for_ack_or_response_until_timeout(
1710 &mut rx,
1711 "test-ecu",
1712 &msg,
1713 Duration::from_secs(1),
1714 )
1715 .await;
1716 assert!(
1717 matches!(result, Ok(None)),
1718 "expected Ok(None), got {result:?}"
1719 );
1720 }
1721
1722 /// A `DiagnosticMessage` received before the ACK triggers the implicit-ACK
1723 /// path: the response is returned as `Ok(Some(_))` for the caller to seed
1724 /// into `read_ecu_responses`.
1725 #[tokio::test]
1726 async fn wait_for_ack_or_response_ok_some_on_diagnostic_msg() {
1727 let (tx, mut rx) = make_broadcast_pair();
1728 let msg = diag_msg();
1729 tx.send(Ok(ecu_msg_response()))
1730 .expect("Failed to send message");
1731
1732 let result = wait_for_ack_or_response_until_timeout(
1733 &mut rx,
1734 "test-ecu",
1735 &msg,
1736 Duration::from_secs(1),
1737 )
1738 .await;
1739
1740 assert!(
1741 matches!(result, Ok(Some(DiagnosticResponse::Msg { .. }))),
1742 "expected Ok(Some(Diagnostic)), got {result:?}"
1743 );
1744 }
1745
1746 /// `BusyRepeatRequest` before the ACK also returns `Ok(Some(_))`.
1747 #[tokio::test]
1748 async fn wait_for_ack_or_response_ok_some_on_busy_repeat_request() {
1749 let (tx, mut rx) = make_broadcast_pair();
1750 let msg = diag_msg();
1751
1752 tx.send(Ok(DiagnosticResponse::Msg {
1753 source_address: ECU_ADDR,
1754 target_address: TESTER_ADDR,
1755 data: vec![
1756 service_ids::NEGATIVE_RESPONSE,
1757 REQUEST_DATA[0],
1758 nrc::BUSY_REPEAT_REQUEST,
1759 ],
1760 }))
1761 .expect("Failed to busy repeat request");
1762
1763 let result = wait_for_ack_or_response_until_timeout(
1764 &mut rx,
1765 "test-ecu",
1766 &msg,
1767 Duration::from_secs(1),
1768 )
1769 .await;
1770
1771 assert!(
1772 matches!(result, Ok(Some(DiagnosticResponse::Msg { .. }))),
1773 "expected Ok(Some(Diagnostic(BusyRepeatRequest NRC))), got {result:?}"
1774 );
1775 }
1776
1777 /// A `DiagnosticMessageNack` returns `Err(Nack(_))`.
1778 #[tokio::test]
1779 async fn wait_for_ack_or_response_err_on_nack() {
1780 use doip_definitions::payload::{DiagnosticMessageNack, DiagnosticNackCode};
1781
1782 let (tx, mut rx) = make_broadcast_pair();
1783 let msg = diag_msg();
1784
1785 tx.send(Ok(DiagnosticResponse::Nack(DiagnosticMessageNack {
1786 source_address: ECU_ADDR.to_be_bytes(),
1787 target_address: TESTER_ADDR.to_be_bytes(),
1788 nack_code: DiagnosticNackCode::UnknownTargetAddress,
1789 })))
1790 .expect("Failed to send nack");
1791
1792 let result = wait_for_ack_or_response_until_timeout(
1793 &mut rx,
1794 "test-ecu",
1795 &msg,
1796 Duration::from_secs(1),
1797 )
1798 .await;
1799
1800 assert!(
1801 matches!(result, Err(DiagServiceError::Nack(_))),
1802 "expected Nack, got {result:?}"
1803 );
1804 }
1805
1806 /// A `GenericNack` returns `Err(Nack(_))`.
1807 #[tokio::test]
1808 async fn wait_for_ack_or_response_err_on_generic_nack() {
1809 use doip_definitions::payload::{GenericNack, NackCode};
1810
1811 let (tx, mut rx) = make_broadcast_pair();
1812 let msg = diag_msg();
1813
1814 tx.send(Ok(DiagnosticResponse::GenericNack(GenericNack {
1815 nack_code: NackCode::InvalidPayloadLength,
1816 })))
1817 .expect("Failed to send generic nack");
1818
1819 let result = wait_for_ack_or_response_until_timeout(
1820 &mut rx,
1821 "test-ecu",
1822 &msg,
1823 Duration::from_secs(1),
1824 )
1825 .await;
1826
1827 assert!(
1828 matches!(result, Err(DiagServiceError::Nack(_))),
1829 "expected Nack, got {result:?}"
1830 );
1831 }
1832
1833 /// When the timeout expires with no message, `Err(Timeout)` is returned.
1834 #[tokio::test]
1835 async fn wait_for_ack_or_response_err_on_timeout() {
1836 let (_tx, mut rx) = make_broadcast_pair();
1837 let msg = diag_msg();
1838
1839 let result = wait_for_ack_or_response_until_timeout(
1840 &mut rx,
1841 "test-ecu",
1842 &msg,
1843 Duration::from_millis(200),
1844 )
1845 .await;
1846
1847 assert!(
1848 matches!(result, Err(DiagServiceError::Timeout)),
1849 "expected Timeout, got {result:?}"
1850 );
1851 }
1852
1853 /// When the broadcast receiver is closed, `Err(NoResponse(_))` is returned.
1854 #[tokio::test]
1855 async fn wait_for_ack_or_response_err_when_receiver_closed() {
1856 let (tx, mut rx) = make_broadcast_pair();
1857 let msg = diag_msg();
1858
1859 drop(tx);
1860
1861 let result = wait_for_ack_or_response_until_timeout(
1862 &mut rx,
1863 "test-ecu",
1864 &msg,
1865 Duration::from_secs(1),
1866 )
1867 .await;
1868
1869 assert!(
1870 matches!(result, Err(DiagServiceError::NoResponse(_))),
1871 "expected NoResponse, got {result:?}"
1872 );
1873 }
1874
1875 /// An unrelated message type (e.g. `AliveCheckResponse`) must be skipped;
1876 /// the loop should continue until the real ACK arrives.
1877 #[tokio::test]
1878 async fn wait_for_ack_or_response_skips_unrelated_messages() {
1879 let (tx, mut rx) = make_broadcast_pair();
1880 let msg = diag_msg();
1881
1882 tx.send(Ok(DiagnosticResponse::AliveCheckResponse))
1883 .expect("Failed to send alive check response");
1884 tx.send(Ok(DiagnosticResponse::Ack((ECU_ADDR, vec![]))))
1885 .expect("Failed to send ack");
1886
1887 let result = wait_for_ack_or_response_until_timeout(
1888 &mut rx,
1889 "test-ecu",
1890 &msg,
1891 Duration::from_secs(1),
1892 )
1893 .await;
1894
1895 drop(tx);
1896 assert!(
1897 matches!(result, Ok(None)),
1898 "expected Ok(None), got {result:?}"
1899 );
1900 }
1901
1902 #[tokio::test]
1903 async fn read_ecu_responses_forwards_message_and_exits_on_sender_close() {
1904 let (tx, mut rx) = make_broadcast_pair();
1905 let (resp_tx, mut resp_rx) = make_response_channel();
1906
1907 tx.send(Ok(ecu_msg_response()))
1908 .expect("Failed to send message");
1909
1910 // Spawn the response reader in a separate task.
1911 let handle = tokio::spawn(async move {
1912 read_ecu_responses(&mut rx, "test-ecu", &resp_tx, None).await;
1913 });
1914
1915 // Read the forwarded Msg.
1916 let item = tokio::time::timeout(Duration::from_secs(1), resp_rx.recv())
1917 .await
1918 .expect("timeout")
1919 .expect("channel closed");
1920 assert!(
1921 matches!(item, Ok(Some(TransportResponse::UdsResponse(_)))),
1922 "expected Message, got {item:?}"
1923 );
1924
1925 // Drop the receiver to close the sender; the loop should exit.
1926 drop(resp_rx);
1927 tokio::time::timeout(Duration::from_secs(2), handle)
1928 .await
1929 .expect("read_ecu_responses did not exit after sender closed")
1930 .expect("task panicked");
1931 }
1932
1933 #[tokio::test]
1934 async fn read_ecu_responses_forwards_multiple_responses() {
1935 let (tx, mut rx) = make_broadcast_pair();
1936 let (resp_tx, mut resp_rx) = make_response_channel();
1937 tx.send(Ok(DiagnosticResponse::Msg {
1938 source_address: ECU_ADDR,
1939 target_address: TESTER_ADDR,
1940 data: vec![
1941 service_ids::NEGATIVE_RESPONSE,
1942 REQUEST_DATA[0],
1943 nrc::RESPONSE_PENDING,
1944 ],
1945 }))
1946 .expect("Failed to send pending response");
1947 tx.send(Ok(ecu_msg_response()))
1948 .expect("Failed to send response");
1949 // Keep tx alive across awaits so the channel is not closed while the
1950 // spawned task is reading buffered messages.
1951 let _tx = tx;
1952
1953 let handle = tokio::spawn(async move {
1954 read_ecu_responses(&mut rx, "test-ecu", &resp_tx, None).await;
1955 });
1956
1957 let first = tokio::time::timeout(Duration::from_secs(1), resp_rx.recv())
1958 .await
1959 .expect("timeout")
1960 .expect("closed");
1961 assert!(
1962 matches!(
1963 first,
1964 Ok(Some(TransportResponse::Pending(
1965 PendingNrc::ResponsePending { .. }
1966 )))
1967 ),
1968 "expected ResponsePending, got {first:?}"
1969 );
1970
1971 let second = tokio::time::timeout(Duration::from_secs(1), resp_rx.recv())
1972 .await
1973 .expect("timeout")
1974 .expect("closed");
1975 assert!(
1976 matches!(second, Ok(Some(TransportResponse::UdsResponse(_)))),
1977 "expected Message, got {second:?}"
1978 );
1979
1980 drop(resp_rx);
1981 tokio::time::timeout(Duration::from_secs(2), handle)
1982 .await
1983 .expect("task did not exit")
1984 .expect("task panicked");
1985 }
1986
1987 /// When `first` is `Some`, it is forwarded before reading from the receiver -
1988 /// this is the implicit-ACK seed path.
1989 #[tokio::test]
1990 async fn read_ecu_responses_forwards_first_then_receiver() {
1991 let (tx, mut rx) = make_broadcast_pair();
1992 let (resp_tx, mut resp_rx) = make_response_channel();
1993
1994 let first = ecu_msg_response();
1995 tx.send(Ok(DiagnosticResponse::Msg {
1996 source_address: ECU_ADDR,
1997 target_address: TESTER_ADDR,
1998 data: vec![
1999 service_ids::NEGATIVE_RESPONSE,
2000 REQUEST_DATA[0],
2001 nrc::RESPONSE_PENDING,
2002 ],
2003 }))
2004 .expect("Failed to send pending response");
2005 let _tx = tx;
2006
2007 let handle = tokio::spawn(async move {
2008 read_ecu_responses(&mut rx, "test-ecu", &resp_tx, Some(first)).await;
2009 });
2010
2011 let first_item = tokio::time::timeout(Duration::from_secs(1), resp_rx.recv())
2012 .await
2013 .expect("timeout")
2014 .expect("closed");
2015 assert!(
2016 matches!(first_item, Ok(Some(TransportResponse::UdsResponse(_)))),
2017 "expected Message from first, got {first_item:?}"
2018 );
2019
2020 let second_item = tokio::time::timeout(Duration::from_secs(1), resp_rx.recv())
2021 .await
2022 .expect("timeout")
2023 .expect("closed");
2024 assert!(
2025 matches!(
2026 second_item,
2027 Ok(Some(TransportResponse::Pending(
2028 PendingNrc::ResponsePending { .. }
2029 )))
2030 ),
2031 "expected ResponsePending from receiver, got {second_item:?}"
2032 );
2033
2034 drop(resp_rx);
2035 tokio::time::timeout(Duration::from_secs(2), handle)
2036 .await
2037 .expect("task did not exit")
2038 .expect("task panicked");
2039 }
2040}
Back to top