fix: PATCH NatsJS stream subjects instead PUT

Lancetnik · Aug 26, 2023 · 48ce79e · 48ce79e
1 parent 8f3c384
commit 48ce79e
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 9 deletions.
diff --git a/propan/__about__.py b/propan/__about__.py
@@ -2,7 +2,7 @@
 
 from unittest.mock import Mock
 
-__version__ = "0.1.5.20"
+__version__ = "0.1.5.21"
 
 
 INSTALL_MESSAGE = (

diff --git a/propan/_compat.py b/propan/_compat.py
@@ -122,11 +122,7 @@ def update_model_example(model: Type[BaseModel]) -> Type[BaseModel]:
             return type(
                 model.__name__,
                 (model,),
-                {
-                    "Config": type(
-                        "Config", (model.Config,), {"schema_extra": schema}
-                    )
-                },
+                {"Config": type("Config", (model.Config,), {"schema_extra": schema})},
             )
         else:
             return model

diff --git a/propan/brokers/nats/nats_js_broker.py b/propan/brokers/nats/nats_js_broker.py
@@ -133,21 +133,26 @@ async def close(
 
     async def _start(self):
         await super()._start()
+        subjects = {h.subject for h in self.handlers}
         try:  # pragma: no branch
             await self._connection.add_stream(
                 config=self._stream_config,
-                subjects=[h.subject for h in self.handlers],
+                subjects=tuple(subjects),
             )
         except nats.js.errors.BadRequestError as e:
+            old_config = (
+                await self._connection.stream_info(self._stream_config.name)
+            ).config
+
             c = self._get_log_context(None, "")
             if (
                 e.description
                 == "stream name already in use with a different configuration"
             ):
-                self._log(e, logging.WARNING, c, exc_info=e)
+                self._log(e, logging.WARNING, c)
                 await self._connection.update_stream(
                     config=self._stream_config,
-                    subjects=[h.subject for h in self.handlers],
+                    subjects=tuple(set(old_config.subjects).union(subjects)),
                 )
             else:  # pragma: no cover
                 self._log(e, logging.ERROR, c, exc_info=e)