conc

Documentation

Index

• func IDFunc[T any](input T) T
• type Batcher
  • func NewBatcher[T any, U any](inputChan chan T) *Batcher[T, U]
  • func NewIDBatcher[T any](inputChan chan T) *Batcher[T, []T]
  • func (fo *Batcher[T, U]) InputChannel() chan<- T
  • func (fo *Batcher[T, U]) New() chan U
  • func (fo *Batcher[T, U]) Remove(output chan U)
  • func (fo *Batcher[T, U]) Send(value T)
  • func (fo *Batcher[T, U]) Stop()
• type BatcherCmd
• type FanIn
  • func NewFanIn[T any](outChan chan T) *FanIn[T]
  • func (fi *FanIn[T]) Add(inputs ...<-chan T)
  • func (fi *FanIn[T]) Channel() chan T
  • func (fi *FanIn[T]) IsRunning() bool
  • func (fi *FanIn[T]) Remove(target <-chan T)
  • func (fi *FanIn[T]) Stop()
• type FanInCmd
• type FanOut
  • func NewFanOut[T any, U any](inputChan chan T, mapper func(T) U) *FanOut[T, U]
  • func NewIDFanOut[T any](input chan T, mapper func(T) T) *FanOut[T, T]
  • func (fo *FanOut[T, U]) InputChannel() chan<- T
  • func (fo *FanOut[T, U]) IsRunning() bool
  • func (fo *FanOut[T, U]) New() <-chan U
  • func (fo *FanOut[T, U]) Remove(output <-chan U)
  • func (fo *FanOut[T, U]) Send(value T)
  • func (fo *FanOut[T, U]) Stop()
• type FanOutCmd
• type Hub
  • func NewHub[M any](router Router[M]) *Hub[M]
  • func (h *Hub[M]) Connect(writer HubWriter[M]) *HubClient[M]
  • func (s *Hub[M]) Send(eventKey string, message M, callbackChan chan M) error
  • func (s *Hub[M]) Stop()
• type HubClient
  • func (h *HubClient[M]) Disconnect()
  • func (h *HubClient[M]) GetId() string
  • func (h *HubClient[M]) Subscribe(eventKeys ...string)
  • func (h *HubClient[M]) Unsubscribe(eventKeys ...string)
• type HubControlEvent
• type HubMessage
• type HubWriter
• type KVRouter
  • func NewKVRouter[M any]() *KVRouter[M]
  • func (r *KVRouter[M]) AddRoute(client *HubClient[M], eventKeys ...string) error
  • func (r *KVRouter[M]) Remove(client *HubClient[M]) error
  • func (r *KVRouter[M]) RemoveRoute(client *HubClient[M], eventKeys ...string) error
  • func (r *KVRouter[M]) RouteMessage(msg HubMessage[M]) error
• type Pipe
  • func NewIDPipe[T any](input <-chan T, output chan<- T, mapper func(T) T) *Pipe[T, T]
  • func NewPipe[T any, U any](input <-chan T, output chan<- U, mapper func(T) U) *Pipe[T, U]
  • func (p *Pipe[T, U]) Stop()
• type ReaderChan
  • func NewReader[R any, D any](read func() (R, error)) *ReaderChan[R, D]
  • func (ch *ReaderChan[R, D]) IsRunning() bool
  • func (rc *ReaderChan[R, D]) ResultChannel() chan ValueOrError[R]
  • func (ch *ReaderChan[T, D]) Stop() error
• type ReaderWriterBase
  • func (rwb *ReaderWriterBase[T, D]) WaitForFinish()
• type Router
• type ValueOrError
• type WriterChan
  • func NewWriter[W any, D any](write func(msg W) error) *WriterChan[W, D]
  • func (ch *WriterChan[W, D]) IsRunning() bool
  • func (wc *WriterChan[W, D]) Send(req W) bool
  • func (ch *WriterChan[T, D]) Stop() error

Functions

func IDFunc

func IDFunc[T any](input T) T

Types

type Batcher

type Batcher[T any, U any] struct {
	FlushPeriod time.Duration
	Joiner      func(inputs []T) (outputs U)
	// contains filtered or unexported fields
}

func NewBatcher

func NewBatcher[T any, U any](inputChan chan T) *Batcher[T, U]

func NewIDBatcher

func NewIDBatcher[T any](inputChan chan T) *Batcher[T, []T]

func (*Batcher[T, U]) InputChannel

func (fo *Batcher[T, U]) InputChannel() chan<- T

func (*Batcher[T, U]) New

func (fo *Batcher[T, U]) New() chan U

func (*Batcher[T, U]) Remove

func (fo *Batcher[T, U]) Remove(output chan U)

func (*Batcher[T, U]) Send

func (fo *Batcher[T, U]) Send(value T)

func (*Batcher[T, U]) Stop

func (fo *Batcher[T, U]) Stop()

type BatcherCmd

type BatcherCmd[T any] struct {
	Name    string
	Channel chan T
}

type FanIn

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

func NewFanIn

func NewFanIn[T any](outChan chan T) *FanIn[T]

func (*FanIn[T]) Add

func (fi *FanIn[T]) Add(inputs ...<-chan T)

func (*FanIn[T]) Channel

func (fi *FanIn[T]) Channel() chan T

func (*FanIn[T]) IsRunning

func (fi *FanIn[T]) IsRunning() bool

func (*FanIn[T]) Remove

func (fi *FanIn[T]) Remove(target <-chan T)

func (*FanIn[T]) Stop

func (fi *FanIn[T]) Stop()

type FanInCmd

type FanInCmd[T any] struct {
	Name           string
	AddedChannel   <-chan T
	RemovedChannel <-chan T
}

type FanOut

type FanOut[T any, U any] struct {
	Mapper func(inputs T) U
	// contains filtered or unexported fields
}

*

• FanOut takes a message from one chanel, applies a mapper function
• and fans it out to N output channels.

func NewFanOut

func NewFanOut[T any, U any](inputChan chan T, mapper func(T) U) *FanOut[T, U]

func NewIDFanOut

func NewIDFanOut[T any](input chan T, mapper func(T) T) *FanOut[T, T]

*

• Creates a new IDFanout bridge.

func (*FanOut[T, U]) InputChannel

func (fo *FanOut[T, U]) InputChannel() chan<- T

func (*FanOut[T, U]) IsRunning

func (fo *FanOut[T, U]) IsRunning() bool

func (*FanOut[T, U]) New

func (fo *FanOut[T, U]) New() <-chan U

func (*FanOut[T, U]) Remove

func (fo *FanOut[T, U]) Remove(output <-chan U)

func (*FanOut[T, U]) Send

func (fo *FanOut[T, U]) Send(value T)

func (*FanOut[T, U]) Stop

func (fo *FanOut[T, U]) Stop()

type FanOutCmd

type FanOutCmd[T any] struct {
	Name           string
	AddedChannel   chan T
	RemovedChannel <-chan T
}

type Hub

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

func NewHub

func NewHub[M any](router Router[M]) *Hub[M]

func (*Hub[M]) Connect

func (h *Hub[M]) Connect(writer HubWriter[M]) *HubClient[M]

func (*Hub[M]) Send

func (s *Hub[M]) Send(eventKey string, message M, callbackChan chan M) error

func (*Hub[M]) Stop

func (s *Hub[M]) Stop()

*

• Stop the Hub and cleanup.

type HubClient

type HubClient[M any] struct {
	WriteMessage func(HubMessage[M]) error
	// contains filtered or unexported fields
}

func (*HubClient[M]) Disconnect

func (h *HubClient[M]) Disconnect()

func (*HubClient[M]) GetId

func (h *HubClient[M]) GetId() string

func (*HubClient[M]) Subscribe

func (h *HubClient[M]) Subscribe(eventKeys ...string)

func (*HubClient[M]) Unsubscribe

func (h *HubClient[M]) Unsubscribe(eventKeys ...string)

type HubControlEvent

type HubControlEvent[M any] struct {
	Client               *HubClient[M]
	Quit                 bool
	Pause                bool
	AddedSubscriptions   []string
	RemovedSubscriptions []string
}

type HubMessage

type HubMessage[M any] struct {
	EventKey string
	Message  M
	Callback chan M
}

type HubWriter

type HubWriter[M any] func(HubMessage[M]) error

type KVRouter

type KVRouter[M any] struct {
	// contains filtered or unexported fields
}

*

• A type of router that maintains an eventKey -> Conn[] by
• a fixed event keys.

func NewKVRouter

func NewKVRouter[M any]() *KVRouter[M]

func (*KVRouter[M]) AddRoute

func (r *KVRouter[M]) AddRoute(client *HubClient[M], eventKeys ...string) error

func (*KVRouter[M]) Remove

func (r *KVRouter[M]) Remove(client *HubClient[M]) error

func (*KVRouter[M]) RemoveRoute

func (r *KVRouter[M]) RemoveRoute(client *HubClient[M], eventKeys ...string) error

func (*KVRouter[M]) RouteMessage

func (r *KVRouter[M]) RouteMessage(msg HubMessage[M]) error

type Pipe

type Pipe[T any, U any] struct {
	// contains filtered or unexported fields
}

func NewIDPipe

func NewIDPipe[T any](input <-chan T, output chan<- T, mapper func(T) T) *Pipe[T, T]

func NewPipe

func NewPipe[T any, U any](input <-chan T, output chan<- U, mapper func(T) U) *Pipe[T, U]

func (*Pipe[T, U]) Stop

func (p *Pipe[T, U]) Stop()

type ReaderChan

type ReaderChan[R any, D any] struct {
	ReaderWriterBase[R, D]

	Read func() (R, error)
	// contains filtered or unexported fields
}

func NewReader

func NewReader[R any, D any](read func() (R, error)) *ReaderChan[R, D]

func (*ReaderChan[R, D]) IsRunning

func (ch *ReaderChan[R, D]) IsRunning() bool

*

• Returns whether the connection reader/writer loops are running.

func (*ReaderChan[R, D]) ResultChannel

func (rc *ReaderChan[R, D]) ResultChannel() chan ValueOrError[R]

*

• Returns the conn's reader channel.

func (*ReaderChan[T, D]) Stop

func (ch *ReaderChan[T, D]) Stop() error

*

• This method is called to stop the channel
• If already connected then nothing is done and nil
• is not already connected, a connection will first be established
• (including auth and refreshing tokens) and then the reader and
• writers are started. SendRequest can be called to send requests
• to the peer and the (user provided) msgChannel will be used to
• handle messages from the server.

type ReaderWriterBase

type ReaderWriterBase[T any, D any] struct {
	Info D
	// Time allowed to read the next pong message from the peer.
	WaitTime time.Duration
	// contains filtered or unexported fields
}

func (*ReaderWriterBase[T, D]) WaitForFinish

func (rwb *ReaderWriterBase[T, D]) WaitForFinish()

*

• Waits until the socket connection is disconnected or manually stopped.

type Router

type Router[M any] interface {
	AddRoute(client *HubClient[M], eventKeys ...string) error
	RemoveRoute(client *HubClient[M], eventKeys ...string) error
	Remove(client *HubClient[M]) error
	RouteMessage(HubMessage[M]) error
}

*

• In our streamer the main problem is given a message, identifying
• all connections it should be sent to. Instead of having a fixed
• key -> HubClient[] mapping, it is easier to simply for each
• message use a router delegate to identify all HubClients that
• are candidates. This Router does that.

type ValueOrError

type ValueOrError[T any] struct {
	Value T
	Error error
}

type WriterChan

type WriterChan[W any, D any] struct {
	ReaderWriterBase[W, D]

	Write   func(msg W) error
	OnClose func()
	// contains filtered or unexported fields
}

func NewWriter

func NewWriter[W any, D any](write func(msg W) error) *WriterChan[W, D]

func (*WriterChan[W, D]) IsRunning

func (ch *WriterChan[W, D]) IsRunning() bool

*

• Returns whether the connection reader/writer loops are running.

func (*WriterChan[W, D]) Send

func (wc *WriterChan[W, D]) Send(req W) bool

func (*WriterChan[T, D]) Stop

func (ch *WriterChan[T, D]) Stop() error

*

• This method is called to stop the channel
• If already connected then nothing is done and nil
• is not already connected, a connection will first be established
• (including auth and refreshing tokens) and then the reader and
• writers are started. SendRequest can be called to send requests
• to the peer and the (user provided) msgChannel will be used to
• handle messages from the server.