This is an automated email from the ASF dual-hosted git repository.

bossenti pushed a commit to branch fix-function-output-timestamp
in repository https://gitbox.apache.org/repos/asf/streampipes.git

commit ae64f3f7ad7e2bf2ddfa864cd7fe1b4c91b86375
Author: bossenti <[email protected]>
AuthorDate: Wed Mar 13 14:33:29 2024 +0100

    fix: do not overwrite timestamp if available
---
 .../streampipes/functions/streampipes_function.py                     | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git 
a/streampipes-client-python/streampipes/functions/streampipes_function.py 
b/streampipes-client-python/streampipes/functions/streampipes_function.py
index 6e93f1d2fa..ae481467b5 100644
--- a/streampipes-client-python/streampipes/functions/streampipes_function.py
+++ b/streampipes-client-python/streampipes/functions/streampipes_function.py
@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+import datetime
 from abc import ABC, abstractmethod
 from time import time
 from typing import Any, Dict, List, Optional
@@ -63,7 +64,8 @@ class StreamPipesFunction(ABC):
         -------
         None
         """
-        event["timestamp"] = int(1000 * time())
+        if "timestamp" not in event.keys():
+            event["timestamp"] = int(1000 * time())
         self.output_collectors[stream_id].collect(event)
 
     def getFunctionId(self) -> FunctionId:

Reply via email to