Merge pull request #14 from AbletonAG/extra-stdout-flush

Add flushes to interactive replay

.github/workflows/python-package.yml

@@ -17,8 +17,8 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest]
-        python-version: [3.6, 3.7, 3.8, 3.9]
-        django-version: [2.2, 3.2]
+        python-version: ["3.8", "3.9", "3.10"]
+        django-version: ["3.2", "4.2"]
     runs-on: ${{ matrix.os }}
     steps:
     - uses: actions/checkout@v2

domain_event_broker/django/management/commands/replay_domain_event.py

@@ -42,23 +42,27 @@ def interactive_filter(self, body: bytes, **kwargs: Any) -> str:
         self.stdout.write("(R)eplay, (D)iscard or (L)eave?")
         # Wait for user input but time out before the RabbitMQ connection is
         # lost.
+        self.stdout.flush()
+        return_value = replay.LEAVE
+        message = "Message left in queue."
         action = input_timeout(30)
-        if action is None:
-            self.stdout.write("Timed out. Message left in queue.")
-            return replay.LEAVE
-        elif action == '':
-            self.stdout.write("Message left in queue.")
-            return replay.LEAVE
-        action = action[0].upper()
-        if action == 'R':
-            self.stdout.write("Message is returned to worker queue.")
-            return replay.RETRY
-        elif action == 'D':
-            self.stdout.write("Message has been discarded.")
-            return replay.DISCARD
+        # there had been user input
+        if action is not None:
+            # If the user input has been more that "Return", take the first char
+            # Defaults to (L)eave
+            action = action[0].upper() if action else ""
+            if action == 'R':
+                message = "Message is returned to worker queue."
+                return_value = replay.RETRY
+            elif action == 'D':
+                message = "Message has been discarded."
+                return_value = replay.DISCARD
         else:
-            self.stdout.write("Message left in queue.")
-            return replay.LEAVE
+            message = f"Timed out. {message}"
+
+        self.stdout.write(message)
+        self.stdout.flush()
+        return return_value
 
     def handle(self, *args: Any, **options: Any) -> None:
         callback = replay.retry_event

domain_event_broker/transport.py

@@ -17,7 +17,7 @@
 
 def publish_domain_event(routing_key: str,
                          data: Dict[str, Any],
-                         domain_object_id: str = None,
+                         domain_object_id: Optional[str] = None,
                          uuid_string: Optional[str] = None,
                          timestamp: Optional[float] = None,
                          connection_settings: Optional[str] = '',

setup.cfg

@@ -3,7 +3,7 @@ description-file = README.md
 
 [versioneer]
 VCS = git
-style = git-describe
+style = pep440
 versionfile_source = domain_event_broker/_version.py
 versionfile_build = domain_event_broker/_version.py
 tag_prefix =