Mediators

On this page

Mediators form a critical part of Comunica's core architecture. They are responsible for selecting one actor from a bus, based on a given task.

While actors perform the actual logic, they never communicate with each other directly. Instead, they always communicate through mediators, to reduce coupling between actors. If a different actor selection technique is needed, a different mediator can be configured without having to change any buses or actors.

Mediator implementations

Name	Package	Description
Race	`@comunica/mediator-race`	Picks the first actor that resolves its test.
Number	`@comunica/mediator-number`	Mediates over a single number field. It can either choose the actor with the maximum or with the minimum value.
All	`@comunica/mediator-all`	Special mediator that runs all actors that resolve their test in parallel.
Combine Pipeline	`@comunica/mediator-combine-pipeline`	Special mediator that goes over all actors in sequence and forwards I/O. This requires the action input and the actor output to be of the same type.
Combine Union	`@comunica/mediator-combine-union`	Special mediator that takes the union of all actor results.

Mediator types

Comunica contains several packages named @comunica/mediatortype-* that expose interfaces that extend the IActorTest interface. These interfaces can be reused in different actors to indicate what properties can be mediated over.

The following mediator types are available:

Name	Package	Description
HTTP Requests	`@comunica/mediatortype-httprequests`	Number of HTTP requests required for an action.
Iterations	`@comunica/mediatortype-iterations`	Number of iterations that are needed for joining streams.
Priority	`@comunica/mediatortype-priority`	Priority of an actor, for example used for parsers and serializers in content negotiation.
Time	`@comunica/mediatortype-time`	Estimated time an action will take.

Configuring and using a mediator

Defining a component mediator parameter

The following components file shows how a mediatorJoin parameter is added to @comunica/actor-query-operation-join:

{
  "@context": [
    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-query-operation-join/^1.0.0/components/context.jsonld",
    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/bus-query-operation/^1.0.0/components/context.jsonld"
  ],
  "@id": "npmd:@comunica/actor-query-operation-join",
  "components": [
    {
      "@id": "caqoj:Actor/QueryOperation/Join",
      "@type": "Class",
      "extends": "cbqo:Actor/QueryOperationTypedMediated",
      "requireElement": "ActorQueryOperationJoin",
      "comment": "A comunica Join Query Operation Actor.",
      "parameters": [
        {
          "@id": "caqoj:mediatorJoin",
          "comment": "A mediator for joining Bindings streams",
          "required": true,
          "unique": true
        }
      ],
      "constructorArguments": [
        {
          "extends": "cbqo:Actor/QueryOperationTypedMediated/constructorArgumentsObject",
          "fields": [
            {
              "keyRaw": "mediatorJoin",
              "value": "caqoj:mediatorJoin"
            }
          ]
        }
      ]
    }
  ]
}

Instantiating a component mediator

The following config file shows how we instantiate an actor with a race mediator over the RDF join bus (@comunica/bus-rdf-join):

{
  "@context": [
    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/actor-query-operation-join/^1.0.0/components/context.jsonld",
    "https://linkedsoftwaredependencies.org/bundles/npm/@comunica/mediator-race/^1.0.0/components/context.jsonld"
  ],
  "@id": "urn:comunica:my",
  "actors": [
    {
      "@id": "config-sets:sparql-queryoperators.json#myJoinQueryOperator",
      "@type": "ActorQueryOperationJoin",
      "caqoj:mediatorJoin": {
        "@id": "config-sets:sparql-queryoperators.json#mediatorRdfJoin",
        "@type": "MediatorRace",
        "cc:Mediator/bus": { "@id": "cbrj:Bus/RdfJoin" }
      }
    }
  ]
}

Invoking a mediator in TypeScript

Invoking the mediator in a TypeScript actor implementation is done like this:

import { ActionContext, IActorTest, Mediator } from '@comunica/core';
import { ActorRdfJoin, IActionRdfJoin } from '@comunica/bus-rdf-join';
import { IMediatorTypeIterations } from '@comunica/mediatortype-iterations';

export class ActorQueryOperationJoin extends ActorQueryOperationTypedMediated<Algebra.Join> {

  public readonly mediatorJoin: Mediator<ActorRdfJoin,
  IActionRdfJoin, IMediatorTypeIterations, IActorQueryOperationOutput>;

  public constructor(args: IActorQueryOperationJoinArgs) {
    super(args, 'join');
  }

  public async testOperation(pattern: Algebra.Join, context: ActionContext): Promise<IActorTest> {
    return true;
  }

  public async runOperation(pattern: Algebra.Join, context: ActionContext): Promise<IActorQueryOperationOutput> {
    const myAction: IActionRdfJoin = { ... }; 
    return this.mediatorJoin.mediate(myAction);
  }
}

export interface IActorQueryOperationJoinArgs extends IActorQueryOperationTypedMediatedArgs {
  mediatorJoin: Mediator<ActorRdfJoin, IActionRdfJoin, IMediatorTypeIterations, IActorQueryOperationOutput>;
}