streampool

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

type MessageQueueId interface {
	MessageQueueId() string
	DrpcMessage() drpc.Message
}

type Option func(*streamPool)

type PeerGetter func(ctx context.Context) (peers []peer.Peer, err error)

type SizeableMessage interface {
	Size() int
}

type StreamConfig struct {
	// SendQueueSize size of the queue for write per peer
	SendQueueSize int
	// DialQueueWorkers how many workers will dial to peers
	DialQueueWorkers int
	// DialQueueSize size of the dial queue
	DialQueueSize int
}

type StreamPool interface {
	app.ComponentRunnable
	// AddStream adds new outgoing stream into the pool
	AddStream(stream drpc.Stream, queueSize int, tags ...string) (err error)
	// ReadStream adds new incoming stream and synchronously reads it
	ReadStream(stream drpc.Stream, queueSize int, tags ...string) (err error)
	// Send sends a message to given peers. A stream will be opened if it is not cached before. Works async.
	Send(ctx context.Context, msg drpc.Message, target PeerGetter) (err error)
	// SendById sends a message to given peerIds. Works only if stream exists
	SendById(ctx context.Context, msg drpc.Message, peerIds ...string) (err error)
	// Broadcast sends a message to all peers with given tags. Works async.
	Broadcast(ctx context.Context, msg drpc.Message, tags ...string) (err error)
	// AddTagsCtx adds tags to stream, stream will be extracted from ctx
	AddTagsCtx(ctx context.Context, tags ...string) error
	// RemoveTagsCtx removes tags from stream, stream will be extracted from ctx
	RemoveTagsCtx(ctx context.Context, tags ...string) error
	// RemoveTagsById removes tags from a specific stream by id, for callers that
	// track streamId out of band. Missing streams and tags are ignored.
	RemoveTagsById(streamId uint32, tags ...string) error
	// Streams gets all streams for specific tags
	Streams(tags ...string) (streams []drpc.Stream)
}