conc

May 23, 2024

Changes in this version

type FanOut

+ SendSync bool

+ type Mapper struct

+ MapFunc func(I) (O, bool, bool)

+ OnDone func(p *Mapper[I, O])

+ func NewMapper[T any, U any](input <-chan T, output chan<- U, mapper func(T) (U, bool, bool)) *Mapper[T, U]

May 21, 2024

Changes in this version

+ type Reducer struct

+ FlushPeriod time.Duration

+ ReduceFunc func(inputs []T) (outputs U)

+ func NewIDReducer[T any](inputChan chan T, outputChan chan []T) *Reducer[T, []T]

+ func NewReducer[T any, U any](inputChan chan T, outputChan chan U) *Reducer[T, U]

+ func (fo *Reducer[T, U]) Flush()

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

+ func (fo *Reducer[T, U]) SendChan() chan<- T

+ func (fo *Reducer[T, U]) Stop()

Mar 17, 2024

Changes in this version

type Map

+ func (m *Map[K, V]) LDelete(k K)

+ func (m *Map[K, V]) LGet(k K) (V, bool)

+ func (m *Map[K, V]) LHas(k K) bool

+ func (m *Map[K, V]) LRange(meth func(K, V) bool)

+ func (m *Map[K, V]) LSet(k K, v V)

Jan 8, 2024

Changes in this version

type FanOut

+ func (fo *FanOut[T]) DebugInfo() any

type Reader

+ func (r *Reader[R]) DebugInfo() any

type RunnerBase

+ func (r *RunnerBase[R]) DebugInfo() any

type Writer

+ func (w *Writer[W]) DebugInfo() any

Nov 26, 2023

Changes in this version

+ type Map struct

+ func NewMap[K comparable, V any]() (out *Map[K, V])

+ func (m *Map[K, V]) Delete(k K, lock bool)

+ func (m *Map[K, V]) Get(k K, lock bool) V

+ func (m *Map[K, V]) Has(k K, lock bool) bool

+ func (m *Map[K, V]) Lock()

+ func (m *Map[K, V]) RLock()

+ func (m *Map[K, V]) RUnlock()

+ func (m *Map[K, V]) Range(lock bool, meth func(K, V) bool)

+ func (m *Map[K, V]) Set(k K, v V, lock bool)

+ func (m *Map[K, V]) Unlock()

+ func (m *Map[K, V]) Update(actions func())

+ func (m *Map[K, V]) View(actions func())

Nov 14, 2023

Changes in this version

type FanOutCmd

+ CallbackChan chan error

May 10, 2023

Changes in this version

+ var InvalidHubClientError = errors.New("A hub client must have either a reader or a writer or both")

+ type Connector struct

+ OnClose func()

+ OnMessage func(msg Message[M]) error

+ func NewConnector(connect func() error, read ReaderFunc[M]) *Connector[M]

+ func (c *Connector[M]) Start()

+ func (c *Connector[M]) Stop()

type FanIn

+ OnChannelRemoved func(fi *FanIn[T], inchan <-chan T)

+ func (fi *FanIn[T]) Count() int

+ func (fi *FanIn[T]) RecvChan() chan T

type FanOut

+ func (fo *FanOut[T, U]) Add(output chan<- U, filter func(T) bool)

+ func (fo *FanOut[T, U]) Count() int

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

type FanOutCmd

+ SelfOwned bool

type HubClient

+ OnClose func()

+ Read HubReader[M]

+ Write HubWriter[M]

+ type HubReader func() Message[M]

+ type Input struct

+ type Message struct

+ Error error

+ Source interface{}

+ Value T

type Pipe

+ OnDone func(p *Pipe[T])

+ type Reader struct

+ OnDone func(r *Reader[R])

+ Read ReaderFunc[R]

+ func (rc *Reader[R]) RecvChan() <-chan Message[R]

+ type ReaderFunc func() (R, error)

+ type RunnerBase struct

+ func NewRunnerBase(stopVal C) RunnerBase[C]

+ func (r *RunnerBase[C]) IsRunning() bool

+ func (r *RunnerBase[C]) Stop() error

+ type Writer struct

+ Write WriterFunc[W]

+ func (wc *Writer[W]) Send(req W) bool

+ func (wc *Writer[W]) SendChan() chan W

+ type WriterFunc func(W) error

May 4, 2023

Changes in this version

+ type TopicIdType interface

May 3, 2023

Changes in this version

+ type EventType interface

type HubMessage

+ Error error

May 3, 2023

Changes in this version

+ type Broadcaster struct

+ func NewBroadcaster() *Broadcaster[M]

+ func (r *Broadcaster[M]) Add(client *HubClient[M]) error

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

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

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

+ func (r *Broadcaster[M]) RouteMessage(eventKey string, msg M, source *HubClient[M]) error

type FanOutCmd

+ Filter FilterFunc[T]

+ type FilterFunc func(T) bool

type KVRouter

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

type Router

+ Add func(client *HubClient[M]) error

Apr 27, 2023

Changes in this version

+ type Hub struct

+ func NewHub(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 struct

+ WriteMessage func(HubMessage[M]) error

+ 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 struct

+ AddedSubscriptions []string

+ Client *HubClient[M]

+ Pause bool

+ Quit bool

+ RemovedSubscriptions []string

+ type HubMessage struct

+ Callback chan M

+ EventKey string

+ Message M

+ type HubWriter func(HubMessage[M]) error

+ type KVRouter struct

+ func NewKVRouter() *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 Router interface

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

+ Remove func(client *HubClient[M]) error

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

+ RouteMessage func(HubMessage[M]) error

Sep 12, 2022

Changes in this version

+ type BatcherCmd struct

+ Channel chan T

+ Name string

+ type FanInCmd struct

+ AddedChannel <-chan T

+ Name string

+ RemovedChannel <-chan T

+ type FanOutCmd struct

+ AddedChannel chan T

+ Name string

+ RemovedChannel <-chan T

Sep 12, 2022

Changes in this version

+ type Batcher struct

+ FlushPeriod time.Duration

+ Joiner func(inputs []T) (outputs U)

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

+ func NewIDBatcher(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 FanIn

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

type FanOut

+ Mapper func(inputs T) U

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

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

type Pipe

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

Sep 12, 2022

Changes in this version

+ func IDFunc(input T) T

type FanIn

+ func NewFanIn() *FanIn[T]

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

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

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

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

type FanOut

+ Joiner func(inputs []T) (outputs U)

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

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

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

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

type Pipe

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

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

type ReaderChan

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

Sep 8, 2022

Changes in this version

+ func FanIn(readers ...chan T) (chan T, chan FanCmd[T])

+ func Pipe(input chan T, output chan T, onDone func()) (cmdChan chan string)

+ type FanCmd struct

+ Channel chan T

+ Name string

+ type FanOut struct

+ FlushPeriod time.Duration

+ func NewFanOut() *FanOut[T]

+ func (fo *FanOut[T]) AddOutputs(channels ...chan []T)

+ func (fo *FanOut[T]) NewOutput() (output chan []T)

+ func (fo *FanOut[T]) RemoveOutput(output chan []T)

+ func (fo *FanOut[T]) Send(value T)

+ func (fo *FanOut[T]) Start(output chan []T)

+ type ReaderChan struct

+ Read func() (*R, error)

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

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

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

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

+ func (rc *ReaderChan[R, D]) Start() (err error)

+ type ReaderWriterBase struct

+ Info D

+ WaitTime time.Duration

+ func (rwb *ReaderWriterBase[T, D]) Start() error

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

+ type ValueOrError struct

+ Error error

+ Value *T

+ type WriterChan struct

+ OnClose func()

+ Write func(msg *W) error

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

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

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

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

+ func (wc *WriterChan[W, D]) Start() (err error)