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)

   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)

   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',
          
           auto_ack=False  # Manual ack
       )
   
       print("Waiting for messages...")
       channel.start_consuming()
   

3. Publisher Confirms

   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)

   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,
              
               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

   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

   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

   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)

   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)

   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)

   #include <coap3/coap.h>
   
   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)

   # 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)

   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

   # 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: </sensors/temp>;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)

   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)

   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)

   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

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

   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)

   module vehicle {
     module adas {
       struct CameraFrame {
         @key long camera_id;
         sequence<octet, 2073600> 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

   // 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

   // Publisher with custom QoS
   dds::topic::Topic<CameraFrame> 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<CameraFrame> 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

   class CameraListener : public dds::sub::NoOpDataReaderListener<CameraFrame> {
     void on_data_available(dds::sub::DataReader<CameraFrame>& 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<CameraFrame> reader(
     subscriber, topic, reader_qos,
     new CameraListener(), dds::core::status::StatusMask::data_available()
   );
   

4. Content Filtering

   // Subscribe only to front camera (ID < 4)
   dds::topic::ContentFilteredTopic<CameraFrame> filtered_topic(
     topic,
     "FrontCameras",
     dds::topic::Filter("camera_id < 4")
   );
   dds::sub::DataReader<CameraFrame> reader(subscriber, filtered_topic);
   

QoS Configuration XML

<?xml version="1.0" encoding="UTF-8"?>
<dds xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  <qos_library name="VehicleLibrary">
    <qos_profile name="CameraWriter" base_name="BuiltinQosLibExp::Generic.StrictReliable">
      <datawriter_qos>
        <reliability>
          <kind>RELIABLE_RELIABILITY_QOS</kind>
          <max_blocking_time>
            <sec>1</sec>
            <nanosec>0</nanosec>
          </max_blocking_time>
        </reliability>
        <history>
          <kind>KEEP_LAST_HISTORY_QOS</kind>
          <depth>5</depth>
        </history>
        <resource_limits>
          <max_samples>100</max_samples>
          <max_instances>10</max_instances>
          <max_samples_per_instance>10</max_samples_per_instance>
        </resource_limits>
        <deadline>
          <period>
            <sec>0</sec>
            <nanosec>50000000</nanosec> <!-- 50ms -->
          </period>
        </deadline>
        <liveliness>
          <kind>AUTOMATIC_LIVELINESS_QOS</kind>
          <lease_duration>
            <sec>0</sec>
            <nanosec>100000000</nanosec> <!-- 100ms -->
          </lease_duration>
        </liveliness>
      </datawriter_qos>
    </qos_profile>
  </qos_library>
</dds>

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

   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)