add python example, add CP message routing key

razvanphp · razvanphp · commit c3e5e8bfcb87 · 2025-10-29T00:45:04.000+02:00
diff --git a/README.md b/README.md
@@ -35,25 +35,22 @@ RabbitMQ already excels at message durability and routing, but traditionally the
 ## Installation
 
 This plugin works only with modern versions of RabbitMQ 4.x based on AMQP 1.0.
+You can [build from source](https://www.rabbitmq.com/plugin-development.html) or you can download the latest release build from GitHub. Then, unzip and place the `rabbitmq_web_ocpp-4.x.x.ez` file into your `/etc/rabbitmq/plugins/` folder.
 Like all plugins, it [must be enabled](https://www.rabbitmq.com/plugins.html) before it can be used:
 
 ``` bash
 # this might require sudo
 rabbitmq-plugins enable rabbitmq_web_ocpp
 ```
 
-## Documentation
-
-For all configuration options, please refer to the nearly identical plugin, [RabbitMQ Web MQTT guide](https://www.rabbitmq.com/web-mqtt.html).
-
+Detailed instructions on how to install a plugin into RabbitMQ broker can be found [here](https://www.rabbitmq.com/plugins.html#installing-plugins).
 
-## Building From Source
+Note that release branches (`v4.1.x` vs. `main`) and target RabbitMQ version need to be taken into account
+when building plugins from source.
 
- * [Generic plugin build instructions](https://www.rabbitmq.com/plugin-development.html)
- * Instructions on [how to install a plugin into RabbitMQ broker](https://www.rabbitmq.com/plugins.html#installing-plugins)
+## Documentation
 
-Note that release branches (`stable` vs. `master`) and target RabbitMQ version need to be taken into account
-when building plugins from source.
+For all configuration options, please refer to the nearly identical plugin, [RabbitMQ Web MQTT guide](https://www.rabbitmq.com/web-mqtt.html).
 
 ## Enterprise-Grade Hosting & SLA Support
 
diff --git a/examples/python/.gitignore b/examples/python/.gitignore
@@ -0,0 +1 @@
+__pycache__/
diff --git a/examples/python/README.md b/examples/python/README.md
@@ -0,0 +1,17 @@
+# OCPP AMQP Server
+
+This directory contains Python examples for handling OCPP (Open Charge Point Protocol) messages using RabbitMQ native OCPP to AMQP plugin. It uses the popular python OCPP lib, [mobilityhouse/ocpp](https://github.com/mobilityhouse/ocpp).
+
+## Files
+
+- `server.py` - CSMS / OCPP server
+- `amqp_charge_point.py` - Overrides the original ChargePoint class
+- `requirements.txt` - Python dependencies
+
+## Key Differences from traditional WebSocket Approach
+
+- Uses RabbitMQ message broker for communication
+- Messages are persistent and can survive connection drops
+- Better scalability and load distribution
+- Supports complex routing patterns
+- Can handle multiple charge points efficiently
diff --git a/examples/python/amqp_charge_point.py b/examples/python/amqp_charge_point.py
@@ -0,0 +1,66 @@
+import asyncio
+from datetime import datetime
+
+from aio_pika import Message, Queue, Exchange
+from ocpp.v16 import ChargePoint as cp
+
+
+class AmqpChargePoint(cp):
+    """
+    An OCPP ChargePoint implementation that is tightly integrated with AMQP.
+    
+    This class directly consumes from an AMQP queue and publishes responses
+    to an exchange, making it completely self-contained.
+    """
+
+    def __init__(self, queue: Queue, exchange: Exchange, **kwargs):
+        """
+        Args:
+            queue (aio_pika.Queue): The AMQP queue to consume messages from
+            exchange (aio_pika.Exchange): The AMQP exchange to publish responses to
+        """
+        # Pass None as the id since we'll extract it dynamically from routing keys
+        super().__init__(None, self, **kwargs)
+        self.queue = queue
+        self.exchange = exchange
+
+    async def _send(self, message: str):
+        """
+        Overrides the base `_send` method to publish the message to the
+        AMQP exchange using the current charge point ID from the routing key.
+        """
+
+        self.logger.info("%s: send %s", self.id, message)
+        msg = Message(
+            body=message.encode(),
+            content_type='application/json',
+            timestamp=int(datetime.now().timestamp()),
+        )
+        await self.exchange.publish(msg, routing_key=self.id)
+
+    async def start(self):
+        """
+        Start consuming from the AMQP queue and process messages directly.
+        Completely overrides the base class implementation.
+        """
+        async with self.queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                # Acknowledge the message automatically after processing
+                async with message.process(requeue=False):
+                    try:
+                        # Extract routing key and charge point ID
+                        routing_key = getattr(message, 'reply_to', '') or ''
+                        if not routing_key:
+                            self.logger.warning("Received message without reply_to set")
+                            continue
+                        self.id = routing_key
+                        
+                        # Decode message body
+                        body = message.body.decode()
+                        
+                        # Process the message directly
+                        self.logger.info("%s: receive message %s", routing_key, body)
+                        await self.route_message(body)
+                        
+                    except Exception as e:
+                        self.logger.exception("Error processing message: %s", e)
diff --git a/examples/python/requirements.txt b/examples/python/requirements.txt
@@ -0,0 +1,2 @@
+ocpp==2.0.0
+aio-pika==9.5.7
diff --git a/examples/python/server.py b/examples/python/server.py
@@ -0,0 +1,106 @@
+import asyncio
+import logging
+from datetime import datetime, timezone
+
+import aio_pika
+from aio_pika import Message, ExchangeType
+
+from ocpp.routing import on
+from ocpp.v16 import call_result
+from ocpp.v16.enums import Action, RegistrationStatus
+
+from amqp_charge_point import AmqpChargePoint
+
+EXCHANGE_NAME = 'amq.topic'
+QUEUE_NAME = 'ocpp.worker'
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class MyChargePoint(AmqpChargePoint):
+    """
+    Extends the AmqpChargePoint to add application-specific message handlers.
+    """
+    @on(Action.boot_notification)
+    async def on_boot_notification(
+        self, charge_point_vendor, charge_point_model, **kwargs
+    ):
+        logger.info(
+            "#### Boot notification from vendor %s, model %s",
+            charge_point_vendor,
+            charge_point_model,
+        )
+        return call_result.BootNotification(
+            current_time=datetime.now(tz=timezone.utc).isoformat(),
+            interval=10,
+            status=RegistrationStatus.accepted,
+        )
+
+    @on(Action.status_notification)
+    async def on_status_notification(
+        self,
+        status,
+        connector_id,
+        error_code,
+        **kwargs,
+    ):
+        logger.info(
+            "#### Status notification from connector %s: status=%s, error=%s",
+            connector_id,
+            status,
+            error_code,
+        )
+        return call_result.StatusNotification()
+
+    @on(Action.heartbeat)
+    async def on_heartbeat(self, **kwargs):
+        return call_result.Heartbeat(
+            current_time=datetime.now(tz=timezone.utc).isoformat()
+        )
+
+async def main():
+    """Main async entry: connect to RabbitMQ and start the OCPP charge point."""
+    connection = None
+
+    try:
+        connection = await aio_pika.connect_robust(
+            host='localhost',
+            port=5672,
+            virtualhost='/',
+            login='guest',
+            password='guest',
+        )
+
+        channel = await connection.channel()
+        exchange = await channel.declare_exchange(EXCHANGE_NAME, ExchangeType.TOPIC, durable=True)
+        queue = await channel.declare_queue(QUEUE_NAME, durable=True)
+
+        # Add bindings to route messages to our queue
+        # Bind to receive messages for all message types (wildcard pattern)
+        await queue.bind(exchange, routing_key='ocpp16.#')
+
+        # Create and start the charge point
+        charge_point = MyChargePoint(queue, exchange)
+
+        logger.info('Listening on queue "%s" and publishing to exchange "%s"', QUEUE_NAME, EXCHANGE_NAME)
+        
+        # Start the charge point (this will run forever)
+        await charge_point.start()
+
+    except asyncio.CancelledError:
+        logger.info('Shutting down...')
+    except Exception:
+        logger.exception('Fatal error in main')
+    finally:
+        if connection:
+            await connection.close()
+
+
+if __name__ == '__main__':
+    print('Demo OCPP Central System (AMQP-backed, Stateless Worker)')
+    print('Publishing responses to exchange: amq.topic with routing_key = <chargePointId>')
+    print()
+
+    asyncio.run(main())
diff --git a/src/rabbit_web_ocpp_processor.erl b/src/rabbit_web_ocpp_processor.erl
@@ -178,15 +178,20 @@ process_incoming(McOcpp = #ocpp_msg{},
             %% 2. Validate Exchange
             case rabbit_exchange:lookup(ExchangeName) of
                 {ok, Exchange} ->
-                    %% 3. Prepare annotations and initialize Message Container (mc) with mc_ocpp type
-                    Anns = #{?ANN_EXCHANGE => ?DEFAULT_EXCHANGE_NAME},
+                    %% 3. Generate structured routing key (protocolver.actionname.req/conf)
+                    RoutingKey = generate_routing_key(McOcpp, State),
+                    
+                    %% 4. Prepare annotations and initialize Message Container (mc) with mc_ocpp type
+                    Anns = #{?ANN_EXCHANGE => ?DEFAULT_EXCHANGE_NAME,
+                             ?ANN_ROUTING_KEYS => [RoutingKey]},
                     case mc:init(mc_ocpp, McOcpp, Anns, #{}) of
                         McMsg ->
-                            ?LOG_DEBUG("Created message container for ClientId ~ts: ~tp", [ClientId, McMsg]),
-                            %% 4. Trace (optional)
+                            ?LOG_DEBUG("Created message container for ClientId ~ts with routing key ~ts: ~tp", 
+                                      [ClientId, RoutingKey, McMsg]),
+                            %% 5. Trace (optional)
                             rabbit_trace:tap_in(McMsg, [ExchangeName], ConnName, User#user.username, TraceState),
 
-                            %% 5. Publish to the exchange with the ClientId as routing key
+                            %% 6. Publish to the exchange with the structured routing key
                             case rabbit_exchange:route(Exchange, McMsg, #{}) of
                                 ok ->
                                     ?LOG_INFO("Message routed successfully for ClientId ~ts", [ClientId]),
@@ -383,6 +388,34 @@ terminate(Reason, Infos, _State = #state{queue_states = QStates,
 
 %% --- Internal Functions ---
 
+%% @doc Generates a structured routing key in the format: ocpp.protocolver.actionname.req/conf
+%% Examples: "ocpp.ocpp16.BootNotification.req", "ocpp.ocpp20.Heartbeat.conf", "ocpp.ocpp201.StatusNotification.req"
+-spec generate_routing_key(#ocpp_msg{}, state()) -> binary().
+generate_routing_key(#ocpp_msg{msg_type = MsgType, action = Action}, 
+                     #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
+    %% Convert protocol version atom to binary string
+    ProtoVerBin = atom_to_binary(ProtoVer, utf8),
+
+    %% Handle action name (may be undefined for responses)
+    ActionBin = case Action of
+        undefined -> <<"response">>;  % For CALLRESULT/CALLERROR without action
+        ActionName when is_binary(ActionName) -> ActionName;
+        _ -> <<"unknown">>
+    end,
+    
+    %% Determine message direction (req/conf/error)
+    MsgTypeBin = case MsgType of
+        ?OCPP_MESSAGE_TYPE_CALL -> <<"req">>;      % Request from charge point
+        ?OCPP_MESSAGE_TYPE_SEND -> <<"req">>;      % Request in OCPP 2.1
+        ?OCPP_MESSAGE_TYPE_CALLRESULT -> <<"conf">>; % Response/confirmation
+        ?OCPP_MESSAGE_TYPE_CALLERROR -> <<"error">>; % Error response
+        ?OCPP_MESSAGE_TYPE_CALLRESULTERROR -> <<"error">>; % Error in OCPP 2.1
+        _ -> <<"unknown">>
+    end,
+    
+    %% Construct routing key: ocpp.protocolver.actionname.req|conf|error
+    <<ProtoVerBin/binary, ".", ActionBin/binary, ".", MsgTypeBin/binary>>.
+
 %% @doc Extracts relevant metadata from a parsed OCPP message list.
 % -spec extract_ocpp_metadata(list()) -> {ok, MsgId :: binary(), Action :: binary() | undefined} | {error, term()}.
 % extract_ocpp_metadata([?OCPP_MESSAGE_TYPE_CALL, MsgId, Action, _Payload]) when is_binary(MsgId), is_binary(Action) -> {ok, MsgId, Action};
@@ -552,7 +585,7 @@ consume_from_queue(State = #state{cfg = #cfg{queue_name = QName, client_id = Cli
                      limiter_active => false,
                      mode => {simple_prefetch, Prefetch},
                      consumer_tag => ?CONSUMER_TAG,
-                     exclusive_consume => false, % Allow other consumers? (Usually false for OCPP)
+                     exclusive_consume => true, % Allow other consumers? (Usually false for OCPP)
                      args => [],
                      ok_msg => undefined,
                      acting_user => User#user.username},
