Core Concepts

FsShelter is a library for building stream-processing pipelines in F#. A pipeline is a directed graph where data flows from sources through processors, each transforming or routing messages along the way.

You don't need to know anything about Apache Storm to use FsShelter — this page covers the core ideas from scratch.

What is stream processing?

Imagine a data pipeline: events arrive continuously (sensor readings, log entries, user clicks), get transformed, filtered, aggregated, then stored or forwarded. Instead of collecting events into a batch and processing them later, stream processing handles each event as it arrives — giving you low latency and the ability to react in real time.

A stream-processing topology is a directed acyclic graph (DAG) of processing steps. FsShelter lets you define these graphs with F# types and functions, then run them in-process with high-performance message passing.

Key terminology

Topology schema

While most stream-processing systems use dynamically typed messages, FsShelter uses F# discriminated unions to statically type every stream. Mistakes and inconsistencies between producers and consumers are caught at compile time, not at runtime.

Every DU case becomes a distinct stream. The fields of each case become the tuple's payload:

type BasicSchema = 
    | Original of int
    | Incremented of int

It is often handy to define a record type shared across streams:

type Number = { X:int; Desc:string }

type RecordSchema = 
    | Original of int
    | Described of Number
    | Translated of Number

Joining tuples from multiple streams is also supported:

type RecordsSchema = 
    | Original of Number
    | Doubled of Number * Number

Generic or nested schemas work too. You can define a base schema and extend it:

type SimpleSchema = 
    | Value of int
    | Transformed of int

type NestedSchema<'a> = 
    | Named of string
    | [<NestedStream>] Nested of 'a

A topology using Topology<NestedSchema<SimpleSchema>> combines both sets of streams. For more about schema structure, grouping expressions, and serializer details, see Schema reference.

Components

FsShelter components are plain F# functions — no base classes, no interfaces, no framework inheritance.

A spout returns an option: Some tuple to emit, None when there's nothing to emit right now:

// numbers spout - produces a message
let numbers source = Some(BasicSchema.Original(source()))

A bolt receives a tuple and optionally emits downstream. Pattern matching on the schema DU gives you exhaustive, type-safe dispatch:

// add one bolt - transforms and emits
let addOne (input, emit) = 
    match input with
    | BasicSchema.Original(x) -> BasicSchema.Incremented(x + 1)
    | _ -> failwithf "unexpected input: %A" input
    |> emit

// terminating bolt - consumes without emitting
let logResult (info, input) = 
    match input with
    | BasicSchema.Incremented(x) -> info (sprintf "%A" x)
    | _ -> failwithf "unexpected input: %A" input

The arguments to your component functions are wired up when you define the topology — you choose exactly what each function receives. There's no global state or magic injection.

Async components

When your spout or bolt needs to perform I/O (database queries, HTTP calls, file access), you can use the async variants. Instead of returning a value directly, async components return a Task:

open System.Threading.Tasks

// async spout - produces a message via async I/O
let asyncNumbers source : Task<BasicSchema option> =
    task {
        let value = source()
        return Some(BasicSchema.Original value)
    }

// async bolt - transforms via async I/O
let asyncAddOne (input, emit) : Task<unit> =
    task {
        match input with
        | BasicSchema.Original(x) -> BasicSchema.Incremented(x + 1) |> emit
        | _ -> failwithf "unexpected input: %A" input
    }

The type signatures are:

• AsyncNext<'a, 't> — an async spout function: 'a -> Task<'t option>
• AsyncConsume<'a> — an async bolt function: 'a -> Task<unit>

Async components run on async executors — the Disruptor thread awaits each Task, allowing non-blocking I/O without dedicating extra threads. All configuration (withParallelism, withExecutors, etc.) and delivery guarantees (reliable/unreliable) work identically.

Topology DSL

FsShelter provides a computation expression for defining topologies. You declare components and connect them with arrows:

• --> connects two components on a stream (unanchored — fire and forget)
• ==> connects with anchoring (for reliable delivery — see below)
• Shuffle.on distributes tuples across instances by hashing
• Group.by routes tuples with the same key to the same instance (affinity)
• withParallelism n runs n instances of a component

Here's a minimal topology:

let source = 
    let rnd = Random()
    fun () -> rnd.Next(0, 100)

open FsShelter.DSL
open FsShelter.Multilang

let sampleTopology = 
    topology "Sample" { 
        let s1 = 
            numbers
            |> Spout.runUnreliable (fun log cfg -> source) ignore
        let b1 = 
            addOne
            |> Bolt.run (fun log cfg tuple emit -> (tuple, emit))
            |> withParallelism 2
        let b2 = 
            logResult
            |> Bolt.run (fun log cfg tuple emit -> ((log LogLevel.Info), tuple))
            |> withParallelism 2
        
        yield s1 --> b1 |> Shuffle.on BasicSchema.Original
        yield b1 --> b2 |> Shuffle.on BasicSchema.Incremented
    }

The lambda arguments for the run methods construct the arguments passed to your component functions:

• log is a logging factory
• cfg is the runtime configuration
• tuple is the incoming message (schema DU instance)
• emit is a function to emit a new tuple downstream

log and cfg are curried once at startup. The tuple and emit arguments arrive per-message.

The async variants have the same structure — just swap Spout.run* for Spout.run*Async and Bolt.run for Bolt.runAsync:

// Async spout (reliable)
let s1 = asyncNumbers
         |> Spout.runReliableAsync (fun log cfg -> source) (fun s -> ack, nack) ignore

// Async spout (unreliable)
let s1 = asyncNumbers
         |> Spout.runUnreliableAsync (fun log cfg -> source) ignore

// Async bolt (auto-ack on success)
let b1 = asyncAddOne
         |> Bolt.runAsync (fun log cfg tuple emit -> (tuple, emit))

// Async bolt (always nack — terminator)
let b2 = asyncLog
         |> Bolt.runTerminatorAsync (fun log cfg tuple _ -> (log, tuple))

Sync and async components can be mixed freely in the same topology. The hosting runtime detects the component type and uses the appropriate executor.

Reliable vs unreliable delivery

FsShelter supports two delivery modes:

Unreliable (fire and forget): Tuples are emitted and processed but not tracked. If a bolt fails, the message is lost. This is the simplest mode and has the lowest overhead. Use it when losing an occasional message is acceptable (metrics, logging, non-critical analytics).

Reliable (guaranteed delivery): Every tuple emitted by a spout is tracked through the entire DAG. When all downstream bolts finish processing, the spout receives an ack. If any bolt fails or processing takes too long, the spout receives a nack and can retry. This is implemented via an XOR-tree tracking algorithm (see Acker algorithm for details).

In the DSL, the arrow operator determines the mode:

• ==> creates an anchored stream (starts or continues tracking)
• --> creates an unanchored stream (fire and forget)

See Word Count for an unreliable example and Guaranteed for a reliable example.

Running a topology

FsShelter topologies run entirely in-process — no external infrastructure required. The runtime uses Disruptor ring buffers for high-performance, low-latency message passing between components.

let shutdown = Hosting.run myTopology

This starts all components, wires up the message routing, and returns a function you call to shut down cleanly. For production use with auto-restart and logging, see Running Topologies.

If you need to deploy to an Apache Storm cluster for horizontal scale-out, the FsShelter.Multilang package provides Storm integration. See Word Count: Deploying to Storm for details.

Visualizing topologies

FsShelter can export any topology as a GraphViz DOT graph:

sampleTopology |> DotGraph.writeToConsole

Pipe the output through dot to generate SVG or PNG:

dotnet run -- graph | dot -Tsvg -o topology.svg

Next steps

• *Word Count* — Complete tutorial: schema, components, topology, graph export
• *Guaranteed delivery* — Reliable spouts with ack/nack and anchoring
• *Schema reference* — Grouping expressions, flattening, serializer details
• *Running Topologies* — Entry points, configuration, diagnostics
• *Routing* — How tuples are distributed across component instances
• *Architecture* — Internal runtime structure: tasks, executors, channels
• *Message Flow* — End-to-end processing scenarios with sequence diagrams
• *Acker Algorithm* — XOR-tree tracking for guaranteed delivery