hydro

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Jul 30, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

README

Hydro

Coverage

[!WARNING] Hydro is an experimental project and currently in its research phase. We're still developing it and releases will be unstable. Expect bugs and breaking changes, even completely without notice.

Hydro is the real-time layer for any database or key-value store that has integrated pub/sub (PostgreSQL, Redis, ...). The goal of the project is to provide relatively low-latency real time updates for any data with solid caching to make sure real-time updates don't add any significant delays to responses.

More will follow soon.

Documentation

Index

Constants

View Source 
const RecommendedWantInterval = 20 * time.Second // The interval how often you should call "Want" on a subscription
View Source 
const SubscriptionDuration = 60 * time.Second // The duration after which a subscription will be deleted

Variables

View Source 
var (
	ErrChannelNotRegistered     = errors.New("channel is not registered")
	ErrChannelAlreadyRegistered = errors.New("channel already registered by different worker")
)

Standardized errors for pub/sub

View Source 
var Log = log.New(os.Stdout, "hydro ", log.Flags())

Functions

This section is empty.

Types

type BatchOptions 

type BatchOptions struct {
	BatchDuration time.Duration
	MaxAmount     int
}

type BatchOutput 

type BatchOutput[I comparable, O any] struct {
	Err     error
	Outputs map[I]O
}

type BatchRequest 

type BatchRequest[I comparable, O any] struct {
	Inputs    []I
	Collector chan BatchOutput[I, O]
}

type Batcher 

type Batcher[I comparable, O any] struct {
	Options   BatchOptions
	Collector chan BatchRequest[I, O]
	BatchFunc func([]I) (map[I]O, error)
}

func (*Batcher[I, O]) Do 

func (b *Batcher[I, O]) Do(inputs []I) (map[I]O, error)

Submit a task to the batcher

func (*Batcher[I, O]) Init 

func (b *Batcher[I, O]) Init()

type Change 

type Change[T any] interface {
	Stack(c Change[T]) Change[T]
}

type Config 

type Config[DB any, PS IPubSubBackend[DB]] struct {

	// The backend for Hydro's pub/sub model (if not set we'll use a local backend that acts as a replacement for a dedicated pub/sub service)
	PubSubBackend PS
}

type DatabaseListenerCreate 

type DatabaseListenerCreate[DB any, C Change[C]] struct {
	Identifier string                                   // Identifier for the listener (REQUIRED)
	Get        func(DB, []string) (map[string]C, error) // Get the base data from results of listeners or just with key (required)
	OnChange   func(key string, change Change[C])       // Called when the listener receives a change (required)

	PoolConfig PoolConfig // Config for the pooling of subscription workers
}

type DatabaseListenerDictionary 

type DatabaseListenerDictionary[DB any, PS IPubSubBackend[DB], C Change[C]] struct {
	Instance   *Instance[DB, PS] // Hydro instance related
	Identifier string            // Unique identifier for this listener dictionary
	// contains filtered or unexported fields
}

func NewListenerDictionary 

func NewListenerDictionary[DB any, PS IPubSubBackend[DB], C Change[C]](instance *Instance[DB, PS], create DatabaseListenerCreate[DB, C]) *DatabaseListenerDictionary[DB, PS, C]

Helper function for initializing a new listener dictionary properly

func (*DatabaseListenerDictionary[DB, PS, C]) Get 

func (ld *DatabaseListenerDictionary[DB, PS, C]) Get(db DB, keys []string) (map[string]C, error)

Get the value for keys from the listener dictionary (makes sure we can add batching in the future)

func (*DatabaseListenerDictionary[DB, PS, C]) GetIdentifier 

func (ld *DatabaseListenerDictionary[DB, PS, C]) GetIdentifier() string

func (*DatabaseListenerDictionary[DB, PS, C]) Reset 

func (ld *DatabaseListenerDictionary[DB, PS, C]) Reset(ctx context.Context, db DB, keys []string) error

Reset all values for the keys back to the original value by re-getting them and pushing that update

func (*DatabaseListenerDictionary[DB, PS, C]) Subscribe 

func (ld *DatabaseListenerDictionary[DB, PS, C]) Subscribe(db DB, keys []string, identifier string, subscription func(change Change[C])) error

func (*DatabaseListenerDictionary[DB, PS, C]) Unsubscribe 

func (ld *DatabaseListenerDictionary[DB, PS, C]) Unsubscribe(identifier string, keys []string)

Remove subscriptions for an identifier

func (*DatabaseListenerDictionary[DB, PS, C]) Update 

func (ld *DatabaseListenerDictionary[DB, PS, C]) Update(ctx context.Context, db DB, key string, change Change[C]) error

Update a key with a change, will broadcast the event to all subscribers + cache it

type IPubSubBackend 

type IPubSubBackend[DB any] interface {
	CreateWorker() ISubWorker
	Publish(ctx context.Context, database DB, channel string, message string) error
}

type ISubWorker 

type ISubWorker interface {
	Subscribe(ctx context.Context, channels ...string) error
	Unsubscribe(ctx context.Context, channels ...string) error
	OnMessage(func(channel string, message string))
	OnError(func(channel string, err error))
	Close()
}

type Identifiable 

type Identifiable interface {
	GetIdentifier() string
}

type Instance 

type Instance[DB any, PS IPubSubBackend[DB]] struct {
	// contains filtered or unexported fields
}

func New 

func New[DB any, PS IPubSubBackend[DB]](config *Config[DB, PS]) *Instance[DB, PS]

Create a new Hydro instance

type ListenerSubscriptions 

type ListenerSubscriptions[DB any, PS IPubSubBackend[DB], C Change[C]] struct {
	// contains filtered or unexported fields
}

A manager of subscriptions to a Listener that automatically evicts them statelessly when no longer wanted.

Also aggressively caches the current return value from the listener. This is done by stacking the changes using the Stack method from the Change interface. OnSubscribe is usually pretty expensive and managing it like this makes sure we always only call it exactly once.

func NewSubs 

func NewSubs[DB any, PS IPubSubBackend[DB], C Change[C]](instance *Instance[DB, PS]) *ListenerSubscriptions[DB, PS, C]

Create a new manager of listener subscriptions

func (*ListenerSubscriptions[DB, PS, C]) Add 

func (ls *ListenerSubscriptions[DB, PS, C]) Add(identifier string, subscription func(c Change[C]))

Mark a listener subscription as wanted (identifier is a unique identifier of the subscription)

func (*ListenerSubscriptions[T, PS, C]) Delete 

func (ls *ListenerSubscriptions[T, PS, C]) Delete(identifier string)

Delete a subscription from the subscriptions

func (*ListenerSubscriptions[T, PS, C]) DisableQueuing 

func (ls *ListenerSubscriptions[T, PS, C]) DisableQueuing(base Change[C])

After DisableQueuing the subscriptions start to actually send changes when they are received, before all are queued

func (*ListenerSubscriptions[T, PS, C]) IsQueuing 

func (ls *ListenerSubscriptions[T, PS, C]) IsQueuing() bool

Check if the subscriptions are currently still in queuing mode

func (*ListenerSubscriptions[T, PS, C]) OnChange 

func (ls *ListenerSubscriptions[T, PS, C]) OnChange(change Change[C])

Handles a change and sends it to all subscribers of the listener

type LocalPubSub 

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

func NewLocalPubSub 

func NewLocalPubSub() *LocalPubSub

func (*LocalPubSub) CreateWorker 

func (lpb *LocalPubSub) CreateWorker() ISubWorker

func (*LocalPubSub) Publish 

func (w *LocalPubSub) Publish(ctx context.Context, database any, channel string, message string) error

type LocalSubWorker 

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

func (*LocalSubWorker) Close 

func (w *LocalSubWorker) Close()

func (*LocalSubWorker) OnError 

func (w *LocalSubWorker) OnError(handler func(channel string, err error))

func (*LocalSubWorker) OnMessage 

func (w *LocalSubWorker) OnMessage(handler func(channel string, message string))

func (*LocalSubWorker) Subscribe 

func (w *LocalSubWorker) Subscribe(ctx context.Context, channels ...string) error

func (*LocalSubWorker) Unsubscribe 

func (w *LocalSubWorker) Unsubscribe(ctx context.Context, channels ...string) error

type PoolConfig 

type PoolConfig struct {
	// How many subscriptions should, at maximum, be done by one worker (default: 100)
	MaxAmountByWorker int
}

type SubPool 

type SubPool[DB any, PS IPubSubBackend[DB]] struct {
	// contains filtered or unexported fields
}

func NewPubSubPool 

func NewPubSubPool[DB any, PS IPubSubBackend[DB]](backend PS, config PoolConfig) *SubPool[DB, PS]

func (*SubPool[DB, T]) Close 

func (p *SubPool[DB, T]) Close()

func (*SubPool[DB, PS]) OnError 

func (p *SubPool[DB, PS]) OnError(handler func(channel string, err error))

OnError sets the error handler for all workers

func (*SubPool[DB, PS]) OnMessage 

func (p *SubPool[DB, PS]) OnMessage(handler func(channel string, message string))

OnMessage sets the message handler for all workers

func (*SubPool[DB, PS]) Subscribe 

func (p *SubPool[DB, PS]) Subscribe(ctx context.Context, channels ...string) error

Subscribe subscribes to channels, distributing them across workers

func (*SubPool[DB, PS]) Unsubscribe 

func (p *SubPool[DB, PS]) Unsubscribe(ctx context.Context, channels ...string) error

Unsubscribe unsubscribes from channels

type Subscription 

type Subscription[C Change[C]] = func(c Change[C])

Source Files

View all Source files

Directories

Path Synopsis
pkg
postgresql module

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.