grpcsync

Documentation

Overview

Package grpcsync implements additional synchronization primitives built upon the sync package.

Index

• func OnceFunc(f func()) func()
• type CallbackSerializer
  • func NewCallbackSerializer(ctx context.Context) *CallbackSerializer
  • func (cs *CallbackSerializer) Done() <-chan struct{}
  • func (cs *CallbackSerializer) ScheduleOr(f func(ctx context.Context), onFailure func())
  • func (cs *CallbackSerializer) TrySchedule(f func(ctx context.Context))
• type Event
  • func NewEvent() *Event
  • func (e *Event) Done() <-chan struct{}
  • func (e *Event) Fire() bool
  • func (e *Event) HasFired() bool
• type PubSub
  • func NewPubSub(ctx context.Context) *PubSub
  • func (ps *PubSub) Done() <-chan struct{}
  • func (ps *PubSub) Publish(msg any)
  • func (ps *PubSub) Subscribe(sub Subscriber) (cancel func())
• type Subscriber

Functions

func OnceFunc

func OnceFunc(f func()) func()

OnceFunc returns a function wrapping f which ensures f is only executed once even if the returned function is executed multiple times.

Types

type CallbackSerializer

type CallbackSerializer struct {
	// contains filtered or unexported fields
}

CallbackSerializer provides a mechanism to schedule callbacks in a synchronized manner. It provides a FIFO guarantee on the order of execution of scheduled callbacks. New callbacks can be scheduled by invoking the Schedule() method.

This type is safe for concurrent access.

func NewCallbackSerializer

func NewCallbackSerializer(ctx context.Context) *CallbackSerializer

NewCallbackSerializer returns a new CallbackSerializer instance. The provided context will be passed to the scheduled callbacks. Users should cancel the provided context to shutdown the CallbackSerializer. It is guaranteed that no callbacks will be added once this context is canceled, and any pending un-run callbacks will be executed before the serializer is shut down.

func (*CallbackSerializer) Done

func (cs *CallbackSerializer) Done() <-chan struct{}

Done returns a channel that is closed after the context passed to NewCallbackSerializer is canceled and all callbacks have been executed.

func (*CallbackSerializer) ScheduleOr

func (cs *CallbackSerializer) ScheduleOr(f func(ctx context.Context), onFailure func())

ScheduleOr schedules the provided callback function f to be executed in the order it was added. If the context passed to NewCallbackSerializer has been canceled before this method is called, the onFailure callback will be executed inline instead.

Callbacks are expected to honor the context when performing any blocking operations, and should return early when the context is canceled.

func (*CallbackSerializer) TrySchedule

func (cs *CallbackSerializer) TrySchedule(f func(ctx context.Context))

TrySchedule tries to schedule the provided callback function f to be executed in the order it was added. This is a best-effort operation. If the context passed to NewCallbackSerializer was canceled before this method is called, the callback will not be scheduled.

Callbacks are expected to honor the context when performing any blocking operations, and should return early when the context is canceled.

type Event

type Event struct {
	// contains filtered or unexported fields
}

Event represents a one-time event that may occur in the future.

func NewEvent

func NewEvent() *Event

NewEvent returns a new, ready-to-use Event.

func (*Event) Done

func (e *Event) Done() <-chan struct{}

Done returns a channel that will be closed when Fire is called.

func (*Event) Fire

func (e *Event) Fire() bool

Fire causes e to complete. It is safe to call multiple times, and concurrently. It returns true iff this call to Fire caused the signaling channel returned by Done to close.

func (*Event) HasFired

func (e *Event) HasFired() bool

HasFired returns true if Fire has been called.

type PubSub

type PubSub struct {
	// contains filtered or unexported fields
}

PubSub is a simple one-to-many publish-subscribe system that supports messages of arbitrary type. It guarantees that messages are delivered in the same order in which they were published.

Publisher invokes the Publish() method to publish new messages, while subscribers interested in receiving these messages register a callback via the Subscribe() method.

Once a PubSub is stopped, no more messages can be published, but any pending published messages will be delivered to the subscribers. Done may be used to determine when all published messages have been delivered.

func NewPubSub

func NewPubSub(ctx context.Context) *PubSub

NewPubSub returns a new PubSub instance. Users should cancel the provided context to shutdown the PubSub.

func (*PubSub) Done

func (ps *PubSub) Done() <-chan struct{}

Done returns a channel that is closed after the context passed to NewPubSub is canceled and all updates have been sent to subscribers.

func (*PubSub) Publish

func (ps *PubSub) Publish(msg any)

Publish publishes the provided message to the PubSub, and invokes callbacks registered by subscribers asynchronously.

func (*PubSub) Subscribe

func (ps *PubSub) Subscribe(sub Subscriber) (cancel func())

Subscribe registers the provided Subscriber to the PubSub.

If the PubSub contains a previously published message, the Subscriber's OnMessage() callback will be invoked asynchronously with the existing message to begin with, and subsequently for every newly published message.

The caller is responsible for invoking the returned cancel function to unsubscribe itself from the PubSub.

type Subscriber

type Subscriber interface {
	// OnMessage is invoked when a new message is published. Implementations
	// must not block in this method.
	OnMessage(msg any)
}

Subscriber represents an entity that is subscribed to messages published on a PubSub. It wraps the callback to be invoked by the PubSub when a new message is published.