conc

Documentation

Index

• func FanIn[T any](readers ...chan T) (chan T, chan FanCmd[T])
• func Pipe[T any](input chan T, output chan T, onDone func()) (cmdChan chan string)
• type FanCmd
• type FanOut
  • func NewFanOut[T any]() *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
  • func NewReader[R any, D any](read func() (*R, error)) *ReaderChan[R, D]
  • func (rc *ReaderChan[R, D]) Channel() chan ValueOrError[R]
  • func (ch *ReaderChan[R, D]) IsRunning() bool
  • func (rc *ReaderChan[R, D]) Start() (err error)
  • func (ch *ReaderChan[T, D]) Stop() error
• type ReaderWriterBase
  • func (rwb *ReaderWriterBase[T, D]) Start() error
  • func (rwb *ReaderWriterBase[T, D]) WaitForFinish()
• 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 (wc *WriterChan[W, D]) Start() (err error)
  • func (ch *WriterChan[T, D]) Stop() error

Functions

func FanIn

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

func Pipe

func Pipe[T any](input chan T, output chan T, onDone func()) (cmdChan chan string)

Types

type FanCmd

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

type FanOut

type FanOut[T any] struct {
	FlushPeriod time.Duration
	// contains filtered or unexported fields
}

func NewFanOut

func NewFanOut[T any]() *FanOut[T]

func (*FanOut[T]) AddOutputs

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

func (*FanOut[T]) NewOutput

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

func (*FanOut[T]) RemoveOutput

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

func (*FanOut[T]) Send

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

func (*FanOut[T]) Start

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

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]) Channel

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

*

• Returns the conn's reader channel.

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]) Start

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

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]) Start

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

func (*ReaderWriterBase[T, D]) WaitForFinish

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

*

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

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[W, D]) Start

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

Start writer goroutine

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.