Merge pull request #391 from capocchi/version-5.1

Version 5.1

Author: capocchi

devsimpy/DEVSKernel/BrokerDEVS/ARCHITECTURE.md

@@ -22,6 +22,7 @@
 import json
 from typing import Any, Dict, Optional, Callable
 from threading import Event
+from queue import Queue
 
 from DEVSKernel.BrokerDEVS.Core.BrokerAdapter import BrokerAdapter
 
@@ -99,9 +100,8 @@ def __init__(self, broker_address: str, port: int = 1883, **kwargs):
         self.port = port
         self.client = mqtt.Client()
         self.subscribed_topics = []
-        self.message_queue = []
+        self.message_queue = Queue()  # Thread-safe queue for messages
         self.connected_event = Event()
-        self.last_message = None
 
         # Set up callbacks
         self.client.on_connect = self._on_connect
@@ -145,7 +145,7 @@ def _on_connect(self, client, userdata, flags, rc):
     def _on_message(self, client, userdata, msg):
         """Callback for when a PUBLISH message is received from the broker."""
         mqtt_message = MqttMessage(msg.topic, msg.payload, msg.qos)
-        self.last_message = mqtt_message
+        self.message_queue.put(mqtt_message)
 
     def _on_disconnect(self, client, userdata, rc):
         """Callback for when client disconnects from broker."""
@@ -181,14 +181,12 @@ def poll(self, timeout: float = 1.0) -> Optional[MqttMessage]:
         Returns:
             MqttMessage if available, None otherwise
         """
-        if self.last_message is not None:
-            msg = self.last_message
-            self.last_message = None
-            return msg
-        
-        # Wait for a message
-        time.sleep(min(timeout, 0.1))
-        return self.last_message
+        try:
+            # Try to get a message from the queue with timeout
+            return self.message_queue.get(timeout=timeout)
+        except:
+            # Queue is empty or timeout occurred
+            return None
 
     def close(self) -> None:
         """Close the consumer connection."""

devsimpy/DEVSKernel/BrokerDEVS/Brokers/mqtt/MqttAdapter.py

@@ -6,7 +6,7 @@
 logger = logging.getLogger("DEVSKernel.BrokerDEVS.InMemoryKafkaWorker")
 logger.setLevel(LOGGING_LEVEL)
 
-from DEVSKernel.BrokerDEVS.InMemoryKafkaWorker import InMemoryKafkaWorker
+from DEVSKernel.BrokerDEVS.Workers.InMemoryKafkaWorker import InMemoryKafkaWorker
 
 from DEVSKernel.BrokerDEVS.Core.BrokerMessageTypes import (
 	BaseMessage,
@@ -22,7 +22,7 @@
 	SimulationDone,
 )
 
-from DEVSKernel.BrokerDEVS.MS4Me.ms4me_kafka_wire_adapters import StandardWireAdapter
+from DEVSKernel.BrokerDEVS.DEVSStreaming.ms4me_kafka_wire_adapters import StandardWireAdapter
 from DomainInterface.Object import Message
 
 class MS4MeKafkaWorker(InMemoryKafkaWorker):

devsimpy/DEVSKernel/BrokerDEVS/Brokers/mqtt/data/mosquitto.db

@@ -12,7 +12,7 @@
 #
 # GENERAL NOTES AND REMARKS:
 #
-# MQTT-specific worker for MS4Me distributed DEVS simulation.
+# MQTT-specific worker for DEVSStreaming distributed DEVS simulation.
 # Mirrors MS4MeKafkaWorker but uses MQTT as the message broker.
 #
 ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##
@@ -53,15 +53,15 @@
 
 class MS4MeMqttWorker(threading.Thread):
     """
-    MQTT-based worker thread for MS4Me distributed DEVS simulation.
+    MQTT-based worker thread for DEVSStreaming distributed DEVS simulation.
     
     Manages one atomic DEVS model running in-memory while communicating
     with the coordinator via MQTT topics.
     
     Features:
     - Receives commands from coordinator on input topic
     - Sends model outputs to coordinator on output topic
-    - Handles MS4Me message serialization/deserialization
+    - Handles DEVSStreaming message serialization/deserialization
     - Thread-safe operations
     - Automatic reconnection on disconnect
     """
@@ -119,7 +119,12 @@ def __init__(
         self.password = password
 
         # Wire adapter for message serialization
-        self.wire_adapter = wire_adapter
+        if wire_adapter is None:
+            # Default to StandardWireAdapter (pickle-based) to match proxy defaults
+            from DEVSKernel.BrokerDEVS.DEVSStreaming.ms4me_mqtt_wire_adapters import StandardWireAdapter
+            self.wire_adapter = StandardWireAdapter()
+        else:
+            self.wire_adapter = wire_adapter
 
         # MQTT client - try VERSION2 API first (paho-mqtt 2.x)
         try:

…y/DEVSKernel/BrokerDEVS/MS4Me/DockerFile …rnel/BrokerDEVS/DEVSStreaming/DockerFiledevsimpy/DEVSKernel/BrokerDEVS/MS4Me/DockerFile renamed to devsimpy/DEVSKernel/BrokerDEVS/DEVSStreaming/DockerFile devsimpy/DEVSKernel/BrokerDEVS/MS4Me/DockerFile renamed to devsimpy/DEVSKernel/BrokerDEVS/DEVSStreaming/DockerFile

@@ -13,8 +13,8 @@
 
 from DEVSKernel.PyDEVS.SimStrategies import DirectCouplingPyDEVSSimStrategy
 from DomainInterface import DomainStructure, DomainBehavior
-from DEVSKernel.BrokerDEVS.MS4Me.MS4MeKafkaWorker import MS4MeKafkaWorker
-from DEVSKernel.BrokerDEVS.MS4Me.ms4me_kafka_wire_adapters import StandardWireAdapter
+from DEVSKernel.BrokerDEVS.DEVSStreaming.MS4MeKafkaWorker import MS4MeKafkaWorker
+from DEVSKernel.BrokerDEVS.DEVSStreaming.ms4me_kafka_wire_adapters import StandardWireAdapter
 from DEVSKernel.BrokerDEVS.Core.BrokerMessageTypes import (
     BaseMessage,
     SimTime,
@@ -27,11 +27,11 @@
     TransitionDone,
     SimulationDone,
 )
-from DEVSKernel.BrokerDEVS.Proxies.BrokerReceiverProxy import KafkaReceiverProxy
-from DEVSKernel.BrokerDEVS.Proxies.BrokerStreamProxy import KafkaStreamProxy
-from DEVSKernel.BrokerDEVS.MS4Me.auto_kafka import ensure_kafka_broker
+from DEVSKernel.BrokerDEVS.Proxies.kafka import KafkaReceiverProxy
+from DEVSKernel.BrokerDEVS.Proxies.kafka import KafkaStreamProxy
+from DEVSKernel.BrokerDEVS.DEVSStreaming.auto_kafka import ensure_kafka_broker
 from DEVSKernel.BrokerDEVS.logconfig import configure_logging, LOGGING_LEVEL
-from DEVSKernel.BrokerDEVS.MS4Me.kafkaconfig import KAFKA_BOOTSTRAP, AUTO_START_KAFKA_BROKER
+from DEVSKernel.BrokerDEVS.DEVSStreaming.kafkaconfig import KAFKA_BOOTSTRAP, AUTO_START_KAFKA_BROKER
 
 
 configure_logging()
@@ -238,13 +238,11 @@ def __init__(
         # Instantiate proxies (replaces direct Producer/Consumer)
         self._stream_proxy = StreamProxyClass(
             self.bootstrap,
-            wire_adapter=self.wire,
         )
 
         self._receiver_proxy = ReceiverProxyClass(
             self.bootstrap,
             group_id,
-            wire_adapter=self.wire,
         )
 
         # Subscribe to output topic
@@ -666,7 +664,7 @@ def _simulate_for_ms4me(self, T=1e8):
                 )
 
             self._terminate_workers()
-            logger.info("MS4Me BrokerDEVS Simulation Ended")
+            logger.info("DEVSStreaming BrokerDEVS Simulation Ended")
 
     def __del__(self):
         """Cleanup: close proxies properly"""

…nel/BrokerDEVS/MS4Me/MS4MeKafkaWorker.py …erDEVS/DEVSStreaming/MS4MeKafkaWorker.pydevsimpy/DEVSKernel/BrokerDEVS/MS4Me/MS4MeKafkaWorker.py renamed to devsimpy/DEVSKernel/BrokerDEVS/DEVSStreaming/MS4MeKafkaWorker.py devsimpy/DEVSKernel/BrokerDEVS/MS4Me/MS4MeKafkaWorker.py renamed to devsimpy/DEVSKernel/BrokerDEVS/DEVSStreaming/MS4MeKafkaWorker.py

@@ -1,10 +1,10 @@
 # -*- coding: utf-8 -*-
 """
 SimStrategyMqttMS4Me - MQTT-based distributed DEVS simulation strategy
-with in-memory workers (threads) using typed DEVS messages (MS4Me format).
+with in-memory workers (threads) using typed DEVS messages (DEVSStreaming format).
 
 Mirrors SimStrategyKafkaMS4Me but uses MQTT as the message broker instead of Kafka.
-Maintains compatibility with MS4Me message standardization and worker coordination.
+Maintains compatibility with DEVSStreaming message standardization and worker coordination.
 """
 
 import logging
@@ -13,8 +13,8 @@
 
 from DEVSKernel.PyDEVS.SimStrategies import DirectCouplingPyDEVSSimStrategy
 from DomainInterface import DomainStructure, DomainBehavior
-from DEVSKernel.BrokerDEVS.MS4Me.MS4MeMqttWorker import MS4MeMqttWorker
-from DEVSKernel.BrokerDEVS.MS4Me.ms4me_mqtt_wire_adapters import StandardWireAdapter
+from DEVSKernel.BrokerDEVS.DEVSStreaming.MS4MeMqttWorker import MS4MeMqttWorker
+from DEVSKernel.BrokerDEVS.DEVSStreaming.ms4me_mqtt_wire_adapters import StandardWireAdapter
 from DEVSKernel.BrokerDEVS.Core.BrokerMessageTypes import (
     BaseMessage,
     SimTime,
@@ -27,11 +27,11 @@
     TransitionDone,
     SimulationDone,
 )
-from DEVSKernel.BrokerDEVS.Proxies.MqttReceiverProxy import MqttReceiverProxy
-from DEVSKernel.BrokerDEVS.Proxies.MqttStreamProxy import MqttStreamProxy
-from DEVSKernel.BrokerDEVS.MS4Me.auto_mqtt import ensure_mqtt_broker
+from DEVSKernel.BrokerDEVS.Proxies.mqtt import MqttReceiverProxy
+from DEVSKernel.BrokerDEVS.Proxies.mqtt import MqttStreamProxy
+from DEVSKernel.BrokerDEVS.DEVSStreaming.auto_mqtt import ensure_mqtt_broker
 from DEVSKernel.BrokerDEVS.logconfig import configure_logging, LOGGING_LEVEL
-from DEVSKernel.BrokerDEVS.MS4Me.mqttconfig import MQTT_BROKER_ADDRESS, MQTT_BROKER_PORT, AUTO_START_MQTT_BROKER
+from DEVSKernel.BrokerDEVS.DEVSStreaming.mqttconfig import MQTT_BROKER_ADDRESS, MQTT_BROKER_PORT, AUTO_START_MQTT_BROKER
 
 
 configure_logging()
@@ -225,20 +225,20 @@ def resolve_hierarchical(model, target_port, visited=None):
 
 ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##
 #
-# MQTT SIMULATION STRATEGY WITH MS4Me MESSAGES
+# MQTT SIMULATION STRATEGY WITH DEVSStreaming MESSAGES
 #
 ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ## ##
 
 
 class SimStrategyMqttMS4Me(DirectCouplingPyDEVSSimStrategy):
     """
-    MQTT-based distributed DEVS simulation strategy with MS4Me message format.
+    MQTT-based distributed DEVS simulation strategy with DEVSStreaming message format.
     
     Orchestrates distributed DEVS simulation where:
     - Each atomic model runs in a worker thread
     - Workers communicate via MQTT topics
     - Coordinator manages simulation timing and routing
-    - Messages follow MS4Me standardization
+    - Messages follow DEVSStreaming standardization
     
     Features:
     - Thread-based in-memory workers
@@ -262,7 +262,7 @@ def __init__(
         mqtt_password: str = None,
     ):
         """
-        Initialize MQTT MS4Me simulation strategy.
+        Initialize MQTT DEVSStreaming simulation strategy.
         
         Args:
             simulator: PyDEVS simulator instance
@@ -349,7 +349,6 @@ def __init__(
             self.broker_address,
             self.broker_port,
             client_id=f"coordinator-{group_id}",
-            wire_adapter=self.wire,
             username=self.mqtt_username,
             password=self.mqtt_password,
         )
@@ -358,7 +357,6 @@ def __init__(
             self.broker_address,
             self.broker_port,
             client_id=f"receiver-{group_id}",
-            wire_adapter=self.wire,
             username=self.mqtt_username,
             password=self.mqtt_password,
         )
@@ -418,7 +416,6 @@ def _create_workers(self):
                 broker_port=self.broker_port,
                 in_topic=in_topic,
                 out_topic=out_topic,
-                wire_adapter=self.wire,
                 username=self.mqtt_username,
                 password=self.mqtt_password,
             )
@@ -470,7 +467,7 @@ def _await_msgs_from_mqtt(self, pending: Optional[List] = None) -> Dict:
     # ------------------------------------------------------------------
 
     def _simulate_for_ms4me(self, T=1e8):
-        """Simulate using standard MQTT MS4Me message routing."""
+        """Simulate using standard MQTT DEVSStreaming message routing."""
         try:
             # STEP 0: distributed init
             logger.info("Initializing atomic models...")
@@ -687,7 +684,7 @@ def _simulate_for_ms4me(self, T=1e8):
                 )
 
             self._terminate_workers()
-            logger.info("MS4Me MqttMS4Me Simulation Ended")
+            logger.info("DEVSStreaming MqttMS4Me Simulation Ended")
 
             # Call terminate to set end_flag and exit the simulation thread loop
             self._simulator.terminate()