Idea: Implicit Readers and Writers

[jenspots]

If we take a look at the following pipeline.

<reader> a jvm:MemoryChannelReader.
<writer> a jvm:MemoryChannelWriter.

<bridge>
  a jvm:MemoryChannel;
  jvm:reader <reader1>;
  jvm:writer <writer1>.

[]
  a jvm:Range;
  jvm:start "0"^^xsd:integer;
  jvm:end "10"^^xsd:integer;
  jvm:step "1"^^xsd:integer;
  jvm:output <writer>.

[]
  a jvm:Reporter;
  jvm:input <reader>.

The declaration of the channel is quite verbose. An alternative way would be the following.

<channel>
  a jvm:Channel;
  jvm:input <proc/range>;
  jvm:output <proc/reporter>.

<proc/range> a jvm:Range;
  jvm:start "0"^^xsd:integer;
  jvm:end "10"^^xsd:integer;
  jvm:step "1"^^xsd:integer.

<proc/reporter> a jvm:Reporter.

Note that we still want to be able to enforce the existence of a channel using the processor SHACL shape. We would check for the existence of a jvm:Channel which has the processor as its jvm:input or jvm:output for readers and writers respectively. We just wouldn't use processor arguments for these any more.

Accessing the channel from the Reporter could be as simple as extending the Processor base class with getReader and getWriter, both would take in the URI of the source/target processor.

val reader = this.getReader("<proc/range>");

Note: the only reason we want to include the channel definition in the Turtle file is for static analysis using SHACL. That way we can enforce the existence of channels before runtime.

This would also work with HTTP channels. We could even simply add the desired port, or just let the runner decide at runtime.

<channel>
  a jvm:HttpChannel;
  jvm:port 80; # Optional!
  jvm:input <proc/range>;
  jvm:output <proc/reporter>.

[jenspots]

Additional note: we wouldn't even need to specify the type of channel. If the runner decides it wants to host two processors on different machines for example, it spawns HTTP readers/writers. If they run on the same machine, simply use memory channels.

A super runner would be able to do this as well. This way, it can pick the most optimal type of connection between runners. For example, if a memory channel exists from JS to JVM, it can use that. If it doesn't exist from JS to Python, use HTTP.

It now also has the freedom to distribute processors however it wants.

[jenspots]

Actually, the following won't work.

val reader = this.getReader("<proc/range>");

The processors are written in a pipeline agnostic way, so they can't access readers by URI.

An alternative would be using named readers and writers. For example, simply using input.

val reader = this.getReader("input");

The channel definition would require this to be included.

<channel>
  a jvm:Channel;
  jvm:input <proc/range>;
  jvm:inputTarget "input";
  jvm:output <proc/reporter>
  jvm:outputTarget "output".

Most channels only take in one reader and one writer at most, so we could use input and output as default names. The channel name would not need to be changed.

val reader = this.getReader(); // Default reader

But one could argue that inputTarget shouldn't be a property of the channel, but rather of the reader itself. Circling us back to our current solution 😅