stream

Documentation

Overview

Package stream offers an abstraction layer for stream-processing platforms, such as NATS streaming.

Index

• func PubWithContext(ctx context.Context, pub Pub) context.Context
• func SubWithContext(ctx context.Context, sub Sub) context.Context
• func WithContext(ctx context.Context, ps Stream) context.Context
• type MsgHandler
• type Pub
  • func PubFromContext(ctx context.Context) Pub
• type Stream
  • func FromContext(ctx context.Context) Stream
  • func Log(s Stream) Stream
  • func NopStream() Stream
  • func Recover(s Stream) Stream
  • func Stats(s Stream) Stream
  • func Trace(s Stream) Stream
• type Sub
  • func SubFromContext(ctx context.Context) Sub
• type Subscription
• type SubscriptionOption

Functions

func PubWithContext

func PubWithContext(ctx context.Context, pub Pub) context.Context

PubWithContext returns a copy of parent in which the `Pub` is stored

func SubWithContext

func SubWithContext(ctx context.Context, sub Sub) context.Context

SubWithContext returns a copy of parent in which the `Sub` is stored

func WithContext

func WithContext(ctx context.Context, ps Stream) context.Context

WithContext returns a copy of parent in which `Stream`, `Pub`, and `Sub` are stored

Types

type MsgHandler

type MsgHandler func(context.Context, []byte) error

type Pub

type Pub interface {
	// Publish publishes data to the channel ch
	// Publish(ctx context.Context, ch string, data []byte) error
	Publish(ctx context.Context, ch string, data []byte) error
}

Pub is the publish interface

func PubFromContext

func PubFromContext(ctx context.Context) Pub

PubFromContext returns a `Pub` instance associated with `ctx`, or the local `Pub` if no instance could be found.

type Stream

type Stream interface {
	Pub
	Sub

	// Start does the initialisation work to bootstrap a Stream adapter.
	// For example, this function may open a connection, start an event loop, etc.
	Start(ctx context.Context) error
	// Drain signals to the Stream client/server that inbound messages should
	// no longer be accepted, but outbound messages can still be delivered.
	Drain()
	// Close closes the client/server for both inbound/outbound messages
	Close() error
}

func FromContext

func FromContext(ctx context.Context) Stream

FromContext returns a `Stream` instance associated with `ctx`, or the local `Sub` if no instance could be found.

func Log

func Log(s Stream) Stream

Log wraps `p` with trace logs

func NopStream

func NopStream() Stream

NopStream returns a Stream adapter which discards all pubs

func Recover

func Recover(s Stream) Stream

Recover wraps `p` with a middleware that recovers from panics in Subscribe

func Stats

func Stats(s Stream) Stream

Stats wraps `p` with stats

func Trace

func Trace(s Stream) Stream

Trace wraps `p` with a trace middleware

type Sub

type Sub interface {
	// Subscribe subscribes the message handler h to the channel ch.
	// All subscriptions with the same q will form a queue group.
	// Each message will be delivered to only one subscriber per queue group.
	// Subscribe(q, ch string, h MsgHandler) error
	Subscribe(q, ch string, f MsgHandler, opts ...SubscriptionOption) (Subscription, error)
}

Sub is the subscribe interface

func SubFromContext

func SubFromContext(ctx context.Context) Sub

SubFromContext returns a `Sub` instance associated with `ctx`, or the local `Sub` if no instance could be found.

type Subscription

type Subscription interface {
	// Unsubscribe removes interest in the subscription.
	// For durables, it means that the durable interest is also removed from
	// the server. Restarting a durable with the same name will not resume
	// the subscription, it will be considered a new one.
	Unsubscribe() error
}

Subscription represents a subscription to the streaming platform

type SubscriptionOption

type SubscriptionOption interface{}