RDF-Connect Specification

Living Standard,

Issue Tracking:
GitHub
Inline In Spec
Editor:
RDF-Connect Team

CC0 To the extent possible under law, the editors have waived all copyright and related or neighboring rights to this work. In addition, as of 13 August 2026, the editors have made this specification available under the Open Web Foundation Agreement Version 1.0, which is available at https://www.openwebfoundation.org/the-agreements/the-owf-1-0-agreements-granted-claims/owfa-1-0. Parts of this work may be from another specification document. If so, those parts are instead covered by the license of that specification document.

Abstract

Some abstract

1. Introduction

RDF-Connect is a modular framework for building and executing multilingual data processing pipelines using RDF as the configuration and orchestration layer.

It enables fine-grained, reusable processor components that exchange streaming data, allowing workflows to be described declaratively across programming languages and environments. RDF-Connect is especially well suited for data transformation, integration, and linked data publication.


   
2. Usage Paths

   

This section is complete in terms of content, but may be reorganized or rewritten for clarity during editorial review.


   
Depending on your use case, you may only need a subset of this specification:

   
   
3. Design Goals

   

This section is complete in terms of content, but may be reorganized or rewritten for clarity during editorial review.


   
RDF-Connect is designed to be language-agnostic, enabling seamless integration of processor components written in diverse programming languages such as JavaScript, Python, and shell scripts.
This flexibility allows developers to leverage existing tools and libraries within their language of choice, avoiding the constraints of monolithic, single-language frameworks.

   
A key architectural principle of RDF-Connect is that it operates in a streaming-by-default mode.
Data flows between processors via a central orchestrator, supporting real-time and large-scale data processing scenarios.
This streaming model ensures efficient memory usage and enables continuous transformation and publication of data as it is ingested.

   
The configuration of pipelines, processors, inputs, and outputs is expressed semantically using RDF.
This use of semantic configuration promotes clarity, extensibility, and interoperability by describing the structure and behavior of the system in a machine-readable, standards-based format.

   
Processor components in RDF-Connect are designed for reusability.
A processor defined once can be reused across multiple pipelines and in varying contexts, reducing duplication and encouraging modular design practices.
This modularity fosters rapid prototyping and easier maintenance of complex workflows.

   
Transparency is a fundamental goal of the framework.
By leveraging RDF vocabularies to describe pipeline components and their interactions, RDF-Connect makes it easy to inspect, document, and reason about the behavior and structure of a pipeline, both during development and after deployment.

   
Finally, RDF-Connect provides native support for provenance tracking using the PROV-O ontology.
Each step in the pipeline can be traced to its source, enabling detailed lineage tracking.
This capability is especially critical in applications involving data quality, reproducibility, and compliance.

   
4. Concepts

   
This section introduces the core concepts underpinning the RDF-Connect framework.
These concepts collectively define the modular, streaming, and language-agnostic architecture that enables RDF-Connect to support sophisticated, provenance-aware data pipelines.

   
4.1. Pipeline

   
At the heart of RDF-Connect is the notion of a pipeline, which defines a structured sequence of processing steps.
Each step corresponds to a processor, configured with parameters and paired with an appropriate runner that governs its execution.
Pipelines specify the flow of data between processors using readers and writers, forming a streaming architecture in which data is continuously passed along and transformed.
The pipeline configuration itself is expressed in RDF, making it semantically explicit and machine-interpretable.

   
4.2. Processor

   
A processor is a modular, reusable software component that performs a discrete data processing task.
In a typical scenario, a processor receives input data through a reader, transforms it according to its logic, and emits the result via a writer.
However, processors are also flexible enough to support single-directional tasks, such as those that only produce or only consume data.
Crucially, processors are implementation-agnostic — they can be written in any programming language and integrated into pipelines via language-specific runners.
This makes processors the building blocks of RDF-Connect’s cross-language interoperability.

   
4.3. Runner

   
A runner is responsible for executing a processor on behalf of the orchestrator.
Each runner targets a specific language or execution environment, enabling processors written in different languages to participate seamlessly in a pipeline.
For example, a JavaScript processor would be executed using a NodeRunner, which knows how to initialize and manage JavaScript-based components.
In this sense, a runner serves as an execution strategy, abstracting away the platform-specific details of launching and interacting with a processor.

   
4.4. Orchestrator

   
The orchestrator is the central component that interprets and runs RDF-Connect pipelines.
It parses the RDF-based configuration, initializes and assigns runners, instantiates processors, and manages the data flow between components.
Acting as the conductor of the system, the orchestrator ensures that processors execute in the correct order and that data is passed efficiently and correctly through the pipeline.
Its role is fundamental to realizing the streaming and semantic integration goals of RDF-Connect.

   
4.5. Reader / Writer

   
Readers and writers provide the streaming interfaces that connect processors within a pipeline.
A writer streams data out from a processor, while a reader receives data into a processor.
Together, they define how data flows between pipeline steps in an idiomatic and composable way.
This separation of concerns allows for flexible data routing and makes it easy to compose and recombine processors across different pipeline configurations.

   
5. SHACL as Configuration Schema

   
RDF-Connect uses SHACL [shacl] not only as a data validation mechanism but also as a schema language for defining the configuration interface of components such as processors and runners.
These SHACL shapes enable:

   
   
Shapes define required and optional configuration properties, which are transformed into JSON objects at startup, according to the pipeline.

   

    
    This SHACL shape definition defines a configuration structure for a processor.
    In RDF-Connect, such shapes are used to describe required parameters.
    They result in a well-typed JSON object that developers can rely on during implementation.


    
SHACL shape defining some required configuration for a processor


[] a sh:NodeShape;
    sh:targetClass <FooBar>;
    sh:property [
        sh:name "repeat";         # JSON field name
        sh:datatype xsd:integer;  # specify datatype
        sh:maxCount 1;
        sh:path :repeat;
    ], [
        sh:name "messages";
        sh:datatype xsd:string;
        sh:path :msg;
    ].


    
Processor configuration


<MyProcessor> a <FooBar>;
    :repeat 10;
    :msg "Hello", "World".


    
Results in the following JSON structure.


{ 
    "repeat": 10,
    "messages": [ "Hello", "World"]
}


   

   
5.1. Mapping SHACL to Configuration Structures

   
Each sh:property statement in a SHACL shape directly maps to a field in a configuration object, such as a JSON structure.
This enables semantically rich, machine-validated configurations for processors and pipelines within RDF-Connect.

   
5.1.1. Required Fields

   
sh:path

   
This property indicates the RDF predicate that must be present on the target resource.
In the context of configuration mapping, sh:path is used to extract the corresponding value from the data graph.
It defines the link between the RDF representation and the configuration data being generated or interpreted.

   
sh:name

   
The sh:name property provides the external field name used in the resulting configuration object.
This allows decoupling the internal RDF predicate (as defined in sh:path) from how the value appears in the configuration file.

   
sh:datatype or sh:class

   
These properties define the expected type of the configuration field.
Use sh:datatype for primitive types such as xsd:string, xsd:boolean, or xsd:anyURI.
When the expected value is a nested resource, i.e., an object with its own structured fields, use sh:class instead.
This signals that the value is an embedded configuration object to be interpreted according to its own SHACL shape, enabling deeply nested configuration structures.

   
5.1.2. Optional Fields

   
sh:minCount

   
This property defines the minimum number of values that must be provided.
If the actual number of values is below this threshold, a validation error will be raised.
A sh:minCount of 1 or higher indicates that the field is required in the configuration.

   
sh:maxCount

   
This property sets the maximum number of values allowed for a field.
It also determines how the field should be interpreted structurally.
If sh:maxCount is set to 1, the corresponding field is treated as a single value (T).
If it is greater than 1 or left unspecified, the field is interpreted as a list or array of values (T[]).
This behavior ensures that the structure of the configuration aligns with user expectations and downstream processing logic.

   
5.2. Nested Shapes and Component Types

   
Configuration fields can reference structured values instead of primitive literals.
This is supported via sh:class, which indicates that the value must conform to another shape associated with a given RDF class.

   
For example:


sh:property [
  sh:path rdfc:input ;
  sh:name "input" ;
  sh:class rdfc:Reader ;
]


   
This defines a configuration field named input whose value is expected to be an instance of the class rdfc:Reader.

   
5.2.1. Use of sh:class and sh:targetClass

   
The sh:class predicate is used within a property constraint to indicate that the value of a field must be an RDF resource belonging to a specific class.
This class must, in turn, have a shape associated with it that defines its expected structure.

   
To make this connection, sh:targetClass is used on a sh:NodeShape.
This associates the shape with a class so that tools can look up the shape definition when they encounter an RDF resource of that class.

   
In RDF-Connect, sh:targetClass allows reusable schemas to be defined for configuration components.
These can then be referenced in other shapes through sh:class, enabling configuration structures that are both modular and type-safe.

   
5.2.2. Special Component Types: rdfc:Reader and rdfc:Writer

   
rdfc:Reader and rdfc:Writer are two special classes to represent input and output endpoints in RDF-Connect pipelines.
Fields that are declared with sh:class rdfc:Reader or sh:class rdfc:Writer are not simply nested configuration objects.
They serve as runtime injection points for streaming data.

   
At execution time, the runner is responsible for resolving these references into actual language-native abstractions.
Depending on the programming environment, these may be JavaScript streams, async iterators, or callback-based interfaces.

   
For example:


sh:property [
  sh:path rdfc:input ;
  sh:name "input" ;
  sh:class rdfc:Reader ;
  sh:minCount 1 ;
]


   
This declares an input field that must be provided and will be automatically connected to an input stream by the runner.
This design decouples the declarative pipeline configuration from the underlying data transport logic, allowing developers to focus on processor behavior rather than infrastructure.

   
5.3. Example: Putting it all together

   
The following SHACL definitions and RDF instance demonstrate how a FooBar processor might be configured to append text to each incoming message and send the result to an output channel.


[ ] a sh:NodeShape;
  sh:targetClass :Channels;
  sh:property [
    sh:path rdfc:input ;
    sh:name "input" ;
    sh:class rdfc:Reader ;
    sh:minCount 1 ;
    sh:maxCount 1 ;
  ], [
    sh:path rdfc:output ;
    sh:name "output" ;
    sh:class rdfc:Writer ;
    sh:minCount 1 ;
    sh:maxCount 1 ;
  ].

[ ] a sh:NodeShape ;
  sh:targetClass <FooBar2> ;
  sh:property [
    sh:path :channel ;
    sh:name "channels" ;
    sh:minCount 1 ;
    sh:class :Channels ;
  ], [
    sh:path :append ;
    sh:name "append" ;
    sh:minCount 1 ;
    sh:maxCount 1 ;
    sh:datatype xsd:string ;
  ].



<foobar> a <FooBar2>;
    :channel [
        #`a :Channels` is not  required, this is implicit from the definition
        rdfc:input <channel1>;
        rdfc:output <channel2>;
    ], [
        rdfc:input <channel2>;
        rdfc:output <channel1>;
    ];
    :append " World!".


   
This results in the following JSON object:


{
  channels: [
        { 
            "input": { /* idiomatic input stream for channel <channel1> */ } ,
            "output": {/* idiomatic output stream for channel <channel2> */ } ,
        }, {
            "input": { /* idiomatic input stream for channel <channel2> */ } ,
            "output": {/* idiomatic output stream for channel <channel1> */ } ,
        }
  ],
  append: " World!"
}


   
6. RDF-Connect by Layer

   
Communication between the orchestrator and the runners happens using a strongly typed protocol defined in Protocol Buffers (protobuf).
This enables language-independent and efficient communication across processes and machines.

   
The protobuf server is the orchestrator, which is the central point.
The orchestrator starts all runners and notifies the runners of the different processors they should start.
The orchestrator is also the message post office, allowing messages to be sent to the correct runner which will relay those messages to the correct processor.

   
6.1. Communication Protocol: Orchestrator ↔ Runner

   
RDF-Connect uses a bidirectional communication protocol based on Protocol Buffers (protobuf) for interaction between the Orchestrator and Runners.
The orchestrator manages execution, while runners host and execute individual processors.

   
6.1.1. Messages Sent to Runners

   
The orchestrator can send the following messages to the runner:

   
   
6.1.2. Messages Sent from Runners

   
The runner can send the following messages to the orchestrator:

   
   
6.2. Orchestrator

   
The orchestrator is the central runtime entity in RDF-Connect.
It reads the pipeline configuration, sets up the runners, initiates processors, and routes messages between them.
It ensures the data flow graph described by the pipeline is brought to life across isolated runtimes.
The orchestrator acts as a coordinator, not an executor. Each runner is responsible for running the actual processor code, but the orchestrator ensures the pipeline as a whole behaves as intended.

   
Responsibilities:

   
   

The remained of this section is intended for developers building custom runners or integrating RDF-Connect into infrastructure.


   
6.2.1. Startup Flow

   
6.2.1.1. Understanding the pipeline file

   
The orchestrator begins execution from a single pipeline RDF file. This file MUST first be expanded by resolving all owl:imports statements recursively.
Once the full RDF graph is assembled, the orchestrator extracts the pipeline to execute by locating a rdfc:Pipeline instance whose subject is the pipeline file itself.
The pipeline is composed of one or more runner–processor pairs, defined via the rdfc:consistsOf property.
Each pair includes:

   
   

    


    
This example pipeline contains three processors divided over two runners.
    The orchestrator starts this pipeline with two runners and provides them with the correct processor configurations.


@prefix rdfc: <https://w3id.org/rdf-connect#>.

<> a rdfc:Pipeline;
  rdfc:consistsOf [
    rdfc:instantiates rdfc:NodeRunner;
    rdfc:processor <sender>, <echo>;
  ], [
    rdfc:instantiates rdfc:RustRunner;
    rdfc:processor <log>;
  ].


   

   
6.2.1.2. Starting the runners

   
Each simple runner should specify two things:

   
   
   
Two kinds of runner instantiation are currently defined, and both MUST end up in the same state: a mainStream between the orchestrator and the runner, over which the runner has identified itself.


   
For a rdfc:CommandRunner, the orchestrator appends two arguments to the configured command: the URL of the orchestrator’s running Protobuf server and the IRI identifying the runner instance.
It then executes the resulting command to spawn the runner process, which is expected to connect back to the orchestrator with the RPC.connect method, setting up a bidirectional stream of messages, dubbed mainStream.

   
A rdfc:TcpRunner is not started by the orchestrator, but is already running and reachable over a plain TCP socket at the host:port given by rdfc:grpc instead of rdfc:command.
For such a runner the TCP dial direction is reversed: the orchestrator MUST connect to that address and write the runner’s IRI over the raw socket, and MUST then hand its end of that socket to the same gRPC server that serves RPC.connect for command runners.
From there the gRPC roles are unchanged — the runner still acts as the gRPC client and the orchestrator’s gRPC server still receives the connection, exactly as in the rdfc:CommandRunner case; only the TCP dial direction and the IRI handoff differ (see § 6.3.2 Connecting Flow).

   
When a message is sent, without identifying how, the message is sent using this mainStream.

   
   
The orchestrator MUST track all active runners, specifically recording which ones have sent an RPC.identify message after startup.
This message confirms that the runner is ready to accept processor assignments.
The orchestrator responds to the runner with a RPC.pipeline message, containing the full expanded pipeline as Turtle, enriched with the provenance triples inferred as described in § 6.2.3 Provenance Inference (e.g. the prov:Activity subclass facts derived from rdfc:jsImplementationOf and similar predicates).
This pipeline configuration is useful when a runner wants to extract processor arguments themselves, instead of using the provided JSON-LD, though as noted in § 6.2.1.3 Starting processors this is not required — the JSON-LD arguments sent per-processor are already sufficient.

   
Once all expected runners have successfully identified themselves, the orchestrator proceeds to the next step in the pipeline initialization sequence.

   
6.2.1.3. Starting processors

   
Once all runners have been successfully identified via RPC.identify, the orchestrator proceeds to initialize the processors defined in the pipeline.
This involves the extraction and transformation of processor configuration data into a format suitable for consumption by the associated runner.

   
Processor Arguments

   
Processor arguments are encoded as JSON-LD objects, providing a structured representation of RDF configuration data.
JSON-LD fits the requirements as it is selected for the following reasons, and it allows encoding of typed literals and nested structures, in alignment with SHACL definitions. 
This while still enabling extensibility, supporting advanced use cases such as capturing full SHACL paths or preserving provenance metadata.

   
Support for JSON-LD is optional for runners.
Runners MAY choose to treat the JSON-LD as plain JSON if they do not require the semantic context or graph-aware features.
However, all runners MUST accept the structure produced by the orchestrator.

   
The arguments JSON-LD object sent in RPC.proc is already fully resolved by the orchestrator and is, by design, sufficient on its own to construct a processor’s arguments: a runner is not expected to re-derive arguments from RDF.
A runner MAY instead extract arguments itself from the RPC.pipeline Turtle (e.g. using a SHACL-shape extraction library), as noted in § 6.2.1.2 Starting the runners, but this is purely an implementation choice, not a requirement — it is only useful for runners that need semantic/graph-aware features the JSON-LD form does not expose, or that already have SHACL-extraction tooling on hand for other reasons. It comes at the cost of duplicating the SHACL-to-configuration mapping logic that the orchestrator already performed once.

   
Processor arguments come from the SHACL shape defined for the processor type.
Each field is mapped following section Mapping SHACL to Configuration Structures.
A JSON-LD @context is generated mapping all sh:name values to the corresponding IRIs from sh:path.
 If the processor instance has a known RDF identifier or rdf:type, these are added to the JSON-LD using @id and @type.

   

    
    Let’s take this shape, note that the shape also includes a rdfc:Reader.


@prefix : <http://example.org/>.
[] a sh:NodeShape;
    sh:targetClass <FooBar>;
    sh:property [
        sh:name "reader";
        sh:property :reader;
        sh:class rdfc:Reader;
        sh:maxCount 1;
    ], [
        sh:name "count";
        sh:property :count;
        sh:datatype xsd:number;
        sh:maxCount 1;
    ].

<SomeId> a <FooBar>;
    :reader <ReaderId>;
    :count 42.


    
    The following JSON-LD structure is built. Which aligns with section Mapping SHACL to Configuration Structures.



{
    "@context": {
        "reader": "http://example.org/reader",
        "count": "http://example.org/count"
    }
    "@id": "SomeId",
    "@type": "FooBar",
    "reader": {
        "@type": "https://w3id.org/rdf-connect#Reader",
        "@id": "ReaderId"
    },
    "count": 42
}


   

   
Processor Definition Extraction

   
In addition to extracting processor instance arguments, the orchestrator MUST also extract the processor definition configuration. This definition provides implementation-specific parameters, typically required to launch the processor in a specific runtime (e.g., JavaScript entrypoints, file paths, class names, etc.).

   
Processor definitions are extracted using the same SHACL-based mechanism described previously. The shape used for this extraction is associated with the programming language or runtime type and MUST be processed in the same way to produce a structured JSON-LD object.

   
RPC message

   
Once both the arguments and definition have been extracted for a processor instance, the orchestrator sends an RPC.proc message to the appropriate runner, initiating the processor launch process.

   
The orchestrator MUST keep an internal record of all processor instance that have been dispatched to a runner, and the runner’s acknowledgment that a processor was successfully launched, as indicated by an incoming RPC.initialized message.

   
No processor may be assumed to be operational until its runner has responded with RPC.initialized.
When all processors are successfully initialized, the orchestrator can start the pipeline.

   
6.2.1.4. Starting the pipeline

   
The orchestrator can start the pipeline by sending a RPC.start message to each runner.

   
The full startup flow is shown in this diagram.



   
6.2.2. Handling messages

   
The orchestrator acts as a message broker between processors.
It is responsible for receiving messages from runners and forwarding them to the appropriate destination runner based on channel identifiers defined in the pipeline.
Importantly, channels support one-to-one communication: each channel is required to have exactly one Reader and one Writer.

   
6.2.2.1. Normal messages

   
When a runner sends a RPC.msg message to the orchestrator, the message includes a channel IRI indicating its logical destination.

   
The orchestrator MUST:

   
    
    
  1. 
     
    Resolve which processor consumes this channel.
    
    
  2. 
     
    Determine which runner is responsible for this processor.
    
    
  3. 
     
    Forward the message to the relevant runner using RPC.msg.
    
   

   
The consuming runner MUST respond with a RPC.processed message that indicates that the processor has consumed the message.
The orchestrator MUST forward this message to the producing runner.
Only when the runner received the RPC.processed message, the processor is allowed to send a new message on this channel.



   
6.2.2.2. Streaming messages

   
When the payload of a message is large, the streaming message protocol SHOULD be used.
This protocol enables large messages to be sent incrementally over a separate gRPC stream while maintaining channel-based routing.

   
Just like normal messages on the mainStream include a back pressure mechanism, each streaming message also has one.

   
The process is as follows:

   
    
    
  1. 
     
    Sender (runner) initiates a sendStreamMessage gRPC stream to the orchestrator.
    
    
  2. 
     
    Sender identifies this stream message by sending a RPC.StreamIdentify which includes a localSequenceNumber, the channel URI and the runner URI.
    
    
  3. 
     
    The orchestrator translates this localSequenceNumber to a globalSequenceNumber and sends a RPC.streamMsg over mainStream to the receiving runner with the globalSequenceNumber and channel URI.
    
    
  4. 
     
    Receiving runner connects to the orchestrator using receiveStreamMessage, and sends an identifying StreamControl with the globalSequenceNumber.
    
    
  5. 
     
    Once the receiving runner is connected, the orchestrator sends a StreamControl message to the Sending runner with a streamSequenceNumber value of 0, indicating that the stream has been established.
    
   

   
For each chunk sent by the sending processor, the following steps MUST take place:

   
    
    
  1. 
     
    Sender sends a StreamChunk message with the data of the chunk to the orchestrator.
    
    
  2. 
     
    The orchestrator relays this to the connected receiveStreamMessage stream.
    
    
  3. 
     
    The receiving runner forwards the chunk to the processor, enabling it to process the data.
    
    
  4. 
     
    The receiving runner returns a StreamControl message with the next streamSequenceNumber, this indicates that the chunk has been handled.
    
    
  5. 
     
    The orchestrator relays this StreamControl message to the sending runner.
    
    
  6. 
     
    The processor is permitted to send the next chunk of data.
    
   

   
Once all data has been sent, the sender closes the stream message stream.
Then orchestrator MUST close the associated receiving stream, this indicates the end of the streaming message for the receiving runner.



   
6.2.3. Provenance Inference

   
Before starting any runner, the orchestrator computes the pipeline’s PROV-O provenance metadata: it combines the expanded pipeline quads with the RDF-Connect ontology and a fixed set of N3 rules, and materializes the derived triples (such as inferring that a processor instance is a prov:Activity, as illustrated in § 6.2.1.2 Starting the runners).
The resulting quads, merged with the original pipeline quads, are what is sent to each runner as the RPC.pipeline Turtle; runners therefore do not need their own reasoning support to observe this inferred provenance.

   
During and after execution, the orchestrator additionally records runtime timing as further provenance: a prov:startedAtTime for each processor once RPC.start is sent, a prov:endedAtTime for each processor once all channels have closed, and a prov:generatedAtTime for each channel at the moment it closes.
Implementations MAY expose an option to persist the combined (inferred plus timing) provenance graph to a file once the pipeline finishes.

   
6.3. Runner

   
A runner in RDF-Connect is responsible for managing and executing processors within a specific execution context—typically a programming language runtime.
Each runner MUST connect with the orchestrator’s Protobuf server and follow the RDF-Connect protocol.

   
While minimal runners can be implemented with little overhead, they often become enriched with quality-of-life features to better support developers and processors operating in that language.

   
These quality-of-life features include:

   
   
This is useful for processor to execute longer running operations, like reading a file or consulting an external API.

   
Runners SHOULD coalesce or transform message types to simplify processor implementation.
A streaming message may be aggregated into a single message if the underlying platform supports arbitrarily large strings or buffers,
and a single message may be exposed to the processor as a streaming interface, emitting a single chunk.
This flexibility allows generic processors to be implemented more easily without needing to distinguish between streaming and single-message protocols.

   

    
    Coalescing messages enables simpler processor logic. 
    For example, a processor that performs string replacement on incoming messages may otherwise need to implement both streaming and non-streaming handling.


    
Instead, the runner can convert incoming messages to a streaming form with one chunk, or aggregate streaming chunks into a single message.

    
It is advised that those processors, before forwarding the message, look at the length of each message before determining whether or not this message should be a single message or a streaming message.

   

   
The following sections detail the runner startup flow and describe the expected interactions between runners and the orchestrator during initialization and execution.

   
6.3.1. Pipeline Configuration

   
A runner is defined as an instance of a runner class.
Every runner MUST be linked to its programming context with rdfc:handlesSubjectsOf, and MUST declare how it is reached, so that a mainStream can be established over which it identifies itself with its own IRI.
The object of rdfc:handlesSubjectsOf links runners and processors to a context term. This often refers to the programming language.

   
The runner class determines how that connection is established. Two classes are currently defined:

   
   
Both are described in full in § 6.3.2 Connecting Flow.

   
Each context term is related to a SHACL shape, this specifies the incoming data that the runner can use to start the processors.

   
The full ontology can be found online.

   

    
The NodeRunner, for example, is a program that starts JavaScript processors with Node. It is defined as follows:



rdfc:NodeRunner a rdfc:CommandRunner;
rdfc:handlesSubjectsOf rdfc:jsImplementationOf;
rdfc:command "npx js-runner".

# Note that rdfc:jsImplementationOf is already defined by RDF-Connect as follows
sds:implementationOf rdfs:subPropertyOf rdfs:subClassOf.
rdfc:jsImplementationOf rdfs:subPropertyOf sds:implementationOf.

# Shape that a Js Processor should fulfil;
[ ] a sh:NodeShape;
  # We target it with jsImplementationOf
  sh:targetSubjectsOf rdfc:jsImplementationOf;
  sh:property [
    sh:path rdfc:file;
    sh:name "file";
    sh:minCount 1;
    sh:maxCount 1;
    sh:datatype xsd:string;
  ], [
    sh:path rdfc:class;
    sh:name "clazz";
    sh:maxCount 1;
    sh:datatype xsd:string;
  ].


    
This way, rdfc:jsImplementationOf is a predicate declared only for JavaScript processors.
And a shape is linked with that predicate, runners can expect a file location and a class name to start the JavaScript processors.

   

   
6.3.2. Connecting Flow

   
Every runner MUST establish a bidirectional message stream with the orchestrator, referred to as the mainStream, using the RPC.connect method, and MUST learn the IRI that uniquely identifies it within the pipeline.
The runner class determines how the runner obtains that IRI and how the underlying transport is set up; in both cases below the runner is the gRPC client and the orchestrator’s gRPC server accepts the connection.

   
Each rdfc:CommandRunner is started by the orchestrator using a command defined in the pipeline via rdfc:command.
The orchestrator appends two arguments to this command: the URL of the orchestrator’s Protobuf server and the IRI that uniquely identifies the runner.
Upon startup, the runner MUST connect to the orchestrator using the provided URL via the RPC.connect method.

   
Each rdfc:TcpRunner is already running independently of the orchestrator and MUST listen for plain TCP connections on the host:port given by rdfc:grpc.
The orchestrator connects to it directly over TCP and writes the runner’s IRI on the raw socket.
The runner MUST read that IRI from the socket and MUST then use the same socket as the transport for its own RPC.connect call.
The orchestrator hands its end of the socket to the same gRPC server that handles RPC.connect for rdfc:CommandRunners, so from the orchestrator’s perspective it is an ordinary incoming connection.

   
Once connected, the runner MUST send an RPC.identify message, identifying itself with the provided IRI.
The orchestrator then sends an RPC.pipeline message containing the complete, expanded pipeline definition.
The runner MAY ignore this message.

   
Runners MAY initiate a separate log stream using the RPC.logStream method to transmit log messages to the orchestrator.

   
Each log message sent by a runner MUST be an instance of RPC.LogMessage transmitted over the logStream.

   
Each RPC.LogMessage MUST include:

   
   
The level field MUST correspond to one of the log levels supported by Winston error, warn, info, http, verbose, debug and silly.

   
Entities are labels included when logging the message to standard output (stdout).
Aliases are additional labels that can be used for filtering or correlation but are not displayed in stdout.

   
Entities and aliases MAY be full URIs. The orchestrator SHOULD shorten URIs using the prefix mappings defined in the pipeline configuration.

   
6.3.3. Instantiating Processors

   
After initialization, the orchestrator may send multiple RPC.proc messages to instruct the runner to start specific processors.
Each message includes a processor IRI, a configuration object and an argument object.
Both the configuration and arguments are provided as JSON-LD strings.
The configuration object contains the arguments as defined by the context term following the section Mapping SHACL to Configuration Structures.
The arguments are constructed based on a SHACL shape defined for the processor type.

   
Runners MAY parse these JSON-LD payloads and transform known constructs into idiomatic equivalents.
For example, reader and writer objects are represented as JSON-LD values with: an @id field (containing the channel IRI), and an @type field indicating either https://w3id.org/rdf-connect#Reader or https://w3id.org/rdf-connect#Writer.
Runners are RECOMMENDED to replace these values with appropriate typed objects in the target environment.

   
When a processor has been fully initialized, the runner MUST send an RPC.initialized message, indicating success or failure.
If any runner signals an error during initialization, the orchestrator MUST abort the pipeline execution.

   
Once all processors are successfully initialized, the orchestrator sends an RPC.start message, instructing the runner to start the processors.

   
After all processors complete their execution, the runner MUST gracefully close the mainStream to signal completion.

   
6.3.4. Handling messages

   
Apart from starting processors, the runner also acts as a mediator that makes sure the correct messages are sent to the correct processors.
The orchestrator MAY send any number of RPC.msg or RPC.streamMsg messages.

   
6.3.4.1. Receiving normal messages

   
When an RPC.msg is received, the runner MUST deliver the message to the appropriate processor, using the channel IRI to determine the correct target.
Message routing can follow either a push or pull model depending on the language environment.

   
Runners SHOULD coerce or transform messages into a different representation, including converting a normal message into a streaming message with a single chunk or converting a short streaming message into a normal message.
These transformations SHOULD respect the preferences of the processor and the runner’s internal design constraints.

   
After the message has been handled by the processor, the runner MUST send a RPC.processed message back to the orchestrator.

   
6.3.4.2. Receiving streaming messages

   
When an RPC.streamMsg is received, the runner MUST establish a streaming channel by invoking the RPC.receiveStreamMessage method with the provided stream ID.
This initiates a stream of chunks from the orchestrator.

   
The runner MUST forward these chunks to the appropriate processor.
After each chunk, the runner MUST send a SendingStreamControl message with an incrementing streamSequenceNumber.

   
When the stream is closed, the runner may choose to coalesce all chunks into a single message and forward this to the processor.
After this optional message is handled, the runner MUST send a RPC.processed message back to the orchestrator.

   
6.3.4.3. Sending messages

   
Runners MUST also support outbound communication from processors.
While runners MAY omit support for certain advanced features (such as streaming output), a full implementation is strongly encouraged.

   
To send a normal message, the runner uses the RPC.msg method on the mainStream.
The processor is only allowed to send a new message after receiving a RPC.processed message.

   
To send a streaming message, the runner first initiates the RPC.sendStreamMessage method, which returns a new stream.
The orchestrator responds with ReceivingStreamControl message, containing the expected next chunk to receive (starting from 0).

   
The message is considered complete when the runner closes the stream and the runner received a RPC.processed message.
The processor is only allowed to send a new message after receiving a RPC.processed message.

   
6.3.4.4. Channel Closure

   
Processors MAY indicate that a given channel is closed (i.e., no further messages will be sent).
The runner MUST propagate this information to the orchestrator via an RPC.close message.

   
Similarly, when the orchestrator sends an RPC.close message for a channel, the runner MAY respond by closing or invalidating the corresponding data stream in the processor.

   
6.4. Processor

   
A processor SHOULD be implemented as generically as possible, while maintaining practical usefulness.
In most cases, a processor acts as a lightweight wrapper around an existing library that performs a useful task.

   
In most programming languages, a processor SHOULD implement an interface exposing three methods: initialize, transform, and produce.

   
   
A processor SHOULD have access to a logger instance that is integrated with the RDF-Connect logging system, enabling unified and aggregated logging across the pipeline.

   
A transforming processor is considered finished when all of its input channels are closed, and it MUST close all output channels.

   
Each processor MUST correctly propagate close events, so no stale channels exist.

   
6.5. Pipeline

   

🚧 This section is a work in progress and will be expanded soon.


   
7. Ontology Reference

   
The RDF-Connect ontology provides the terms used in RDF pipeline definitions. See the full RDF-Connect Ontology for details.

   
  

  

   

    Conformance

   

            Conformance requirements are expressed with a combination of descriptive assertions and RFC 2119 terminology.
            The key words “MUST”, “MUST NOT”, “REQUIRED”, “SHALL”, “SHALL NOT”, “SHOULD”, “SHOULD NOT”, “RECOMMENDED”, “MAY”, and “OPTIONAL”
            in the normative parts of this document
            are to be interpreted as described in RFC 2119.
            However, for readability,
            these words do not appear in all uppercase letters in this specification.

        

   

            All of the text of this specification is normative
            except sections explicitly marked as non-normative, examples, and notes. [RFC2119]

        

   

            Examples in this specification are introduced with the words “for example”
            or are set apart from the normative text with class="example", like this:

        

   

            This is an example of an informative example.
        

   

            Informative notes begin with the word “Note”
            and are set apart from the normative text with class="note", like this:

        

   

            Note, this is an informative note.

  


  
References

  
Normative References

  

   
[RFC2119]
   
S. Bradner. Key words for use in RFCs to Indicate Requirement Levels. March 1997. Best Current Practice. URL: https://datatracker.ietf.org/doc/html/rfc2119
  

  
Non-Normative References

  

   
[SHACL]
   
Holger Knublauch; Dimitris Kontokostas. Shapes Constraint Language (SHACL). URL: https://w3c.github.io/data-shapes/shacl/
  

  
Issues Index

  

   

🚧 This section is a work in progress and will be expanded soon.
 ↵