For-Command-Authors

For Command Authors

Audience: you build the commands that Users compose.

The rule of authoring: pick a pattern from patterns/ and supply only your algorithm. The pattern owns every hard part — stream wiring, channels, backpressure, cancellation, teardown — and hands you a plain function to fill in. You write what the command does to a line, never how lines move.

Read The Pipeline Model first if you haven’t. Authors should also read Concurrency & Lifecycle — it explains the teardown your commands inherit for free.

The pattern catalog

Each pattern captures one shape of “how input relates to output.” Find your command’s shape, and the pattern is chosen.

Decision shortcuts

• One line → one line? Map. Need a counter or running state? StatefulMap.
• Keep some lines? Filter. Need the previous line? StatefulFilter.
• Need to see all input first (sort, reverse, tail)? Accumulate (same type out) or Aggregate (one summary out).
• One line becomes many (or none)? Expand.
• “I’ve seen enough” (stop early)? Take / Head — and read why this is special.
• Just observing (write a copy, log)? Tap.

Writing a command

A command is a constructor that returns a Command. Supply the algorithm; that’s it.

func Grep(pattern string) gloo.Command[string, string] {
    return patterns.Filter(func(line string) (bool, error) {
        return strings.Contains(line, pattern), nil
    })
}

func Sort() gloo.Command[string, string] {
    return patterns.Accumulate(func(lines []string) ([]string, error) {
        sort.Strings(lines)
        return lines, nil
    })
}

func WordCount() gloo.Command[string, int] {
    return patterns.Aggregate(func(lines []string) (int, error) {
        return len(lines), nil
    })
}

Returning an error from your function propagates it down the stream and stops the pipeline — see how errors flow.

Stateful commands take a factory

A command is a value that may be reused across many pipelines, possibly concurrently. If a stateful command captured its state directly, two runs would share it. So StatefulMap and StatefulFilter take a factory called once per Execute — each run gets its own fresh state.

func Nl() gloo.Command[string, string] {
    return patterns.StatefulMap(func() func(string) (string, error) {
        n := 0 // fresh per Execute — never shared between pipelines
        return func(line string) (string, error) {
            n++
            return fmt.Sprintf("%d\t%s", n, line), nil
        }
    })
}

Never close over a mutable variable outside the factory — that’s the one way to make a command unsafe to reuse.

Early termination — Take

Take(n) emits the first n items and then stops the upstream — the SIGPIPE analogue. This is the one thing a Filter cannot do: a filter can drop items but can never say “I need nothing more,” so a filter-based head over a huge or infinite source reads everything. Take actively tears the producer down, so:

• Take(3) over an infinite source terminates instantly and leaks no goroutine, and
• a FileSource | Take(3) reads only a small prefix of the file.

Stages downstream of Take still run to completion, so Take(3) before a sort yields three sorted lines — exactly like seq inf | head -3 | sort. Take is the building block for head, grep -m, and sed q. (Head is a readability alias for Take.)

The mechanics behind this — how a downstream stop propagates upstream without leaking — are on Concurrency & Lifecycle.

Exotic commands — FuncCommand

When no pattern fits, drop to FuncCommand[In, Out] and build the output stream yourself. This is the only place an author touches rill. Two primitives produce a stream while inheriting clean teardown:

• GenerateFrom(ctx, in, producer) — produce output derived from an input stream. A downstream stop tears down both your producer and the upstream in, because you pass the stream, not a loose handle.

  func Doubler() gloo.Command[int, int] {
      return gloo.FuncCommand[int, int](func(ctx context.Context, in gloo.Stream[int]) gloo.Stream[int] {
          return gloo.GenerateFrom(ctx, in, func(_ context.Context, send func(int) bool, sendErr func(error)) {
              for item := range in.Chan() {
                  if item.Error != nil { sendErr(item.Error); return }
                  if !send(item.Value * 2) { return } // send==false → consumer stopped; return promptly
              }
          })
      })
  }
  

• WrapFrom(ch, in) — wrap a pure rill transform’s output channel, chaining teardown to in:

  out := rill.OrderedMap(in.Chan(), 1, fn)
  return gloo.WrapFrom(out, in)
  

For an origin producer (a Source, no upstream), use Generate(ctx, producer) and Wrap(ch) — same primitives without the upstream argument.

The golden rule: send returning false means the consumer has walked away — return promptly, exactly as a shell tool dies on SIGPIPE.

Sources

Build a Source to feed a pipeline. All filesystem access goes through afero.Fs — never os.Open. Tests use afero.NewMemMapFs().

Sources are cancellation-aware and handle lines far larger than bufio.Scanner’s 64 KB default (up to MaxLineSize, 1 GiB) via NewLineScanner, so minified JSON and long log lines never abort a pipeline.

Where things belong

Per the framework’s constitution: stream wiring lives in framework/, patterns in framework/patterns/, and each command in its own cmd-* module. Subprocess is a last resort — if an algorithm can be done in pure Go, it must be. New patterns require maintainer approval; exhaust the existing set first.

Next: Errors & Testing shows how to surface errors and how to test a command with in-memory streams.