# Automotive Middleware > Expert in AMQP (Advanced Message Queuing Protocol) middleware for automotive enterprise integration using RabbitMQ and Azure Service Bus. Covers 6 topics across middleware domain. Includes 6 skill files covering AMQP 1.0 OASIS Standard, AUTOSAR Adaptive (comparison), AUTOSAR Adaptive Platform, AWS IoT Core best practices, Apache Qpid Proton, AutomationML for data modeling, Azure IoT Hub protocols, Azure Service Bus protocols and more. - Skill: `pangzhenying2025/automotive-middleware` (Agent Skill) - Install (CLI): `npx skillmds@latest add pangzhenying2025/automotive-middleware` - Raw SKILL.md: https://api.skillmd.com/api/skills/pangzhenying2025/automotive-middleware/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - Author: pangzhenying2025 (https://skillmd.com/u/pangzhenying2025) - Updated: 2026-09-22 - Page: https://skillmd.com/skills/pangzhenying2025/automotive-middleware --- # Automotive Middleware 6 skill files covering middleware domain for automotive software engineering. ## Applicable Standards - AMQP 1.0 OASIS Standard - AUTOSAR Adaptive (comparison) - AUTOSAR Adaptive Platform - AWS IoT Core best practices - Apache Qpid Proton - AutomationML for data modeling - Azure IoT Hub protocols - Azure Service Bus protocols - DDS 1.4 Specification - DDS Security 1.1 - DDS-RTPS 2.5 - DTLS 1.2 (Security) - GDPR compliance for telemetry data - IEC 62541 (OPC UA Standard) - ISO 21434 (Cybersecurity) - ISO 26262 (Functional Safety for ADAS) - ISO 26262 (Functional Safety) - ISO 26262 (when interfacing with vehicle systems) - MQTT 5.0 (comparison context) - MQTT 5.0 Specification - MQTT-OPC UA Gateway (Sparkplug B) - RFC 7252 (CoAP) - RFC 7959 (CoAP Block-Wise Transfer) - RFC 8132 (CoAP PATCH/FETCH) - RFC 8323 (CoAP over TCP/TLS/WebSockets) - ROS 2 Design Principles - RTPS 2.5 Protocol - TISAX (Trusted Information Security Assessment) ## Use Cases - Factory-to-vehicle data exchange - Manufacturing execution system (MES) integration - Order-to-delivery workflow orchestration - Parts inventory and supply chain messaging - Quality assurance data pipelines - Dealer network communication - Low-power ECU communication - Battery monitoring telemetry (minimal overhead) - Wireless sensor networks in vehicle - Tire pressure monitoring systems (TPMS) - V2X communication over 6LoWPAN - Edge device telemetry with cellular IoT (NB-IoT, LTE-M) - ADAS sensor data distribution with deterministic latency - V2X communication with QoS guarantees - Real-time ECU-to-ECU communication - Safety-critical data exchange (ISO 26262 ASIL-D) - Multi-domain vehicle networking - Edge computing data streams - Vehicle-to-Cloud telemetry streaming - Remote diagnostics and OTA updates ## Topics Covered ### Enterprise Messaging - amqp-middleware ### Industrial Integration - opcua-middleware ### Iot Constrained - coap-middleware ### Iot Telemetry - mqtt-middleware ### Real Time Pub Sub - dds-middleware ### Robotics Autonomy - ros2-dds-middleware ## Constraints - 10 msg/sec per vehicle - 10,000 messages - 128KB per message - 256KB (RabbitMQ), 1MB (Azure) - 60 seconds keep-alive - CON messages for commands and configuration - Client certificates required for production - DDS Security mandatory for production deployment - DDS Security mandatory for production vehicles - DTLS 1.2+ required for production - Dead letter queues for all queues - Deterministic latency < 10ms p99 for safety topics - Discovery time < 2 seconds for static discovery - ISO 26262 ASIL-D capable implementations only - Max 10,000 nodes per server (scalability) ## Required Tools - AWS IoT Device SDK - Azure IoT SDK - Azure Service Bus SDK - CoAP.NET (C#) - Copper (Firefox/Chrome plugin) - DDS Monitoring tools (RTI Admin Console / Fast DDS Monitor) - Fast DDS or Cyclone DDS - Gazebo or CARLA simulator - IDL compiler (rtiddsgen / fastddsgen / opendds_idl) - JMeter with MQTT plugin (load testing) - MQTT Explorer (GUI client) - Mosquitto broker - Network simulator (NetEm, WANem) for testing - OPC UA Compliance Test Tool - OpenDDS (opensource) ## Instructions ### amqp-middleware ## AMQP Middleware Expertise You are an expert in AMQP middleware for automotive enterprise systems and factory integration. ### Core Protocol 1. **AMQP Architecture** - Producer: Application publishing messages - Exchange: Routing hub (direct, topic, fanout, headers) - Queue: Message buffer with persistence - Consumer: Application consuming messages - Binding: Route from exchange to queue - Virtual Host: Logical separation (dev/prod) 2. **Message Properties** - Content-Type: application/json, application/protobuf - Delivery-Mode: 1 (transient), 2 (persistent) - Priority: 0-9 (higher = priority queue) - Correlation-ID: Request/response tracking - Reply-To: Return queue for RPC - Expiration: TTL in milliseconds - Message-ID: Unique identifier 3. **Exchange Types** - **Direct**: Routing key exact match ``` Exchange: vehicle.commands Binding: remote_lock → queue.remote_lock Message routing_key: remote_lock → delivered to queue.remote_lock ``` - **Topic**: Wildcard routing (* = one word, # = zero or more) ``` Exchange: vehicle.telemetry Binding: vehicle.*.battery → queue.battery_all Binding: vehicle.tesla.# → queue.tesla_fleet ``` - **Fanout**: Broadcast to all bound queues ``` Exchange: ota.broadcast All queues bound to exchange receive message ``` - **Headers**: Route by message headers (rare) 4. **Quality of Service** - **Publisher Confirms**: Ack from broker when message persisted - **Consumer Acks**: Manual/auto acknowledgment - **Transactions**: Multi-message atomic commit - **Dead Letter Exchange (DLX)**: Route failed messages - **TTL + Max Length**: Queue resource limits ### Automotive Use Cases 1. **Manufacturing Line Integration** - **Scenario**: Robot arm completes battery install → notify next station - **Pattern**: Direct exchange ``` Producer: Robot PLC Exchange: factory.station (direct) Routing Key: station.battery_install.complete Queue: station.quality_check Consumer: QA workstation ``` 2. **Vehicle Configuration Distribution** - **Scenario**: Customer orders custom vehicle → publish to MES - **Pattern**: Topic exchange ``` Producer: Order management system Exchange: vehicle.config (topic) Message: {"vin": "...", "trim": "premium", "color": "blue"} Routing Key: vehicle.model_s.premium Bindings: - vehicle.model_s.* → queue.paint_shop - vehicle.*.premium → queue.interior_line ``` 3. **OTA Update Orchestration** - **Scenario**: Release firmware to 1M vehicles in batches - **Pattern**: Fanout + direct ``` Exchange: ota.release (fanout) Queues: ota.batch_1, ota.batch_2, ..., ota.batch_100 Each queue has 10,000 vehicle IDs Workers consume from queues at controlled rate ``` 4. **Supply Chain Events** - **Scenario**: Battery supplier ships cells → update inventory - **Pattern**: Topic exchange with dead letter ``` Exchange: supply.events (topic) Routing Key: supply.battery.LG.shipped Queue: inventory.battery (TTL=48h, DLX for unprocessed) Consumer: ERP system ``` ### Implementation Patterns 1. **RabbitMQ Publisher (Python)** ```python import pika import json def publish_vehicle_config(vin, config): credentials = pika.PlainCredentials('vehicle_app', 'secure_password') parameters = pika.ConnectionParameters( host='rabbitmq.factory.local', port=5672, virtual_host='/production', credentials=credentials, heartbeat=600, blocked_connection_timeout=300 ) connection = pika.BlockingConnection(parameters) channel = connection.channel() # Declare exchange (idempotent) channel.exchange_declare( exchange='vehicle.config', exchange_type='topic', durable=True ) routing_key = f"vehicle.{config['model']}.{config['trim']}" message = json.dumps({ "vin": vin, "config": config, "timestamp": datetime.utcnow().isoformat() }) # Publish with persistence channel.basic_publish( exchange='vehicle.config', routing_key=routing_key, body=message, properties=pika.BasicProperties( delivery_mode=2, # Persistent content_type='application/json', correlation_id=str(uuid.uuid4()) ) ) connection.close() ``` 2. **RabbitMQ Consumer (Python)** ```python def process_message(ch, method, properties, body): try: data = json.loads(body) vin = data["vin"] config = data["config"] # Process configuration apply_vehicle_config(vin, config) # Manual acknowledgment after successful processing ch.basic_ack(delivery_tag=method.delivery_tag) except Exception as e: print(f"Error processing message: {e}") # Reject and requeue (retry) ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True) def start_consumer(): connection = pika.BlockingConnection(parameters) channel = connection.channel() # Declare queue with DLX channel.queue_declare( queue='queue.paint_shop', durable=True, arguments={ 'x-dead-letter-exchange': 'dlx.vehicle.config', 'x-message-ttl': 86400000, # 24 hours 'x-max-length': 10000 } ) # Bind to exchange channel.queue_bind( exchange='vehicle.config', queue='queue.paint_shop', routing_key='vehicle.model_s.*' ) # Set QoS: prefetch 10 messages channel.basic_qos(prefetch_count=10) # Start consuming channel.basic_consume( queue='queue.paint_shop', on_message_callback=process_message, auto_ack=False # Manual ack ) print("Waiting for messages...") channel.start_consuming() ``` 3. **Publisher Confirms** ```python def publish_with_confirm(channel, exchange, routing_key, message): # Enable publisher confirms channel.confirm_delivery() try: channel.basic_publish( exchange=exchange, routing_key=routing_key, body=message, properties=pika.BasicProperties(delivery_mode=2), mandatory=True # Return if unroutable ) print("Message confirmed by broker") except pika.exceptions.UnroutableError: print("Message was returned (no queue bound)") except pika.exceptions.NackError: print("Message was nacked by broker") ``` 4. **RPC Pattern (Request/Response)** ```python class VehicleRPCClient: def __init__(self): self.connection = pika.BlockingConnection(parameters) self.channel = self.connection.channel() # Exclusive queue for responses result = self.channel.queue_declare(queue='', exclusive=True) self.callback_queue = result.method.queue self.channel.basic_consume( queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True ) self.response = None self.corr_id = None def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body def call(self, vin, command): self.response = None self.corr_id = str(uuid.uuid4()) self.channel.basic_publish( exchange='', routing_key='rpc.vehicle.commands', properties=pika.BasicProperties( reply_to=self.callback_queue, correlation_id=self.corr_id, ), body=json.dumps({"vin": vin, "command": command}) ) # Wait for response (blocking) while self.response is None: self.connection.process_data_events() return json.loads(self.response) # Usage rpc = VehicleRPCClient() result = rpc.call("1HGCM82633A004352", "get_dtc_codes") ``` ### Azure Service Bus (AMQP 1.0) 1. **Queue vs Topic** - **Queue**: Point-to-point (single consumer) - **Topic + Subscriptions**: Pub/sub (multiple consumers) 2. **Python Client** ```python from azure.servicebus import ServiceBusClient, ServiceBusMessage connection_str = "Endpoint=sb://vehicle-namespace.servicebus.windows.net/;..." client = ServiceBusClient.from_connection_string(connection_str) # Send to queue def send_message(queue_name, message_dict): sender = client.get_queue_sender(queue_name) message = ServiceBusMessage( json.dumps(message_dict), content_type="application/json", correlation_id=str(uuid.uuid4()), session_id="vehicle_12345" # Session for ordering ) sender.send_messages(message) sender.close() # Receive from queue def receive_messages(queue_name): receiver = client.get_queue_receiver(queue_name) messages = receiver.receive_messages(max_message_count=10, max_wait_time=5) for msg in messages: data = json.loads(str(msg)) process_vehicle_event(data) receiver.complete_message(msg) # Ack receiver.close() ``` 3. **Topic Subscriptions with Filters** ```python from azure.servicebus.management import ServiceBusAdministrationClient admin_client = ServiceBusAdministrationClient.from_connection_string(connection_str) # Create topic admin_client.create_topic("vehicle-telemetry") # Create subscription with SQL filter admin_client.create_subscription( topic_name="vehicle-telemetry", subscription_name="high-priority-vehicles", rule=CorrelationRuleFilter( sql_filter="priority = 'high' AND region = 'US'" ) ) ``` ### Advanced Patterns 1. **Priority Queues** - RabbitMQ: x-max-priority=10 - Higher priority messages consumed first 2. **Delayed Messages** - RabbitMQ plugin: x-delayed-message exchange - Publish with x-delay header (ms) 3. **Message Deduplication** - Azure Service Bus: Automatic by message_id - RabbitMQ: Application-level (Redis cache) 4. **Dead Letter Handling** ```python def process_dead_letters(): receiver = client.get_queue_receiver("queue.paint_shop/$deadletterqueue") messages = receiver.receive_messages(max_message_count=100) for msg in messages: print(f"DLQ Reason: {msg.dead_letter_reason}") print(f"DLQ Description: {msg.dead_letter_error_description}") # Log to monitoring system log_dead_letter(msg) receiver.complete_message(msg) ``` ### Monitoring & Operations 1. **RabbitMQ Management** - Web UI: http://localhost:15672 - Metrics: Queue depth, publish rate, consumer count - Health check: /api/healthchecks/node 2. **Azure Service Bus Metrics** - Active messages - Dead letter messages - Throttled requests - CPU/memory of namespace 3. **Alerting** - Queue depth > 10,000 → scale consumers - DLQ depth > 100 → investigate failures - Connection errors → network issues ### Security 1. **Authentication** - RabbitMQ: Username/password, LDAP, OAuth2 - Azure Service Bus: Shared Access Signature (SAS), Azure AD 2. **Authorization** - RabbitMQ: Permissions per virtual host (read/write/configure) - Azure Service Bus: RBAC roles (Sender, Receiver, Owner) 3. **Encryption** - TLS 1.2+ for AMQP connections - Azure Service Bus: Encryption at rest (automatic) ### Performance Tuning 1. **Connection Pooling** - Reuse connections across threads - RabbitMQ: 1 connection per application, N channels 2. **Batch Publishing** - Send 100 messages in single network round-trip - Use transactions or publisher confirms 3. **Consumer Scaling** - Horizontal: Multiple consumers on same queue - Vertical: Increase prefetch_count (10-50) ### Testing Strategies 1. **Unit Tests** - Mock pika.BlockingConnection - Test message serialization 2. **Integration Tests** - Run RabbitMQ in Docker - Test end-to-end message flow 3. **Load Tests** - Publish 10,000 msg/sec - Measure latency p95, p99 ### Deliverables When implementing AMQP solutions, provide: 1. Exchange and queue topology diagram 2. Routing key conventions 3. Python/Java implementation with error handling 4. Dead letter queue processing logic 5. Monitoring dashboard configuration 6. Performance test results ### coap-middleware ## CoAP Middleware Expertise You are an expert in CoAP for automotive IoT and resource-constrained embedded systems. ### Core Protocol 1. **CoAP vs HTTP** - CoAP: UDP-based, 4-byte header, binary, optimized for IoT - HTTP: TCP-based, text headers, verbose - CoAP uses REST semantics (GET, POST, PUT, DELETE) - Runs over UDP (default) or TCP/DTLS/WebSockets 2. **Message Types** - **CON (Confirmable)**: Requires ACK (reliable) - **NON (Non-confirmable)**: Fire-and-forget (unreliable) - **ACK (Acknowledgment)**: Response to CON - **RST (Reset)**: Reject invalid message 3. **Request Methods** - GET: Retrieve resource (sensor reading) - POST: Create resource (submit telemetry) - PUT: Update resource (configure ECU) - DELETE: Remove resource - FETCH: Retrieve partial resource - PATCH: Partial update 4. **Response Codes** - 2.01 Created - 2.02 Deleted - 2.03 Valid - 2.04 Changed - 2.05 Content - 4.00 Bad Request - 4.04 Not Found - 5.00 Internal Server Error 5. **URI Structure** ``` coap://192.168.1.10:5683/vehicle/battery/soc coaps://ecu.vehicle.local:5684/sensors/temperature ``` ### Automotive Use Cases 1. **Tire Pressure Monitoring (TPMS)** - Each tire has wireless sensor (BLE + CoAP) - Publishes pressure, temperature to gateway ECU - NON messages @ 10-second intervals (battery saving) - Gateway aggregates and sends to CAN bus 2. **Battery Cell Monitoring** - 96 cell monitoring ICs → CoAP NON to BMS - Voltage, temperature per cell - Total payload: ~200 bytes - 1 Hz sampling, UDP multicast 3. **V2X over 6LoWPAN** - IPv6 over low-power wireless (802.15.4) - CoAP for CAM (Cooperative Awareness Messages) - Header compression (6LoWPAN HC) - < 100 bytes per message 4. **Cellular IoT Telemetry (NB-IoT)** - Vehicle publishes to cloud via NB-IoT - CoAP over UDP (lower overhead than MQTT) - Battery-optimized (PSM/eDRX modes) - ~50 bytes CoAP vs ~150 bytes MQTT CONNECT ### Implementation Patterns 1. **CoAP Server (Python - aiocoap)** ```python import asyncio import aiocoap import aiocoap.resource as resource class BatterySOCResource(resource.Resource): """GET /battery/soc - Return battery state of charge""" async def render_get(self, request): soc = read_battery_soc() # From CAN bus payload = f'{{"soc": {soc}, "unit": "percent"}}'.encode('utf-8') return aiocoap.Message( code=aiocoap.Code.CONTENT, payload=payload, content_format=aiocoap.numbers.ContentFormat.JSON ) class BatteryCommandResource(resource.Resource): """POST /battery/command - Execute battery command""" async def render_post(self, request): command = request.payload.decode('utf-8') result = execute_battery_command(command) return aiocoap.Message( code=aiocoap.Code.CHANGED if result else aiocoap.Code.INTERNAL_SERVER_ERROR ) def main(): root = resource.Site() root.add_resource(['battery', 'soc'], BatterySOCResource()) root.add_resource(['battery', 'command'], BatteryCommandResource()) asyncio.Task(aiocoap.Context.create_server_context(root, bind=('0.0.0.0', 5683))) asyncio.get_event_loop().run_forever() if __name__ == '__main__': main() ``` 2. **CoAP Client (Python)** ```python import asyncio from aiocoap import Context, Message, GET, POST async def fetch_battery_soc(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://192.168.1.10/battery/soc') response = await protocol.request(request).response if response.code.is_successful(): print(f"SOC: {response.payload.decode('utf-8')}") else: print(f"Error: {response.code}") async def send_telemetry(data): protocol = await Context.create_client_context() payload = json.dumps(data).encode('utf-8') request = Message( code=POST, uri='coap://cloud.example.com/telemetry', payload=payload ) # CON message for reliability request.mtype = aiocoap.CON response = await protocol.request(request).response return response.code.is_successful() asyncio.run(fetch_battery_soc()) ``` 3. **CoAP Server (C - libcoap)** ```c #include static void battery_soc_handler( coap_resource_t *resource, coap_session_t *session, const coap_pdu_t *request, const coap_string_t *query, coap_pdu_t *response ) { uint8_t soc = read_battery_soc(); char payload[64]; snprintf(payload, sizeof(payload), "{\"soc\": %d}", soc); coap_pdu_set_code(response, COAP_RESPONSE_CODE_CONTENT); coap_add_data(response, strlen(payload), (uint8_t*)payload); } int main() { coap_context_t *ctx = coap_new_context(NULL); coap_address_t addr; coap_address_init(&addr); addr.addr.sin.sin_family = AF_INET; addr.addr.sin.sin_port = htons(5683); coap_endpoint_t *ep = coap_new_endpoint(ctx, &addr, COAP_PROTO_UDP); coap_resource_t *resource = coap_resource_init( coap_make_str_const("battery/soc"), 0 ); coap_register_handler(resource, COAP_REQUEST_GET, battery_soc_handler); coap_add_resource(ctx, resource); while (1) { coap_io_process(ctx, COAP_IO_WAIT); } return 0; } ``` 4. **Observe (Publish/Subscribe)** ```python # Server: Observable resource class BatterySOCObservable(resource.ObservableResource): def __init__(self): super().__init__() self.soc = 100 asyncio.create_task(self.update_soc()) async def update_soc(self): while True: await asyncio.sleep(1) self.soc = read_battery_soc() self.updated_state() # Notify observers async def render_get(self, request): payload = f'{{"soc": {self.soc}}}'.encode('utf-8') return aiocoap.Message(code=aiocoap.Code.CONTENT, payload=payload) # Client: Observe resource async def observe_battery(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://192.168.1.10/battery/soc', observe=0) observation = protocol.request(request) async for response in observation.observation: print(f"SOC updated: {response.payload.decode('utf-8')}") ``` ### Advanced Features 1. **Block-Wise Transfer (Large Payloads)** - CoAP has 1280-byte MTU limit - Block-wise splits into chunks - Automatic with aiocoap - Example: Firmware OTA (1 MB file) ```python async def download_firmware(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://ota.example.com/firmware.bin') # Block-wise transfer handled automatically response = await protocol.request(request).response with open('firmware.bin', 'wb') as f: f.write(response.payload) ``` 2. **Multicast Discovery** ```python # Client: Discover all CoAP devices on network async def discover_devices(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://224.0.1.187/.well-known/core') response = await protocol.request(request).response print(f"Available resources: {response.payload.decode('utf-8')}") ``` 3. **Resource Directory** - Central registry for CoAP devices - Devices register their endpoints - Clients query directory ``` # Device registers POST coap://directory.local/rd Payload: ;rt="temperature";if="sensor" # Client discovers GET coap://directory.local/rd-lookup/res?rt=temperature ``` ### Security (DTLS) 1. **CoAPS (CoAP over DTLS)** - Port 5684 (default) - PSK (Pre-Shared Key) or PKI (Certificates) - Protects against eavesdropping, replay attacks 2. **PSK Mode (Python)** ```python from aiocoap import Context, Message, GET from aiocoap.credentials import CredentialsMap async def secure_request(): credentials = CredentialsMap() credentials.add_credential( 'coaps://192.168.1.10/*', {'psk': b'secret_key', 'client-identity': b'vehicle_12345'} ) protocol = await Context.create_client_context(credentials=credentials) request = Message(code=GET, uri='coaps://192.168.1.10/battery/soc') response = await protocol.request(request).response ``` 3. **Certificate Mode (C - libcoap)** ```c coap_dtls_pki_t dtls_pki; memset(&dtls_pki, 0, sizeof(dtls_pki)); dtls_pki.version = COAP_DTLS_PKI_SETUP_VERSION; dtls_pki.pki_key.key_type = COAP_PKI_KEY_PEM; dtls_pki.pki_key.key.pem.ca_file = "ca.pem"; dtls_pki.pki_key.key.pem.public_cert = "client.crt"; dtls_pki.pki_key.key.pem.private_key = "client.key"; coap_context_set_pki(ctx, &dtls_pki); ``` ### Performance Optimization 1. **NON Messages for Telemetry** - No ACK required → reduce latency - Trade reliability for speed - Use for non-critical data (speed, RPM) 2. **Token Reuse** - 4-byte token identifies request/response pair - Reuse tokens to reduce overhead 3. **CBOR Encoding** - More compact than JSON - libcbor or cbor2 (Python) ```python import cbor2 payload = cbor2.dumps({"soc": 85, "voltage": 400.5}) request = Message( code=POST, uri='coap://cloud.example.com/telemetry', payload=payload, content_format=aiocoap.numbers.ContentFormat.CBOR ) ``` 4. **Connection Reuse (CoAP over TCP)** - Avoid DTLS handshake per message - WebSocket transport for browser clients ### Comparison with MQTT | Feature | CoAP | MQTT | |---------|------|------| | Protocol | UDP/DTLS | TCP/TLS | | Header | 4 bytes | 2+ bytes | | Pub/Sub | Observe | Native | | QoS | CON/NON | 0/1/2 | | Broker | Optional | Required | | Use Case | Embedded, M2M | Cloud, IoT | | Power | Lower | Higher (TCP) | ### Monitoring & Debugging 1. **Wireshark** - CoAP dissector built-in - Filter: `coap` - Inspect messages, tokens, options 2. **Copper (Browser Plugin)** - Firefox plugin for CoAP - GUI for testing CoAP servers 3. **Logging** ```python import logging logging.basicConfig(level=logging.DEBUG) logging.getLogger('coap').setLevel(logging.DEBUG) ``` ### Testing Strategies 1. **Unit Tests** - Mock CoAP requests/responses - Test resource handlers 2. **Integration Tests** - Run CoAP server locally - Client sends requests, asserts responses 3. **Load Tests** - CoAP-bench tool - Measure requests/sec, latency ### Edge Cases 1. **Packet Loss**: CON with retransmission 2. **Duplicate Detection**: Message ID + token 3. **Congestion Control**: Exponential backoff 4. **Multicast**: Handle multiple responses ### Deliverables When implementing CoAP solutions, provide: 1. URI structure documentation 2. Resource handler implementations (GET/POST/PUT/DELETE) 3. DTLS/PSK configuration 4. Block-wise transfer for large payloads 5. Observe pattern for real-time updates 6. Performance test results (latency, throughput, battery impact) 7. Integration guide for CAN/DDS bridge ### dds-middleware ## DDS Middleware Expertise You are an expert in Data Distribution Service (DDS) middleware for automotive applications. ### Core Architecture 1. **DDS Domain Model** - Domain Participant: Entry point to DDS - Publisher/Subscriber: Data flow direction - DataWriter/DataReader: Message endpoints - Topic: Named data channel - DomainID: Logical network segmentation 2. **QoS Policies (23 Standard Policies)** - **Reliability**: RELIABLE vs BEST_EFFORT - RELIABLE: Guaranteed delivery with acknowledgments - BEST_EFFORT: UDP-like, no retransmissions - **Durability**: VOLATILE, TRANSIENT_LOCAL, TRANSIENT, PERSISTENT - TRANSIENT_LOCAL: Late joiners get historical data - **History**: KEEP_LAST(n), KEEP_ALL - **Deadline**: Max time between samples - **Liveliness**: AUTOMATIC, MANUAL_BY_PARTICIPANT, MANUAL_BY_TOPIC - **Ownership**: SHARED vs EXCLUSIVE (for redundancy) - **TimeBasedFilter**: Throttle data rate at subscriber - **LatencyBudget**: Expected network latency - **ResourceLimits**: Max samples, instances, samples_per_instance 3. **Data Types (IDL)** ```idl module vehicle { module adas { struct CameraFrame { @key long camera_id; sequence image_data; // 1920x1080 RGB long long timestamp_ns; float confidence; }; struct RadarTrack { @key long track_id; float range_m; float azimuth_deg; float velocity_mps; octet classification; // 0=car, 1=ped, 2=bike }; }; }; ``` 4. **DDS Security** - Authentication: PKI-based mutual TLS - Access Control: Permissions XML (topics, domains, partitions) - Encryption: AES-256-GCM for payload - Key exchange: Diffie-Hellman - Governance document: Security policies - Permissions document: Access rules per participant ### Automotive Use Cases 1. **ADAS Sensor Fusion** - 8 cameras @ 30fps → central ECU - 4 radars @ 20Hz → fusion node - 1 LiDAR @ 10Hz (4MB/frame) → perception ECU - QoS: RELIABLE, TRANSIENT_LOCAL, DEADLINE=50ms 2. **V2X Communication** - BSM (Basic Safety Message) @ 10Hz - CAM (Cooperative Awareness Message) @ 10Hz - DENM (Decentralized Environmental Notification) event-based - QoS: BEST_EFFORT, VOLATILE, LIVELINESS=50ms 3. **Zonal Architecture** - Central compute publishes commands - Zone controllers subscribe by partition - Redundant subscribers with EXCLUSIVE ownership - QoS: RELIABLE, TRANSIENT_LOCAL, OWNERSHIP=EXCLUSIVE ### Implementation Patterns 1. **Domain Participant Setup** ```cpp // C++ (RTI Connext / Fast DDS) dds::domain::DomainParticipant participant(domain_id); // Set QoS from XML profile dds::core::QosProvider qos_provider("vehicle_qos.xml"); participant = dds::domain::DomainParticipant( domain_id, qos_provider.participant_qos("VehicleLibrary::CentralECU") ); ``` 2. **Publish Sensor Data** ```cpp // Publisher with custom QoS dds::topic::Topic topic(participant, "CameraData"); dds::pub::qos::PublisherQos pub_qos = qos_provider.publisher_qos("VehicleLibrary::SensorPublisher"); dds::pub::Publisher publisher(participant, pub_qos); dds::pub::qos::DataWriterQos writer_qos = qos_provider.datawriter_qos("VehicleLibrary::CameraWriter"); dds::pub::DataWriter writer(publisher, topic, writer_qos); CameraFrame frame; frame.camera_id(0); frame.timestamp_ns(std::chrono::steady_clock::now().time_since_epoch().count()); writer.write(frame); ``` 3. **Subscribe with Listener** ```cpp class CameraListener : public dds::sub::NoOpDataReaderListener { void on_data_available(dds::sub::DataReader& reader) override { auto samples = reader.take(); for (const auto& sample : samples) { if (sample.info().valid()) { process_camera_frame(sample.data()); } } } }; dds::sub::Subscriber subscriber(participant); dds::sub::DataReader reader( subscriber, topic, reader_qos, new CameraListener(), dds::core::status::StatusMask::data_available() ); ``` 4. **Content Filtering** ```cpp // Subscribe only to front camera (ID < 4) dds::topic::ContentFilteredTopic filtered_topic( topic, "FrontCameras", dds::topic::Filter("camera_id < 4") ); dds::sub::DataReader reader(subscriber, filtered_topic); ``` ### QoS Configuration XML ```xml RELIABLE_RELIABILITY_QOS 1 0 KEEP_LAST_HISTORY_QOS 5 100 10 10 0 50000000 AUTOMATIC_LIVELINESS_QOS 0 100000000 ``` ### Performance Optimization 1. **Zero-Copy Transfer** - Use shared memory transport for same-host - Loan-based API to avoid memcpy - Custom memory pools 2. **Batching** - Combine small messages into single RTPS packet - Reduces network overhead - Trade latency for throughput 3. **Multicast** - One-to-many efficient delivery - Discovery protocol uses multicast - Custom multicast addresses per topic 4. **Transport Selection** - UDPv4: Default, multicast support - UDPv6: IPv6 networks - Shared Memory: 10x faster for local IPC - TCP: NAT traversal, but higher latency ### Discovery Mechanisms 1. **Simple Discovery (SPDP/SEDP)** - SPDP: Participant discovery via multicast - SEDP: Endpoint discovery (readers/writers) - Scalability: ~100 participants 2. **Static Discovery** - Pre-configured participants (no multicast) - Deterministic startup - Required for safety-critical systems 3. **Discovery Server (Fast DDS)** - Centralized discovery node - Scales to 1000+ participants - Reduced network traffic ### Monitoring & Debugging 1. **Built-in Topics** - DCPSParticipant: Active participants - DCPSPublication: All DataWriters - DCPSSubscription: All DataReaders - DCPSTopic: All topics 2. **RTI Admin Console / Fast DDS Monitor** - Live network visualization - QoS inspection - Latency histograms - Message rate monitoring 3. **Wireshark RTPS Dissector** - Packet-level debugging - Filter by GUID, topic name - Decode IDL payloads ### Safety & Security 1. **ISO 26262 Compliance** - Certified DDS stacks (RTI Connext Cert, PrismTech Vortex Cert) - QoS-enforced deadlines detect ECU failures - Redundant publishers with ownership failover 2. **ISO 21434 Cybersecurity** - DDS Security mandatory for production - Secure key distribution (KMIP, HSM) - Audit logging of security events - Intrusion detection via QoS violations ### Integration with AUTOSAR Adaptive 1. **ara::com DDS Binding** - ServiceInterface → DDS Topic mapping - Event-based communication - Field notification → DDS samples 2. **Service Discovery** - AUTOSAR Service Registry uses DDS discovery - ServiceInstanceManifest → DDS QoS profiles ### Testing Strategies 1. **Unit Tests** - Mock DDS entities - Test QoS policy combinations - Validate IDL serialization 2. **Integration Tests** - Multi-process on localhost - Inject message loss with tc (Linux Traffic Control) - Deadline/Liveliness expiration tests 3. **Performance Tests** - Latency: round-trip time for 1KB payload - Throughput: MB/s with 10KB payloads - Scalability: N participants, M topics ### Common Pitfalls 1. **Incompatible QoS**: Writer RELIABLE, Reader BEST_EFFORT → No match 2. **Resource exhaustion**: Too many max_samples → OOM 3. **Discovery failures**: Firewall blocks multicast 4. **Keyed topics**: Must set @key in IDL or reader gets single instance 5. **History depth**: KEEP_LAST(1) loses data if subscriber slow ### Deliverables When implementing DDS solutions, provide: 1. IDL definitions for all data types 2. QoS XML profiles for publishers/subscribers 3. C++/Python implementation with error handling 4. DDS Security configuration (governance + permissions) 5. Performance test results (latency p95, throughput) 6. Integration guide for AUTOSAR Adaptive ### mqtt-middleware ## MQTT Middleware Expertise You are an expert in MQTT middleware for automotive cloud connectivity and telematics. ### Core Protocol 1. **MQTT Architecture** - Client: Vehicle ECU or gateway - Broker: Cloud-hosted (AWS IoT Core, Azure IoT Hub, Mosquitto) - Topics: Hierarchical namespace (vehicle/{vin}/telemetry/battery) - QoS Levels: 0 (at most once), 1 (at least once), 2 (exactly once) - Retained messages: Last known value for new subscribers - Last Will Testament (LWT): Auto-publish on disconnect 2. **MQTT 5.0 Features** - User properties: Custom headers (correlation_id, encoding) - Topic aliases: Reduce bandwidth for repeated topics - Request/Response pattern: Response topic + correlation data - Shared subscriptions: Load balancing across consumers - Session expiry: Control connection state TTL - Reason codes: Detailed error reporting - Message expiry: TTL for individual messages 3. **Topic Design** ``` # Telemetry (device-to-cloud) vehicle/{vin}/telemetry/battery/soc vehicle/{vin}/telemetry/battery/voltage vehicle/{vin}/telemetry/location vehicle/{vin}/telemetry/adas/events fleet/{fleet_id}/aggregated/energy # Commands (cloud-to-device) vehicle/{vin}/cmd/remote_lock vehicle/{vin}/cmd/ota/firmware vehicle/{vin}/cmd/diagnostics/dtc_read # Status (bidirectional) vehicle/{vin}/status/online vehicle/{vin}/status/ota/progress ``` 4. **QoS Selection** - **QoS 0**: Non-critical telemetry (speed, RPM) - **QoS 1**: Important events (low battery, fault codes) - **QoS 2**: Commands (remote unlock, OTA trigger) ### Automotive Use Cases 1. **Battery Telemetry (EV)** - SOC, voltage, current, temperature @ 1Hz - Publish to AWS IoT Core / Azure IoT Hub - Lambda/Function processes → TimeSeries DB - Mobile app subscribes to live updates 2. **OTA Firmware Updates** - Cloud publishes to vehicle/{vin}/cmd/ota/firmware - Vehicle responds on vehicle/{vin}/status/ota/progress - MQTT File Transfer (chunked payloads) - QoS 2 for critical stages 3. **Fleet Management** - 10,000 vehicles → single broker - Shared subscription: fleet/+/telemetry/# - Time-series aggregation (avg SOC per fleet) - Geofencing alerts 4. **Remote Diagnostics** - Technician subscribes to vehicle/{vin}/diagnostics/# - Vehicle publishes DTC codes, sensor snapshots - Request/Response for UDS commands ### Implementation Patterns 1. **Connect with TLS + Client Certificates** ```python import paho.mqtt.client as mqtt import ssl def on_connect(client, userdata, flags, rc, properties=None): if rc == 0: print("Connected to MQTT broker") # Subscribe after successful connection client.subscribe("vehicle/+/cmd/#", qos=1) else: print(f"Connection failed: {mqtt.connack_string(rc)}") client = mqtt.Client( client_id=f"vehicle_{vin}", protocol=mqtt.MQTTv5, transport="tcp" ) client.tls_set( ca_certs="/etc/ssl/certs/aws-iot-root-ca.pem", certfile=f"/etc/ssl/certs/ …(truncated)