Streams and ReaciveX as Nodes in Java

[a.formella]

Hello!

We use Knime in our organization. It’s a great software!
A few days ago I received a question - is it possible to use a Knime in the following scenario?

Scenario:
- When user from UK is registered send an e-mail do proper Seller
- Priority HIGH - send immidiately to Seller A
- Priority LOW - take 5 customers and send in one email to Seller B
- Database publish events using PostgreSQL NOTIFY (in Triggers)
- Knime ran in a docker as a service
- the scenario will be developed

I have started implementing ReaciveX as Nodes in Java (ReactiveX - Operators) but I ran into a few problems.

1. How to use connector port to PostgreSQL (I can’t see source code of org.knime.database.extension.postgres.node.connector.PostgreSQLDBConnectorNodeFactory nor any example)
2. Is it possible to implement “Event Stream” as a port? Do you have any tips?
3. Is it possible to use regular Knime nodes to work with such streams? (Joiner, Row Splitter…) Micro batch? KNIME Streaming Execution? Extends final class BufferedDataTable somehow?
4. How to implement multiple-execution of Nodes? “Postgres Notify Listener Stream” in example

An interesting case is the close of the Stream - Node “Take N” should return as much as it manages to take.

Early draft as an image:

There are several programs that allow for such a scenario to be implemented, but have significant drawbacks compared to Knime: Node-Red, Power Automate, Apache NiFi, SAP Hana Streaming.

Thanks!

[gab1one]

Hi @a.formella,

Very cool idea!

1. To see the code of the database extensions take a look here: Source code for Call Workflow nodes - #7 by gab1one
2. & 3. & 4. Currently, there is no support for “endless streaming” in KNIME AP, as this is fundamentally not expressed in the data model. The closest you can get is by using loops, if you want repeated execution of nodes.

You can model something like this with either loops or by using one workflow to call another one. Take a look at the Call Workflow Service node.

best,
Gabriel