stream

package module
v0.1.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 5, 2026 License: MIT Imports: 1 Imported by: 1

README

stream

stream defines a small publisher–subscriber messaging interface and the related types used to compose event-driven data-pipelines in Go.

The root package is interfaces only — the contract. Concrete building blocks (a NATS JetStream transport, an in-memory test double, a metadata decorator, a subject implementation, …) live in sub-packages and can be mixed and matched, or replaced entirely by your own implementations.

import "github.com/jrgensen/stream"
  • Module path: github.com/jrgensen/stream
  • Go: 1.25+

Why interfaces?

Reading remote messages from a broker (NATS) or persisting state in a database is fundamentally different from the event-passing you do inside a service. The stream interfaces let you keep those concerns in separate layers, each of which only depends on the contract — never on a concrete transport or codec.

  NATS               |
 ------------------  |  Knows about NATS *and* stream. Data abstraction layer.
  nats client        |
 ------------------  —
  stream router      |
 ------------------  |  Knows only about stream interfaces.
  stream handlers    |
 ------------------  —
  db handlers        |
 ------------------  |  Knows about MongoDB *and* stream. Data abstraction layer.
  MongoDB            |

Rules of thumb this design encourages:

  • Decode and validate at the source boundary. Once a message is inside the pipeline it is trusted; internal handlers should not re-parse transport encodings such as JSON.
  • Business logic works on stream.Message, not on transport types. The router and handlers must not know about NATS or the wire codec.
  • Encoding for storage is the storage handler's job. The core packages know nothing about the databases that implement their interfaces.

The contract

Everything below lives in the root stream package.

Message / MutableMessage

Message is a read-only view of a single event on the stream.

type Message interface {
    Subject() Subject          // where the event was published
    Time() time.Time           // when it occurred / was submitted
    Sequence() uint64          // monotonic position within the stream
    Body(dst interface{}) error // decode the value into dst (a struct pointer)
    Meta(dst interface{}) error // decode optional metadata into dst
    RawBody() interface{}       // the undecoded value
    RawMeta() interface{}       // the undecoded metadata
}

Implementor notes:

  • Body/Meta copy the payload into dst. dst is borrowed only for the duration of the call. How the copy happens is up to you — decode a stored byte slice, deep-copy an in-memory struct, etc.
  • RawBody/RawMeta expose the underlying representation for consumers (such as decorators) that want to move bytes around without decoding. Document the concrete type you return; existing helpers accept json.RawMessage, []byte, string, or an arbitrary value.

MutableMessage adds the setters used while a message is being built:

type MutableMessage interface {
    Message
    SetSubject(Subject)
    SetBody(interface{}) error
    SetMeta(interface{}) error
    SetTime(time.Time) error
}

MessageFunc is the factory that produces new mutable messages for a subject. Every Publisher exposes one so callers construct messages that are compatible with the underlying transport:

type MessageFunc func(Subject) MutableMessage
Handling messages
type MessageHandler interface {
    HandleMessage(Message) error
}

MessageHandlerFunc is the idiomatic function adapter (à la http.HandlerFunc):

var h stream.MessageHandler = stream.MessageHandlerFunc(func(m stream.Message) error {
    // ...
    return nil
})
Publish / Subscribe
type Publisher interface {
    Publish(msg Message) error   // may be sync or async; must be thread-safe
    MessageFunc() MessageFunc    // factory for transport-compatible messages
}

type Subscriber interface {
    // Subscribe registers h for the given subjects and returns a Subscription.
    // Call Close() on the subscription to unsubscribe.
    Subscribe(subjects []Subject, h MessageHandler) (Subscription, error)
}

type Subscription interface {
    Close() error
}

// Stream is the full duplex contract: a Publisher + Subscriber you can Close.
type Stream interface {
    Publisher
    Subscriber
    Close() error
}

PublisherFunc adapts a bare func(Message) error to Publisher (its MessageFunc panics — use it only where a factory is not required).

Pipeline metadata

These optional interfaces help wire a graph of publishers and subscribers, and let handlers react to stream lifecycle events.

// Producer advertises the subjects a component may write to.
type Producer interface {
    Produces() []Subject
}

// Consumer is a MessageHandler that advertises the subjects it wants.
type Consumer interface {
    MessageHandler
    Consumes() []Subject
}

// CatchupListener is called once a handler has replayed history up to the
// point the store was at when reading began — useful for de-duplication and
// other "we are now live" optimizations.
type CatchupListener interface {
    CaughtUp()
}
Subject

A Subject addresses events and is composed of a domain and a type.

type Subject interface {
    Domain() string       // the domain part
    Type() string         // the type part
    Subject() string      // canonical "domain.type" string
    Parts() []string      // split on "."
    Match(string) bool    // pattern match (supports "*" wildcards)
}

The canonical implementation lives in subject.

Composing programs

Internal services are built as data-pipelines. An input (usually from NATS) is consumed and processed by one or more subscribers. Each subscriber transforms data and publishes its output to a shared in-process Stream; other handlers subscribe to that stream and repeat the process.

  [handler] --||-- [handler] --||-- [handler]

The || between handlers is a routing component (see xstream) responsible for delivering messages from publishers to subscribers. When the originating NATS event is published, only the first subscriber is known; that subscriber has no knowledge of downstream consumers of the messages it produces.

Packages

Package Kind Description
stream (root) interfaces The messaging contract and function adapters.
subject implementation Canonical Subject (StringSubject, FromStr, FromParts).
jetstream implementation Stream backed by NATS JetStream.
metatagger implementation Publisher decorator that applies default metadata.
xstream implementation Message multiplexer/router wiring Consumers to a Stream.
caughtup implementation Sentinel Message signalling catch-up state.
streamtest implementation In-memory Publisher/Message doubles for tests.

Implementing your own transport

To add a new backend, implement stream.Stream:

  1. MessageFunc() — return a factory that builds MutableMessage values your Publish can serialize. Keeping construction and publishing in the same implementation lets you attach whatever bookkeeping (IDs, versions) you need.
  2. Publish(Message) — serialize and send. Must be safe for concurrent use.
  3. Subscribe([]Subject, MessageHandler) — deliver matching messages to the handler and return a Subscription whose Close tears the subscription down.
  4. Close() — stop accepting new messages and unsubscribe everything.

Use jetstream as a reference and streamtest to exercise handlers without any external infrastructure.

License

Released under the MIT License.

Documentation

Overview

Package stream defines a publisher—subscriber messaging interface and related types used to compose data-pipelines.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type CatchupListener 

type CatchupListener interface {
	// CaughtUp will be called when the subjects that the handler is subscribed
	// to are read up until the state that the data store had before beginning
	// to read events of the event store. This allows the programmer to
	// implement optimizations that use this knowledge to for example discard
	// duplicate events for a given identifier.
	CaughtUp()
}

CatchupListener is an optional interface that can be useful to implement on types that are used in a router context. i.e. a Consumer.

type Consumer 

type Consumer interface {
	MessageHandler

	// Consumes returns a vector that of subjects.
	Consumes() []Subject
}

A Consumer represents interest in a set of subjects.

type Message 

type Message interface {
	// Subject() returns the subject to which this event was published.
	Subject() Subject

	// Time returns the time when this message was submitted to the stream or
	// when the event occurred.
	Time() time.Time

	// Sequence
	Sequence() uint64

	// Body copies the value data onto the type pointed to from dst (dst is
	// expected to be a struct pointer).
	//
	// If the type of the value is a pointer to a struct its content is copied
	// to a new struct of the same type, and dst is pointed to this new value.
	// If the concrete types don't match Value returns an error. However, other
	// mechanisms to store and copy the value may exist - that don't have the
	// same semantics of requiring types to match. This may be if the
	// implementation stores a byte slice, data is is held in a intermediate
	// form, is decoded onto the destination type.
	//
	// dst should be considered a mutable pointer to some T that the Value
	// method borrows for the duration of the call.
	Body(dst interface{}) error

	// Meta is optional meta data about the message. It follows the same
	// semantic as the Body() method.
	Meta(dst interface{}) error

	RawBody() interface{}
	RawMeta() interface{}
}

Message contains meta-data and the value for a given event on the event-stream. This interface is a read-only interface to an event.

type MessageFunc 

type MessageFunc func(Subject) MutableMessage

type MessageHandler 

type MessageHandler interface {
	// HandleMessage accepts a Message. With the Message type and channel we can can
	// deterime type of the Message body—and decode the data to its type.
	HandleMessage(Message) error
}

A Handler receives events of an event stream.

type MessageHandlerFunc 

type MessageHandlerFunc func(Message) error

MessageHandlerFunc is an adapter to allow the use of ordinary functions as Message handlers. If f is a function with the appropriate signature, HandlerFunc(f) is a Handler that calls f.

func (MessageHandlerFunc) HandleMessage 

func (f MessageHandlerFunc) HandleMessage(msg Message) error

HandleMessage calls f(msg).

type MutableMessage 

type MutableMessage interface {
	Message
	SetSubject(Subject)
	SetBody(interface{}) error
	SetMeta(interface{}) error
	SetTime(time.Time) error
}

type Producer 

type Producer interface {
	// Produces defines a set of channels to which it may write to. it returns
	// a vector of subjects.
	Produces() []Subject
}

Producer is a meta-data interface that helps creating a graph of publisher-subscribers.

type Publisher 

type Publisher interface {
	// Publish publishes the provided message to a given subject.
	//
	// Publish can be async or synchronous.
	// Publish should be thread-safe.
	Publish(msg Message) error

	// MessageFunc returns a MessageFunc used by the Publisher
	MessageFunc() MessageFunc
}

Publisher is an interface that wraps the Publish method.

type PublisherFunc 

type PublisherFunc func(Message) error

PublisherFunc type is an adapter to allow the use of ordinary functions as Publishers.

func (PublisherFunc) MessageFunc 

func (f PublisherFunc) MessageFunc() MessageFunc

satisfy Publisher interface

func (PublisherFunc) Publish 

func (f PublisherFunc) Publish(msg Message) error

Publish calls f(g)

type Stream 

type Stream interface {
	Publisher
	Subscriber

	// Close stops publisher from accepting messages and unsubscribes all
	// subscribers.
	Close() error
}

Stream defines a publisher—subscriber messaging interface.

type Subject 

type Subject interface {
	// Domain returns the 'domain' part of the subject.
	Domain() string

	// Type returns the 'type' part of the subject.
	Type() string

	// Subject prints the canonical string representation of a Subject.
	Subject() string

	Parts() []string
	Match(string) bool
}

Subject may be composed of a 'domain' and a 'type'.

type Subscriber 

type Subscriber interface {
	// Subscribe will subcribe handler for the given subject it's interested
	// in. It returns a Subscription or an error.
	//
	// To unsubscribe simply call Close() on the subscription.
	Subscribe(subjects []Subject, h MessageHandler) (Subscription, error)
}

Subscriber is an interface that wraps the Subscribe method.

type Subscription 

type Subscription interface {
	// Close stops unsubscribes subscriber.
	Close() error
}

Subscription represents interest in a given subject.

Source Files

View all Source files

Directories

Path Synopsis
Package metatagger provides a stream.Publisher decorator that applies default metadata values to every message before delegating publication to a wrapped Publisher.
 Package metatagger provides a stream.Publisher decorator that applies default metadata values to every message before delegating publication to a wrapped Publisher.
Package subject provides the canonical StringSubject implementation of the stream.Subject interface.
 Package subject provides the canonical StringSubject implementation of the stream.Subject interface.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.