This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 95f18b67ed fix(#3040): Change random topic to fixed with stream id
(#3041)
95f18b67ed is described below
commit 95f18b67edd8678a4caf594b97fd64c045b94fd7
Author: Philipp Zehnder <[email protected]>
AuthorDate: Wed Jul 17 13:32:15 2024 +0200
fix(#3040): Change random topic to fixed with stream id (#3041)
* fix(#3040): Change random topic to fixed with stream id
* fix(#3040): Remove if statement and sanitize stream id for topic name
* fix(#3040): Sanitized stream id and fix tests
* fix(#3040): Add stream id to river_function
---
.../streampipes/function_zoo/river_function.py | 1 +
.../functions/utils/data_stream_generator.py | 17 +++++++++++++----
.../tests/functions/test_function_handler.py | 9 +++++++--
.../tests/functions/test_river_function.py | 4 +++-
4 files changed, 24 insertions(+), 7 deletions(-)
diff --git
a/streampipes-client-python/streampipes/function_zoo/river_function.py
b/streampipes-client-python/streampipes/function_zoo/river_function.py
index 9bbcab3a74..55c057cf35 100644
--- a/streampipes-client-python/streampipes/function_zoo/river_function.py
+++ b/streampipes-client-python/streampipes/function_zoo/river_function.py
@@ -178,6 +178,7 @@ class OnlineML:
name="prediction",
attributes=attributes,
broker=get_broker_description(client.dataStreamApi.get(stream_ids[0])), #
type: ignore
+ stream_id=stream_ids[0]
)
function_definition =
FunctionDefinition(consumed_streams=stream_ids).add_output_data_stream(output_stream)
self.sp_function = RiverFunction(
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..11b333d412 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,9 +104,18 @@ def create_data_stream(
)
]
+ sanitized_stream_id = stream_id.replace(" ", "")
+
+ # 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.{sanitized_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
+
+ data_stream.element_id = sanitized_stream_id
return data_stream
diff --git a/streampipes-client-python/tests/functions/test_function_handler.py
b/streampipes-client-python/tests/functions/test_function_handler.py
index f1718f7aed..c4968539e0 100644
--- a/streampipes-client-python/tests/functions/test_function_handler.py
+++ b/streampipes-client-python/tests/functions/test_function_handler.py
@@ -373,7 +373,9 @@ class TestFunctionHandler(TestCase):
)
)
- output_stream = create_data_stream("test", attributes={"number":
RuntimeType.INTEGER.value})
+ output_stream = create_data_stream("test",
+ attributes={"number":
RuntimeType.INTEGER.value},
+ stream_id='a1d')
test_function = TestFunctionOutput(
function_definition=FunctionDefinition(
consumed_streams=["urn:streampipes.apache.org:eventstream:uPDKLI"]
@@ -436,7 +438,10 @@ class TestFunctionHandler(TestCase):
)
output_stream = create_data_stream(
- "test", attributes={"number": RuntimeType.INTEGER.value},
broker=SupportedBroker.KAFKA
+ "test",
+ stream_id='a1c',
+ attributes={"number": RuntimeType.INTEGER.value},
+ broker=SupportedBroker.KAFKA
)
test_function = TestFunctionOutput(
function_definition=FunctionDefinition(
diff --git a/streampipes-client-python/tests/functions/test_river_function.py
b/streampipes-client-python/tests/functions/test_river_function.py
index 597ee4963c..73f5a52fa1 100644
--- a/streampipes-client-python/tests/functions/test_river_function.py
+++ b/streampipes-client-python/tests/functions/test_river_function.py
@@ -56,7 +56,9 @@ class TestSupervisedModel:
class TestRiverFunction(TestCase):
def setUp(self) -> None:
self.data_stream = create_data_stream(
- "test", attributes={"number": RuntimeType.FLOAT.value, "bool":
RuntimeType.BOOLEAN.value}
+ "test",
+ attributes={"number": RuntimeType.FLOAT.value, "bool":
RuntimeType.BOOLEAN.value},
+ stream_id="sample-stream"
).to_dict()
self.test_stream_data = [
