group

Documentation

Index

• func ApplyGroup[T any](ops []SetOption[T], o *Options[T])
• type Client
• type ClientGroup
  • func NewClientGroup[T any](strategy DispatchStrategy[Client[T]]) (group ClientGroup[T])
  • func (s ClientGroup[T]) Close() (err error)
  • func (s ClientGroup[T]) Done() <-chan struct{}
  • func (s ClientGroup[T]) Err() (err error)
  • func (s ClientGroup[T]) Forget(seq core.Seq, clientID ClientID)
  • func (s ClientGroup[T]) Has(seq core.Seq, clientID ClientID) (ok bool)
  • func (s ClientGroup[T]) Send(cmd core.Cmd[T], results chan<- core.AsyncResult) (seq core.Seq, clientID ClientID, n int, err error)
  • func (s ClientGroup[T]) SendWithDeadline(cmd core.Cmd[T], results chan<- core.AsyncResult, deadline time.Time) (seq core.Seq, clientID ClientID, n int, err error)
  • func (s ClientGroup[T]) Size() int
• type ClientID
• type DispatchStrategy
• type DispatchStrategyFactory
• type Options
• type RoundRobinStrategy
  • func NewRoundRobinStrategy[T any](sl []T) RoundRobinStrategy[T]
  • func (s RoundRobinStrategy[T]) Next() (t T, index int64)
  • func (s RoundRobinStrategy[T]) Slice() []T
• type RoundRobinStrategyFactory
  • func (RoundRobinStrategyFactory[T]) New(clients []Client[T]) DispatchStrategy[Client[T]]
• type SetOption
  • func WithClientOps[T any](ops ...cln.SetOption) SetOption[T]
  • func WithFactory[T any](factory DispatchStrategyFactory[T]) SetOption[T]
  • func WithReconnect[T any]() SetOption[T]

Functions

func ApplyGroup

func ApplyGroup[T any](ops []SetOption[T], o *Options[T])

Types

type Client

type Client[T any] interface {
	Send(cmd core.Cmd[T], results chan<- core.AsyncResult) (seq core.Seq, n int,
		err error)
	SendWithDeadline(cmd core.Cmd[T], results chan<- core.AsyncResult,
		deadline time.Time) (seq core.Seq, n int, err error)
	Has(seq core.Seq) bool
	Forget(seq core.Seq)
	Err() (err error)
	Close() (err error)
	Done() <-chan struct{}
}

Client represents a client used by the Group for sending commands and receiving results.

type ClientGroup

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

ClientGroup represents a group of clients used to communicate with the server.

It distributes the load across multiple connections, which can improve throughput and resilience. The group selects a client for each operation according to the provided dispatch strategy.

func NewClientGroup

func NewClientGroup[T any](strategy DispatchStrategy[Client[T]]) (
	group ClientGroup[T])

NewClientGroup creates a new ClientGroup using the provided dispatch strategy.

The returned group monitors all clients and automatically closes its Done() channel once all clients have finished.

func (ClientGroup[T]) Close

func (s ClientGroup[T]) Close() (err error)

Close terminates underlying clients.

All pending Commands will receive an error (AsyncResult.Error != nil).

func (ClientGroup[T]) Done

func (s ClientGroup[T]) Done() <-chan struct{}

Done returns a channel that is closed when the ClientGroup terminates.

func (ClientGroup[T]) Err

func (s ClientGroup[T]) Err() (err error)

Err returns a connection error.

func (ClientGroup[T]) Forget

func (s ClientGroup[T]) Forget(seq core.Seq, clientID ClientID)

Forget makes the Client to forget about the Command which still waiting for the result.

After calling Forget, all the results of the corresponding Command will be handled with UnexpectedResultCallback.

func (ClientGroup[T]) Has

func (s ClientGroup[T]) Has(seq core.Seq, clientID ClientID) (ok bool)

Has checks if the Command with the specified sequence number has been sent by the client and still waiting for the Result.

func (ClientGroup[T]) Send

func (s ClientGroup[T]) Send(cmd core.Cmd[T], results chan<- core.AsyncResult) (
	seq core.Seq, clientID ClientID, n int, err error)

Send transmits a Command to the server via one of the clients in the group, selected according to the dispatch strategy.

The corresponding Results from the server are delivered to the provided results channel. If the channel lacks sufficient capacity, receiving all Results may block.

Each Command is assigned a unique sequence number per client, starting from 1:

• The first Command sent by a client gets seq == 1, the second seq == 2, etc.
• seq == 0 is reserved for the Ping-Pong mechanism that ensures connection liveness.

Returns the sequence number assigned to the Command, the ClientID of the client it was sent through, the number of bytes written, and any error encountered (non-nil if the Command was not sent successfully).

func (ClientGroup[T]) SendWithDeadline

func (s ClientGroup[T]) SendWithDeadline(cmd core.Cmd[T],
	results chan<- core.AsyncResult,
	deadline time.Time,
) (seq core.Seq, clientID ClientID, n int, err error)

SendWithDeadline is like Send, but ensures that the Command is transmitted before the specified deadline.

func (ClientGroup[T]) Size

func (s ClientGroup[T]) Size() int

Size returns the number of clients within this ClientGroup.

type ClientID

type ClientID int

ClientID identifies a specific client within a Group.

type DispatchStrategy

type DispatchStrategy[T any] interface {
	Next() (t T, index int64)
	Slice() []T
}

DispatchStrategy is a dispatch strategy.

type DispatchStrategyFactory

type DispatchStrategyFactory[T any] interface {
	New(clients []Client[T]) DispatchStrategy[Client[T]]
}

DispatchStrategyFactory is a factory for a dispatch strategy.

type Options

type Options[T any] struct {
	Factory   DispatchStrategyFactory[T]
	Reconnect bool
	ClientOps []cln.SetOption
}

Options defines the configuration settings for creating a ClientGroup.

type RoundRobinStrategy

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

RoundRobinStrategy implements a round-robin dispatch strategy.

func NewRoundRobinStrategy

func NewRoundRobinStrategy[T any](sl []T) RoundRobinStrategy[T]

NewRoundRobinStrategy creates a new RoundRobinStrategy.

func (RoundRobinStrategy[T]) Next

func (s RoundRobinStrategy[T]) Next() (t T, index int64)

Next returns the next element and its index in the slice, following a round-robin strategy. The index is incremented atomically to ensure thread-safety in concurrent environments.

func (RoundRobinStrategy[T]) Slice

func (s RoundRobinStrategy[T]) Slice() []T

Slice returns the slice of elements underlying this RoundRobinStrategy.

type RoundRobinStrategyFactory

type RoundRobinStrategyFactory[T any] struct{}

RoundRobinStrategyFactory is a factory for a round-robin dispatch strategy.

func (RoundRobinStrategyFactory[T]) New

func (RoundRobinStrategyFactory[T]) New(
	clients []Client[T]) DispatchStrategy[Client[T]]

type SetOption

type SetOption[T any] func(o *Options[T])

func WithClientOps

func WithClientOps[T any](ops ...cln.SetOption) SetOption[T]

WithClientOps sets client-specific options to be applied when initializing each client in the group.

func WithFactory

func WithFactory[T any](factory DispatchStrategyFactory[T]) SetOption[T]

WithFactory sets the dispatch strategy factory for the client group.

The dispatch strategy determines how Commands are distributed among clients. For example, a round-robin strategy will rotate client usage evenly.

func WithReconnect

func WithReconnect[T any]() SetOption[T]

WithReconnect enables automatic reconnection for all clients in the group.

When this option is set, reconnect-capable clients are created, which attempt to re-establish the connection if it's lost during communication.