This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch
3040-python-function-stops-producing-events-on-second-start
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/3040-python-function-stops-producing-events-on-second-start by this
push:
new 60f8ffaa2b fix(#3040): Change random topic to fixed with stream id
60f8ffaa2b is described below
commit 60f8ffaa2b89541d2bbf77641a843e3732fcbd00
Author: Philipp Zehnder <[email protected]>
AuthorDate: Wed Jul 17 10:54:49 2024 +0200
fix(#3040): Change random topic to fixed with stream id
---
.../streampipes/functions/utils/data_stream_generator.py | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git
a/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
b/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
index ee2ad36180..2f10ce625b 100644
---
a/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
+++
b/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
@@ -53,7 +53,7 @@ class RuntimeType(Enum):
def create_data_stream(
name: str,
attributes: Dict[str, str],
- stream_id: Optional[str] = None,
+ stream_id: str = None,
broker: SupportedBroker = SupportedBroker.NATS,
):
"""Creates a data stream
@@ -104,8 +104,15 @@ def create_data_stream(
)
]
+ # Assign a default topic name incorporating the unique stream ID to each
protocol.
+ # This ensures the topic name remains consistent across function restarts,
avoiding reliance on client-side defaults.
+ for protocol in transport_protocols:
+ protocol.topic_definition.actual_topic_name =
f"org.apache.streampipes.connect.{stream_id}"
+
data_stream = DataStream(
- name=name, event_schema=event_schema,
event_grounding=EventGrounding(transport_protocols=transport_protocols)
+ name=name,
+ event_schema=event_schema,
+ event_grounding=EventGrounding(transport_protocols=transport_protocols)
)
if stream_id:
data_stream.element_id = stream_id
