Skip to content

StreamingJSON processors

pkg/processor/streaming_json.go defines processors and helpers for handling large JSON arrays without loading the whole array into memory. They are the JSON-streaming counterpart to split — Split works on in-memory slices, while StreamingJSONProcessor streams elements as they arrive from an io.Reader.

Processors in this file

NewStreamingJSONProcessor

Streams a JSON array from the exchange body, decoding one element at a time and invoking an inner processor for each.

inner := processor.NewPipeline(/* per-element steps */)
streamer := processor.NewStreamingJSONProcessor(inner)
// Used in code; not currently surfaced as a top-level YAML key.

Constructor: NewStreamingJSONProcessor(inner engine.Processor) *StreamingJSONProcessor.

Body type accepted Handling
StreamBody (exchange.IsStreamBody is true) Read directly via the stream reader.
[]byte Wrapped in strings.NewReader.
string Wrapped in strings.NewReader.
io.Reader Used directly.
anything else Returns unsupported body type %T.

How it works:

  1. Creates a json.Decoder over the body reader.
  2. Reads the opening [ token. Returns an error if the body is not a JSON array.
  3. For each element:
  4. Decodes the raw JSON into json.RawMessage.
  5. Creates a child exchange via exchange.New(ctx).
  6. Sets the child body to the raw element bytes.
  7. Adds headers HelixStreamingIndex (0-based string), HelixStreamingParentID (parent's exchange ID).
  8. Copies parent headers into the child, except HelixStreamBody and Content-Length.
  9. Invokes the inner processor with the child exchange.
  10. Appends the child's resulting body to a results slice.
  11. After the closing ], writes the collected results slice to the parent body and sets HelixStreamingCount to the element count.
  12. Honors ctx.Done() between elements for cooperative cancellation.

Notes and gotchas:

  • The processor still collects processed elements into memory (so the parent's body becomes a slice). For pure pass-through fan-out without retention, the inner processor should publish to its target and the caller can ignore the result slice.
  • Only top-level JSON arrays are streamed. Nested arrays are returned as RawMessage and are not unwrapped.
  • Errors decoding a single element abort the whole stream — there is no per-element skip mode.

NewStreamingJSONWriter

A streaming output helper. Writes JSON array elements one by one to an io.Writer, suitable for HTTP response bodies or file output of large result sets.

w := processor.NewStreamingJSONWriter(httpResp)
for _, item := range items {
    if err := w.WriteElement(item); err != nil {
        return err
    }
}
return w.Close()
Method Description
WriteElement(v any) error Writes the next element. Emits [ before the first element and , between subsequent ones.
Close() error Writes the closing ]. If WriteElement was never called, writes an empty [].
Count() int Returns the number of elements written so far.

Pitfalls:

  • The writer is not safe for concurrent use; serialize calls if multiple goroutines produce elements.
  • Close() must be called even if no elements were written — otherwise the consumer sees an incomplete document.

Examples

End-to-end: stream a large JSON array, publish each element, and respond with element count.

// In a custom Helix processor:
proc := processor.NewStreamingJSONProcessor(
    processor.NewPipeline(
        // per-element steps:
        myKafkaPublisher,
    ),
)

A processor that produces a streaming JSON response:

func handle(ctx context.Context, ex *exchange.Exchange) error {
    w := processor.NewStreamingJSONWriter(ex.Out().BodyAsWriter())
    for _, row := range rows {
        if err := w.WriteElement(row); err != nil {
            return err
        }
    }
    return w.Close()
}

Source

pkg/processor/streaming_json.go

See also

  • split — split a slice-shaped body into per-element exchanges (in-memory).
  • dataformat — JSONMarshal/JSONUnmarshal for whole-document handling.