replicators

package module
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 15, 2026 License: MIT Imports: 10 Imported by: 0

README

Replicators

Replicators is an in-process pub/sub fan-out library for high-concurrency Go applications.

Prioritizes bounded memory usage, predictable latency, and throughput over guaranteeing delivery to slow consumers, even automatically detaching (dropping) slow consumers when needed.

Model

When a consumer is slow, one has the following options:

  1. Apply backpressure: slow the producer.
  2. Buffer: absorb bursts (temporarily).
  3. Drop messages: sacrifice completeness.
  4. Drop consumers: sacrifice availability for those consumers.
  5. Persist to durable storage: let consumers catch up later.

replicators combines the first four strategies.

  1. Once all buffers all full, broadcasting will start to block
  2. There's a "send" buffer, and every subscription has its own "receive buffer" (there's also dynamic subscription buffers but they are not relevant to this problem)
  3. The hub will forego delivery of messages to subscribers after a hub-global timeout
  4. The hub will drop consumers after they've missed a configurable (1-x) number of messages

Limitations

No redelivery/retries or seeking (since there is no persistence). If a consumer is dropped, it will have to resubscribe and will not receive any messages sent in the mean time, except for any configured "echo buffer".

Documentation and Examples

GoDoc including examples are found on pkg.go.dev.

Although there are other use cases, I created this library for the purpose of scalable edge replication of broker messages. The below chart illustrates an example topology.

flowchart LR
    broker["Message Broker<br>Event Stream"]
    consumer["Consumer Goroutine"]
    hub["Hub"]

    broker -->|Consume| consumer
    consumer -->|Broadcast| hub

    hub --> sub1["Subscription"]
    hub --> sub2["Subscription"]
    hub --> sub3["Subscription"]
    hub --> sub4["Subscription"]

    sub1 --> grpc1["gRPC Client"]
    sub2 --> grpc2["gRPC Client"]
    sub3 --> ws1["WebSocket Client"]
    sub4 --> sse1["SSE Client"]

A more elaborate toy example of a WebSocket server with dynamic producers can be found the ./examples/ directory. I can recommend the websocat tool to do some poking at it, eg:

go run ./examples/websocket/ &
websocat -v -H='Authorization: Bearer secret' ws://localhost:9001/foo/bar

Key Properties

  • Online attaching and detaching of subscribers
  • Automatic detaching of slow consumers
  • Comprehensive event handling mechanism
  • No 3rd party dependencies
  • Type safe
  • Idiomatic
  • Tested, benchmarked, (partly) optimized

Niceties

  • Bundled slog event handler
  • Native stat (counters, gauges) handler, useful for integration with eg. Prometheus scraping
  • Echo: replicate the last n sent messages to new subscribers

Echo Buffer

A hub can be configured to maintain a buffer of n messages to be sent to new subscribers upon connecting. If the same consumer subscribes again it may receive a message it had previously received through the old subscription.

Event Handlers

Event dispatching is on the hot path. Handlers should be very fast. Consider using HubEventHandlerHub, which creates a simple buffered event channel.

License

MIT

Documentation

Overview

Example 
package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/johnknl/replicators"
)

type MyMsg int

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	start := make(chan struct{})
	subbed := make(chan struct{})
	dropped := make(chan struct{})

	hub := replicators.NewHub(
		ctx,
		replicators.WithDevLogger[MyMsg](),
		replicators.WithCounterHandler[MyMsg](),
		replicators.WithEventHandler(replicators.EventHandlerFunc[MyMsg](func(_ context.Context, e replicators.Event[MyMsg]) {
			switch e.(type) {
			case replicators.EvtSubscribed[MyMsg]:
				close(subbed)
			case replicators.EvtSubDropped[MyMsg]:
				close(dropped)
			}
		})),
	)

	wg := sync.WaitGroup{}

	wg.Go(func() {
		// 1. The first message is never received
		// 2. One message is read
		// 3. Next delivery will be dropped by the hub (but tolerated)
		// 4. Final delivery will be dropped by the hub, and the subscription will be dropped
		for i := range 4 {
			_ = hub.Broadcast(ctx, MyMsg(i))
			if i == 0 {
				close(start)
				<-subbed
			}
		}
	})

	wg.Go(func() {
		// Don't start the next consumer until the first send is dropped
		<-start

		subscription, err := hub.Subscribe(ctx, replicators.WithMaxDeliveryTimeouts[MyMsg](1))
		if err != nil {
			panic(err)
		}

		// We'll read one message, then block until the subscription is
		// dropped by the hub.
		<-subscription.Data()
		<-dropped

		fmt.Println("error: " + subscription.Err().Error()) // nolint:forbidigo // example code
	})

	wg.Wait()

	fmt.Printf("%#v\n", hub.Stats(ctx).Counts) // nolint:forbidigo // example code

}

Output:
error: subscription dropped
&replicators.Counts{Subscriptions:1, Cancellations:0, Sent:4, Timeouts:2, Dropped:1, Delivered:1, Undeliverable:1}

Index

Examples

Constants

View Source 
const (
	// DefaultSendBuffer is the default size of the send buffer.
	DefaultSendBuffer = 10

	// DefaultAttachBuffer is the default size of the attach buffer.
	DefaultAttachBuffer = 1

	// DefaultCancelBuffer is the default size of the cancel buffer.
	DefaultCancelBuffer = 1

	// DefaultDeliveryTimeout is the default timeout for delivering messages to subscribers.
	DefaultDeliveryTimeout = 100 * time.Millisecond

	// DefaultShutdownTimeout is the default timeout for shutting down the hub.
	DefaultShutdownTimeout = 10 * time.Second
)

Variables

View Source 
var (
	// ErrSubscriptionDropped is returned when a subscription is dropped.
	// This happens when the subscriber is too slow to process messages and the tolerance is exceeded.
	ErrSubscriptionDropped = errors.New("subscription dropped")

	// ErrSubscriptionCancelled is returned when a subscription is cancelled.
	// This happens when the subscriber cancels the subscription.
	ErrSubscriptionCancelled = errors.New("subscription cancelled")
)

Functions

func WithAttachBuffer 

func WithAttachBuffer[T any](size int) func(*Hub[T])

WithAttachBuffer sets the buffer size for the attach channel. The attach channel is used to queue new subscribers. Note that on shutdown, the attach buffer is discarded.

func WithCancelBuffer 

func WithCancelBuffer[T any](size int) func(*Hub[T])

WithCancelBuffer sets the buffer size for the cancel channel. The cancel channel is used to cancel subscriptions. Note that on shutdown, the cancel buffer is discarded.

func WithCounterHandler 

func WithCounterHandler[T any]() func(*Hub[T])

WithCounterHandler adds a CounterHandler to the hub.

func WithDeliveryTimeout 

func WithDeliveryTimeout[T any](timeout time.Duration) func(*Hub[T])

WithDeliveryTimeout sets the maximum time to wait for a subscriber to process a message.

func WithDevLogger 

func WithDevLogger[T any]() func(*Hub[T])

WithDevLogger is a convenience function that sets up a development logger using the standard library's slog package.

func WithEchoBuffer added in v0.2.0 

func WithEchoBuffer[T any](n int) func(*Hub[T])

WithEchoBuffer enables the echo feature. When enabled, the hub will echo the last `n` messages sent, to new subscribers.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/johnknl/replicators"
)

type MyMsg int

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	dropped := make(chan struct{})
	hub := replicators.NewHub(ctx,
		replicators.WithDevLogger[MyMsg](),
		replicators.WithEchoBuffer[MyMsg](1),
		replicators.WithEventHandlerFunc[MyMsg](func(_ context.Context, e replicators.Event[MyMsg]) {
			if _, ok := e.(replicators.EvtSubDropped[MyMsg]); ok {
				close(dropped)
			}
		}),
		replicators.WithCounterHandler[MyMsg](),
	)

	// Broadcast a message, it'll be undeliverable because there are no subscribers,
	// but the hub will remember it and send it to the next subscriber.
	// This would be better accomplished with a send buffer but this is just
	// part of the example.
	if err := hub.Broadcast(ctx, MyMsg(42)); err != nil {
		panic(err)
	}

	// Subscribe to the hub, the subscriber will receive the last message
	// that was broadcasted.
	subscription, err := hub.Subscribe(ctx)
	if err != nil {
		panic(err)
	}

	msg := <-subscription.Data()

	fmt.Println(msg) // nolint:forbidigo // example code

	// Now broadcast another message, the subscriber will receive it.
	if err = hub.Broadcast(ctx, MyMsg(43)); err != nil {
		panic(err)
	}

	msg = <-subscription.Data()

	fmt.Println(msg) // nolint:forbidigo // example code

	// Now subscribe to the hub again, the subscriber will receive the last message
	// that was broadcasted.
	subscription2, err := hub.Subscribe(ctx)
	if err != nil {
		panic(err)
	}

	msg = <-subscription2.Data()

	fmt.Println(msg) // nolint:forbidigo // example code

	// Now subscribe again, but don't read the message: subscriber is
	// immediately subject to max delivery timeout (which defaults to 0 timeouts of 100ms).
	_, err = hub.Subscribe(ctx)
	if err != nil {
		panic(err)
	}

	// Wait for the subscription to be dropped by the hub.
	<-dropped

	fmt.Printf("%#v\n", hub.Stats(ctx).Counts) // nolint:forbidigo // example code

}

Output:
42
43
43
&replicators.Counts{Subscriptions:3, Cancellations:0, Sent:2, Timeouts:1, Dropped:1, Delivered:3, Undeliverable:1}

func WithEventHandler 

func WithEventHandler[T any](handler EventHandler[T]) func(*Hub[T])

WithEventHandler adds an EventHandler for the hub. It will wrap any existing EventHandler.To set all event handlers, use WithEventHandlers instead.

func WithEventHandlerFunc 

func WithEventHandlerFunc[T any](fn func(context.Context, Event[T])) func(*Hub[T])

WithEventHandlerFunc adds an EventHandler for the hub.

func WithEventHandlers 

func WithEventHandlers[T any](handlers ...EventHandler[T]) func(*Hub[T])

WithEventHandlers sets the EventHandler(s) for the hub.

func WithMaxDeliveryTimeouts 

func WithMaxDeliveryTimeouts[T any](tolerance int) func(*Subscription[T])

WithMaxDeliveryTimeouts sets the maximum number of timeouts before a subscription is dropped.

func WithReceiveBuffer 

func WithReceiveBuffer[T any](buffSize int) func(*Subscription[T])

WithReceiveBuffer sets the receive buffer size for a subscription.

func WithSendBuffer 

func WithSendBuffer[T any](size int) func(*Hub[T])

WithSendBuffer sets the buffer size for the send channel.

func WithShutdownTimeout 

func WithShutdownTimeout[T any](timeout time.Duration) func(*Hub[T])

WithShutdownTimeout sets the maximum total time to wait for the hub to shutdown gracefully.

func WithSlogger 

func WithSlogger[T any](ctx context.Context, logger *slog.Logger) func(*Hub[T])

WithSlogger is a convenience function that sets up a slog.Logger as an event handler for the hub.

Example 
package main

import (
	"context"
	"log/slog"
	"os"

	"github.com/johnknl/replicators"
)

type MyMsg int

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug}))
	replicators.NewHub(ctx, replicators.WithSlogger[MyMsg](ctx, logger))
}

Types

type CounterHandler 

type CounterHandler[T any] struct {
	// contains filtered or unexported fields
}

CounterHandler is an EventHandler that updates the counters based on the events received.

func NewCounterHandler 

func NewCounterHandler[T any]() *CounterHandler[T]

NewCounterHandler creates a new CounterHandler.

func (*CounterHandler[T]) HandleEvent 

func (s *CounterHandler[T]) HandleEvent(_ context.Context, e Event[T])

HandleEvent updates the stats based on the event type.

type Counts 

type Counts struct {
	// Subscriptions is the total number of subscriptions created.
	Subscriptions int64

	// Cancellations is the total number of subscriptions cancelled.
	Cancellations int64

	// Sent is the total number of *messages* as counted when pushed on the queue.
	Sent int64

	// Timeouts is the total number of *deliveries* abandoned due to timeout.
	Timeouts int64

	// Dropped is the total number of *subscriptions* dropped due to slow consumers.
	Dropped int64

	// Delivered is the total number of copies delivered.
	Delivered int64

	// Undeliverable is the total number of messages that could not be delivered to any subscriber.
	Undeliverable int64
}

Counts contains the current stats for the hub.

type Event 

type Event[T any] interface {
	Name() EventName
}

Event is a marker interface for events that can be handled by an EventHandler.

type EventHandler 

type EventHandler[T any] interface {
	HandleEvent(context.Context, Event[T])
}

EventHandler is a simple interface for handling events.

func DevLoggerHandler 

func DevLoggerHandler[T any]() EventHandler[T]

DevLoggerHandler returns an EventHandler that logs events using the standard library's slog package.

func SlogEventHandler 

func SlogEventHandler[T any](ctx context.Context, logger *slog.Logger) EventHandler[T]

SlogEventHandler returns an EventHandler that logs events using the provided slog.Logger.

Note that the supported levels are determined at creation time and will not be reflected if the logger's level is changed later. The logger will use the context created with the EventHandler, not the context passed when the event is handled.

Disabling warning logging will result in nothing being logged.

type EventHandlerFunc 

type EventHandlerFunc[T any] func(context.Context, Event[T])

EventHandlerFunc is a function type that implements the EventHandler interface.

func (EventHandlerFunc[T]) HandleEvent 

func (f EventHandlerFunc[T]) HandleEvent(ctx context.Context, e Event[T])

HandleEvent calls the function with the event.

type EventName 

type EventName string

EventName is a string that represents the name of an event.

type EvtCancelFailed 

type EvtCancelFailed[T any] struct {
	Sub *Subscription[T]
	Err error
}

EvtCancelFailed is emitted when a subscription cancellation fails.

func (EvtCancelFailed[T]) Name 

func (e EvtCancelFailed[T]) Name() EventName

Name returns the name of the event.

func (EvtCancelFailed[T]) Subscription 

func (e EvtCancelFailed[T]) Subscription() *Subscription[T]

Subscription returns the subscription that failed to be cancelled.

type EvtCancelled 

type EvtCancelled[T any] struct{ Sub *Subscription[T] }

EvtCancelled is emitted when a subscription is explicitly cancelled.

func (EvtCancelled[T]) Name 

func (e EvtCancelled[T]) Name() EventName

Name returns the name of the event.

func (EvtCancelled[T]) Subscription 

func (e EvtCancelled[T]) Subscription() *Subscription[T]

Subscription returns the subscription that was cancelled.

type EvtDelivered 

type EvtDelivered[T any] struct{ Msg T }

EvtDelivered is emitted when a message is delivered to a subscriber.

func (EvtDelivered[T]) Message 

func (e EvtDelivered[T]) Message() T

Message returns the message that was delivered to a subscriber.

func (EvtDelivered[T]) Name 

func (e EvtDelivered[T]) Name() EventName

Name returns the name of the event.

type EvtDeliveryTimeout 

type EvtDeliveryTimeout[T any] struct {
	Msg     T
	Timeout time.Duration
}

EvtDeliveryTimeout is emitted when a message delivery to a subscriber times out.

func (EvtDeliveryTimeout[T]) Message 

func (e EvtDeliveryTimeout[T]) Message() T

Message returns the message that timed out.

func (EvtDeliveryTimeout[T]) Name 

func (e EvtDeliveryTimeout[T]) Name() EventName

Name returns the name of the event.

type EvtDrainCancelled 

type EvtDrainCancelled struct{ Remaining int }

EvtDrainCancelled is emitted when draining the send buffer on shutdown is cancelled due to the context being cancelled, and contains the number of messages left in the queue.

func (EvtDrainCancelled) Name 

func (e EvtDrainCancelled) Name() EventName

Name returns the name of the event.

type EvtDraining 

type EvtDraining struct{ Count int }

EvtDraining is emitted when the hub is draining, and contains the number of messages left in the queue.

func (EvtDraining) Name 

func (e EvtDraining) Name() EventName

Name returns the name of the event.

type EvtNoSubscribers 

type EvtNoSubscribers[T any] struct{ Msg T }

EvtNoSubscribers is emitted when a message is sent but there are no subscribers to receive it.

func (EvtNoSubscribers[T]) Message 

func (e EvtNoSubscribers[T]) Message() T

Message returns the message that was sent but not received by any subscribers.

func (EvtNoSubscribers[T]) Name 

func (e EvtNoSubscribers[T]) Name() EventName

Name returns the name of the event.

type EvtSendContextCancelled 

type EvtSendContextCancelled[T any] struct{ Msg T }

EvtSendContextCancelled is emitted when a message send is cancelled due to the context being cancelled.

func (EvtSendContextCancelled[T]) Message 

func (e EvtSendContextCancelled[T]) Message() T

Message returns the message that was attempted to be sent when the context was cancelled.

func (EvtSendContextCancelled[T]) Name 

func (e EvtSendContextCancelled[T]) Name() EventName

Name returns the name of the event.

type EvtSent 

type EvtSent[T any] struct{ Msg T }

EvtSent is emitted when a message is sent to the hub.

func (EvtSent[T]) Message 

func (e EvtSent[T]) Message() T

Message returns the message that was sent to the hub.

func (EvtSent[T]) Name 

func (e EvtSent[T]) Name() EventName

Name returns the name of the event.

type EvtShutdown 

type EvtShutdown struct{}

EvtShutdown is emitted when the hub is shutting down.

func (EvtShutdown) Name 

func (e EvtShutdown) Name() EventName

Name returns the name of the event.

type EvtSubDropped 

type EvtSubDropped[T any] struct{ Sub *Subscription[T] }

EvtSubDropped is emitted when a subscription is dropped due its consumer being slower than allowed.

func (EvtSubDropped[T]) Name 

func (e EvtSubDropped[T]) Name() EventName

Name returns the name of the event.

type EvtSubscribed 

type EvtSubscribed[T any] struct{ Sub *Subscription[T] }

EvtSubscribed is emitted when a subscription is successfully created.

func (EvtSubscribed[T]) Name 

func (e EvtSubscribed[T]) Name() EventName

Name returns the name of the event.

func (EvtSubscribed[T]) Subscription 

func (e EvtSubscribed[T]) Subscription() *Subscription[T]

Subscription returns the subscription that was created.

type Gauges 

type Gauges struct {
	// Queued is the current size of the message queue.
	Queued int

	// Subscriptions is the current number of subscribers
	// attached to the hub.
	Subscriptions int

	// Waiting is the current number of subscribers waiting
	// to be attached to the hub.
	Waiting int

	// Cancelling is the current number of subscribers waiting
	// to be detached from the hub.
	Cancelling int

	// Buffered is the current total number of messages buffered
	// in the delivery buffers. This is the sum of all messages
	// buffered for all subscribers.
	Buffered int
}

Gauges contains the current gauges for the hub.

type Hub 

type Hub[T any] struct {
	// contains filtered or unexported fields
}

Hub provides the frontend for interacting with the loop. It provides an API for sending messages, subscribing to messages, and cancelling subscriptions.

func NewHub 

func NewHub[T any](ctx context.Context, opts ...func(*Hub[T])) *Hub[T]

NewHub creates a new sub.

func (*Hub[T]) Broadcast 

func (s *Hub[T]) Broadcast(ctx context.Context, msg T) error

Broadcast a message to subscribers. If the context is canceled, this is effectively a no-op.

func (*Hub[T]) Cancel 

func (s *Hub[T]) Cancel(ctx context.Context, sub *Subscription[T]) error

Cancel a subscription. If the context is canceled, cancellation will fail.

func (*Hub[T]) Stats 

func (s *Hub[T]) Stats(ctx context.Context) *Stats

Stats returns a snapshot of hub statistics. When no CounterHandler is attached, the counts will be zero. Note that calling this method will block the main dispatch loop briefly while gathering metrics.

The SnapshottingGuages event is emitted at the start of the stats gathering process.

Calling this method after shutdown may result in a deadlock if the passed context is not canceled.

func (*Hub[T]) Subscribe 

func (s *Hub[T]) Subscribe(ctx context.Context, opts ...func(*Subscription[T])) (*Subscription[T], error)

Subscribe to messages.

When dealing with a subscriber that has very variable message processing time, a larger buffer will help smooth things out. Otherwise, a buffer of 1 is usually sufficient. The tolerance parameter is the number of messages that can be dropped before the subscription is cancelled. If tolerance is equal or lower than 0, the subscription will be cancelled immediately when a message is dropped.

The passed context is used to cancel the subscription. If the context is canceled, the subscription will be cancelled and an error will be returned. Note that the context is not the only way a subscription may be detached.

A subscription can also be cancelled by invoking Cancel() on the hub, and it will be dropped if the subscriber is too slow to process messages and the tolerance is exceeded. When the context is canceled, the subscription's error will be set to context.Context.Err().

Example (WithReceiveBuffer) 
package main

import (
	"context"
	"fmt"
	"time"

	"github.com/johnknl/replicators"
)

type MyMsg int

func main() {
	buffSize := 10
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	subbed := make(chan struct{})

	hub := replicators.NewHub(
		ctx,
		replicators.WithEventHandlerFunc[MyMsg](func(_ context.Context, e replicators.Event[MyMsg]) {
			switch e.(type) {
			case replicators.EvtSubscribed[MyMsg]:
				// Because the default settings use a non-zero attach buffer, some of the data would be
				// missed unless we sync. It would be more straightforward to use a zero attach buffer,
				// but this is just an intentionally belabored example.
				close(subbed)
			}
		}),
	)

	subscription, err := hub.Subscribe(ctx, replicators.WithReceiveBuffer[MyMsg](buffSize))
	if err != nil {
		panic(err)
	}

	// Wait for the subscription to be fully attached before broadcasting messages.
	<-subbed

	// fill up the buffer
	for i := range buffSize {
		_ = hub.Broadcast(ctx, MyMsg(i))
	}

	// read the messages from the receive buffer
	for msg := range subscription.Data() {
		fmt.Println(msg) // nolint:forbidigo // example code
	}

}

Output:
0
1
2
3
4
5
6
7
8
9

type HubEventHandlerHub added in v0.3.0 

type HubEventHandlerHub[T any] struct {
	// contains filtered or unexported fields
}

HubEventHandlerHub is an EventHandler that dispatches events to multiple EventHandlers using a buffered channel.

func NewHubEventHandlerHub added in v0.3.0 

func NewHubEventHandlerHub[T any](ctx context.Context, buffer int, handlers ...EventHandler[T]) *HubEventHandlerHub[T]

NewHubEventHandlerHub creates a new HubEventHandlerHub with the given buffer size and EventHandlers.

func (*HubEventHandlerHub[T]) HandleEvent added in v0.3.0 

func (h *HubEventHandlerHub[T]) HandleEvent(ctx context.Context, e Event[T])

HandleEvent dispatches the event to all registered EventHandlers.

type Ring added in v0.2.0 

type Ring[T any] struct {
	// contains filtered or unexported fields
}

Ring is a generic ring buffer that holds a fixed number of elements of type T.

func NewRing added in v0.2.0 

func NewRing[T any](n int) *Ring[T]

NewRing creates a new ring buffer with the specified size.

func (*Ring[T]) Add added in v0.2.0 

func (r *Ring[T]) Add(v T)

Add adds a new element to the ring buffer. If the buffer is full, it overwrites the oldest element.

func (*Ring[T]) Empty added in v0.2.0 

func (r *Ring[T]) Empty() bool

Empty returns true if the ring buffer is empty.

func (*Ring[T]) Values added in v0.2.0 

func (r *Ring[T]) Values() iter.Seq[T]

Values returns a sequence of the elements in the ring buffer in the order they were added.

type SnapshottingGuages 

type SnapshottingGuages[T any] struct{ Msg T }

SnapshottingGuages is emitted when the hub is snapshotting its gauges.

func (SnapshottingGuages[T]) Name 

func (e SnapshottingGuages[T]) Name() EventName

Name returns the name of the event.

type Stats 

type Stats struct {
	// Gauges contains the current gauges for the hub.
	Gauges *Gauges

	// Counts contains the current counts for the hub.
	Counts *Counts
}

Stats contains the current stats for the hub.

func (*Stats) String 

func (s *Stats) String() string

type Subscription 

type Subscription[T any] struct {
	// contains filtered or unexported fields
}

Subscription represents a subscription to a Hub.

func (*Subscription[T]) Cancel 

func (s *Subscription[T]) Cancel(ctx context.Context) error

Cancel cancels the subscription via the associated Hub.

func (*Subscription[T]) Data 

func (s *Subscription[T]) Data() <-chan T

Data returns a read-only channel that receives data from the subscription.

func (*Subscription[T]) Err 

func (s *Subscription[T]) Err() error

Err returns the error associated with the subscription, if any.

Source Files

View all Source files

Directories

Path Synopsis
examples
sse command
websocket command

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.