1package pubsub
 2
 3import (
 4	"io"
 5	"iter"
 6	"sync"
 7
 8	"github.com/antoniomika/syncmap"
 9)
10
11func NewClient(ID string, rw io.ReadWriter, direction ChannelDirection, blockWrite, replay, keepAlive bool) *Client {
12	return &Client{
13		ID:         ID,
14		ReadWriter: rw,
15		Direction:  direction,
16		Channels:   syncmap.New[string, *Channel](),
17		Done:       make(chan struct{}),
18		finish:     make(chan struct{}),
19		Data:       make(chan ChannelMessage),
20		Replay:     replay,
21		BlockWrite: blockWrite,
22		KeepAlive:  keepAlive,
23	}
24}
25
26/*
27Client is the container for holding state between multiple devices.  A
28client has a direction (input, output, inputout) as well as a way to
29send data to all the associated channels.
30*/
31type Client struct {
32	ID         string
33	ReadWriter io.ReadWriter
34	Channels   *syncmap.Map[string, *Channel]
35	Direction  ChannelDirection
36	Done       chan struct{}
37	Data       chan ChannelMessage
38	Replay     bool
39	BlockWrite bool
40	KeepAlive  bool
41	once       sync.Once
42	onceData   sync.Once
43	finish     chan struct{}
44	finishOnce sync.Once
45}
46
47func (c *Client) GetChannels() iter.Seq2[string, *Channel] {
48	return c.Channels.Range
49}
50
51// Cleanup ends the client. Its ReadWriter stays open for whoever opened it
52// to close, so an ssh session can still send its exit status.
53func (c *Client) Cleanup() {
54	c.once.Do(func() {
55		close(c.Done)
56	})
57}
58
59// Finish ends a subscriber after it writes the message it was last sent, so
60// a publisher that leaves doesn't cut off its own last message.
61func (c *Client) Finish() {
62	c.finishOnce.Do(func() {
63		close(c.finish)
64	})
65}