defoperation FahrenheitToCelcius, in: fahrenheit, out: celcius do
  defcb react(fahrenheit) do
    ((fahrenheit - 32) * (5 / 9)) ~> celcius
  end
end

This operation defines a single callback, react which converts its argument fahrenheit to celcius. The resulting value is emitted to the celcius port. At runtime, the strategy of this operation would call react, after which the emitted value is sent to operations downstream of FahrenheitToCelcius.

defoperation Average, in: value, out: current do
  state_struct total: 0, count: 0

  defcb react(value) do
    total <~ ~f{total} + value
    count <~ ~f{count} + 1
    ~f{total} / ~f{count} ~> current
  end
end

Operations can maintain state. This operation tracks the average of all the values it receives at runtime. It does this by counting the amount of values it receives (in the counter field) and the sum of all these values (total). When Average recieves a new value, it updates its fields, calculates the current average and emits this value to its successors.

Operations can be composed into workflows. These workflows are built in a simple textual language where operations are linked to other operations or to other, nested, workflows. The example shown here defines a data processing pipeline which provides crowdsourced videos of participants in a running competition based on nearby spectators. Its graphical representation is shown below.

workflow do
  node(TCPSource, args: {"localhost", 4555}) ~> node(Parser) ~> joiner.participants
  node(TCPSource, args: {"localhost", 4556}) ~> node(Parser) ~> joiner.spectators

  node(Join, as: joiner, args: [windows_size: 5])
  ~> node(Rate)
  ~> node(Select)
  ~> node(Publish)
end

defstrategy ImmutableLocal do
  defhook deploy(args) do
    Nodes.on_all_worker_cores(fn ->
      create_worker(
        fn -> call_component(:init, [args]).state end,
        :worker,
        :local
      )
    end)
    |> Map.new()
  end

  defhook deliver(msg, _) do
    send(Enum.random(deployment()[Nodes.self()]), msg)
  end

  defhook process(msg, state, :worker) do
    result = call_component(:react, state, nil, [msg])
    emit(result.emit)
    state
  end
end

This strategy handles operations with an immutable state. It does this by creating a worker for every core on every available machine in the cluster; any data emitted by a predecessor of the operation will will be sent to a worker on the local machine, which will call the react callback of the operation.

The workers of this strategy are created in the deploy hook, which is called by the runtime system when an operation needs to be deployed over the cluster. Any data returned by this hook is stored inside the deployment, which is automatically available in all other hooks. The deliver hook is invoked by the runtime when a predecessor of the operation emits data. This strategy selects a worker on the current node and forwards the data to this worker. Finally, the process hook is called when a worker receives data. In this case, the worker executes the react callback of the operation and specifies that the emitted data should be forwarded to all the successors of this operation.

defstrategy ShuffledImmutable, extends: ImmutableLocal do
  defhook send(msg, _) do
    node = Nodes.workers() |> Enum.random()
    send(Enum.random(deployment()[node]), msg)
  end
end

Strategies can be created based on other strategies. In the example shown here, a new strategy is created based on the ImmutableLocal strategy shown in the previous example. Unlike ImmutableLocal, this strategy does not attempt to keep work local, instead, it selects a random machine in the cluster to process the received data.