refactor: move picosh/pubsub into repo

......@@ -45,7 +45,6 @@ require (
	github.com/mmcdole/gofeed v1.3.0
	github.com/neurosnap/go-exif-remove v0.0.0-20221010134343-50d1e3c35577
	github.com/picosh/go-rsync-receiver v0.0.0-20250304201040-fcc11dd22d79
	github.com/picosh/pubsub v0.0.0-20241114191831-ec8f16c0eb88
	github.com/picosh/utils v0.0.0-20260125160622-5c3a9e231ec6
	github.com/pkg/sftp v1.13.9
	github.com/prometheus/client_golang v1.22.0

M go.sum

+0, -2

...	...		@@ -694,8 +694,6 @@ github.com/peterbourgon/diskv/v3 v3.0.1 h1:x06SQA46+PKIUftmEujdwSEpIx8kR+M9eLYsU
694	694		github.com/peterbourgon/diskv/v3 v3.0.1/go.mod h1:kJ5Ny7vLdARGU3WUuy6uzO6T0nb/2gWcT1JiBvRmb5o=
695	695		github.com/picosh/go-rsync-receiver v0.0.0-20250304201040-fcc11dd22d79 h1:MyB9P43hlQ6A2FoP9LGeiTBL3WKToW4gcWd6lQPg/Zg=
696	696		github.com/picosh/go-rsync-receiver v0.0.0-20250304201040-fcc11dd22d79/go.mod h1:4ZICsr6bESoHP8He9DqROlZiMw4hHHjcbDzhtTTDQzA=
697		-	github.com/picosh/pubsub v0.0.0-20241114191831-ec8f16c0eb88 h1:hdxE6rquHHw1/eeqS1b+ojLaxGtN8zOiTUclPwaVbPg=
698		-	github.com/picosh/pubsub v0.0.0-20241114191831-ec8f16c0eb88/go.mod h1:+9hDKIDHQCvGFigCVlIl589BwpT9R4boKhUVc/OgRU4=
699	697		github.com/picosh/utils v0.0.0-20260125160622-5c3a9e231ec6 h1:9KfCtfcx7vrSyGU1K9whdE1crll9Aq+nAZ6c0FzuzvE=
700	698		github.com/picosh/utils v0.0.0-20260125160622-5c3a9e231ec6/go.mod h1:HogYEyJ43IGXrOa3D/kjM1pkzNAyh+pejRyv8Eo//pk=
701	699		github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=

M pkg/apps/pipe/cli.go

+1, -1

......@@ -20,8 +20,8 @@ import (
	"github.com/gorilla/feeds"
	"github.com/picosh/pico/pkg/db"
	"github.com/picosh/pico/pkg/pssh"
	psub "github.com/picosh/pico/pkg/pubsub"
	"github.com/picosh/pico/pkg/shared"
	psub "github.com/picosh/pubsub"
	gossh "golang.org/x/crypto/ssh"
)

M pkg/apps/pipe/ssh.go

+1, -1

......@@ -9,8 +9,8 @@ import (
	"github.com/antoniomika/syncmap"
	"github.com/picosh/pico/pkg/db/postgres"
	"github.com/picosh/pico/pkg/pssh"
	psub "github.com/picosh/pico/pkg/pubsub"
	"github.com/picosh/pico/pkg/shared"
	psub "github.com/picosh/pubsub"
	"golang.org/x/crypto/ssh"
)

M pkg/apps/pipe/ssh_test.go

+1, -1

......@@ -16,8 +16,8 @@ import (
	"github.com/picosh/pico/pkg/db"
	"github.com/picosh/pico/pkg/db/stub"
	"github.com/picosh/pico/pkg/pssh"
	psub "github.com/picosh/pico/pkg/pubsub"
	"github.com/picosh/pico/pkg/shared"
	psub "github.com/picosh/pubsub"
	"github.com/prometheus/client_golang/prometheus"
	"golang.org/x/crypto/ssh"
)

A pkg/pubsub/README.md

+58, -0

......@@ -0,0 +1,58 @@
# pubsub

A generic pubsub implementation for Go.

```go
package main

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"

	"github.com/picosh/pubsub"
)

func main() {
	ctx := context.TODO()
	logger := slog.Default()
	broker := pubsub.NewMulticast(logger)

	chann := []*pubsub.Channel{
		pubsub.NewChannel("my-topic"),
	}

	go func() {
		writer := bytes.NewBufferString("my data")
		_ = broker.Pub(ctx, "pubID", writer, chann, false)
	}()

	reader := bytes.NewBufferString("")
	_ = broker.Sub(ctx, "subID", reader, chann, false)

	// result
	fmt.Println("data from pub:", reader)
}
```

## pubsub over ssh

The simplest pubsub system for everyday automation needs.

Using `wish` we can integrate our pubsub system into an SSH app.

[![asciicast](https://asciinema.org/a/674287.svg)](https://asciinema.org/a/674287)

```bash
# term 1
mkdir ./ssh_data
cat ~/.ssh/id_ed25519 ./ssh_data/authorized_keys
go run ./cmd/example

# term 2
ssh -p 2222 localhost sub xyz

# term 3
echo "hello world" | ssh -p 2222 localhost pub xyz
```

A pkg/pubsub/broker.go

+203, -0

......@@ -0,0 +1,203 @@
package pubsub

import (
	"errors"
	"io"
	"iter"
	"log/slog"
	"sync"
	"time"

	"github.com/antoniomika/syncmap"
)

/*
Broker receives published messages and dispatches the message to the
subscribing clients. An message contains a message topic that clients
subscribe to and brokers use these subscription lists for determining the
clients to receive the message.
*/
type Broker interface {
	GetChannels() iter.Seq2[string, *Channel]
	GetClients() iter.Seq2[string, *Client]
	Connect(*Client, []*Channel) (error, error)
}

type BaseBroker struct {
	Channels *syncmap.Map[string, *Channel]
	Logger   *slog.Logger
}

func (b *BaseBroker) Cleanup() {
	toRemove := []string{}
	for _, channel := range b.GetChannels() {
		count := 0

		for range channel.GetClients() {
			count++
		}

		if count == 0 {
			channel.Cleanup()
			toRemove = append(toRemove, channel.Topic)
		}
	}

	for _, channel := range toRemove {
		b.Channels.Delete(channel)
	}
}

func (b *BaseBroker) GetChannels() iter.Seq2[string, *Channel] {
	return b.Channels.Range
}

func (b *BaseBroker) GetClients() iter.Seq2[string, *Client] {
	return func(yield func(string, *Client) bool) {
		for _, channel := range b.GetChannels() {
			channel.Clients.Range(yield)
		}
	}
}

func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error) {
	for _, channel := range channels {
		dataChannel := b.ensureChannel(channel)
		dataChannel.Clients.Store(client.ID, client)
		client.Channels.Store(dataChannel.Topic, dataChannel)
		defer func() {
			client.Channels.Delete(channel.Topic)
			dataChannel.Clients.Delete(client.ID)

			client.Cleanup()

			count := 0
			for _, cl := range dataChannel.GetClients() {
				if cl.Direction == ChannelDirectionInput || cl.Direction == ChannelDirectionInputOutput {
					count++
				}
			}

			if count == 0 {
				for _, cl := range dataChannel.GetClients() {
					if !cl.KeepAlive {
						cl.Cleanup()
					}
				}
			}

			b.Cleanup()
		}()
	}

	var (
		inputErr  error
		outputErr error
		wg        sync.WaitGroup
	)

	// Pub
	if client.Direction == ChannelDirectionInput || client.Direction == ChannelDirectionInputOutput {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				data := make([]byte, 32*1024)
				n, err := client.ReadWriter.Read(data)
				data = data[:n]

				channelMessage := ChannelMessage{
					Data:      data,
					ClientID:  client.ID,
					Direction: ChannelDirectionInput,
				}

				if client.BlockWrite {
				mainLoop:
					for {
						count := 0
						for _, channel := range client.GetChannels() {
							for _, chanClient := range channel.GetClients() {
								if chanClient.Direction == ChannelDirectionOutput || chanClient.Direction == ChannelDirectionInputOutput {
									count++
								}
							}
						}

						if count > 0 {
							break mainLoop
						}

						select {
						case <-client.Done:
							break mainLoop
						case <-time.After(1 * time.Millisecond):
							continue
						}
					}
				}

				var sendwg sync.WaitGroup

				for _, channel := range client.GetChannels() {
					sendwg.Add(1)
					go func() {
						defer sendwg.Done()
						select {
						case channel.Data <- channelMessage:
						case <-client.Done:
						case <-channel.Done:
						}
					}()
				}

				sendwg.Wait()

				if err != nil {
					if errors.Is(err, io.EOF) {
						return
					}
					inputErr = err
					return
				}
			}
		}()
	}

	// Sub
	if client.Direction == ChannelDirectionOutput || client.Direction == ChannelDirectionInputOutput {
		wg.Add(1)
		go func() {
			defer wg.Done()
		mainLoop:
			for {
				select {
				case data, ok := <-client.Data:
					_, err := client.ReadWriter.Write(data.Data)
					if err != nil {
						outputErr = err
						break mainLoop
					}

					if !ok {
						break mainLoop
					}
				case <-client.Done:
					break mainLoop
				}
			}
		}()
	}

	wg.Wait()

	return inputErr, outputErr
}

func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	dataChannel.Handle()
	return dataChannel
}

var _ Broker = (*BaseBroker)(nil)

A pkg/pubsub/channel.go

+114, -0

......@@ -0,0 +1,114 @@
package pubsub

import (
	"iter"
	"sync"

	"github.com/antoniomika/syncmap"
)

type ChannelDirection int

func (d ChannelDirection) String() string {
	return [...]string{"input", "output", "inputoutput"}[d]
}

const (
	ChannelDirectionInput ChannelDirection = iota
	ChannelDirectionOutput
	ChannelDirectionInputOutput
)

type ChannelAction int

func (d ChannelAction) String() string {
	return [...]string{"data", "close"}[d]
}

const (
	ChannelActionData = iota
	ChannelActionClose
)

type ChannelMessage struct {
	Data      []byte
	ClientID  string
	Direction ChannelDirection
	Action    ChannelAction
}

func NewChannel(topic string) *Channel {
	return &Channel{
		Topic:   topic,
		Done:    make(chan struct{}),
		Data:    make(chan ChannelMessage),
		Clients: syncmap.New[string, *Client](),
	}
}

/*
Channel is a container for a topic.  It holds the list of clients and
a data channel to receive a message.
*/
type Channel struct {
	Topic       string
	Done        chan struct{}
	Data        chan ChannelMessage
	Clients     *syncmap.Map[string, *Client]
	handleOnce  sync.Once
	cleanupOnce sync.Once
}

func (c *Channel) GetClients() iter.Seq2[string, *Client] {
	return c.Clients.Range
}

func (c *Channel) Cleanup() {
	c.cleanupOnce.Do(func() {
		close(c.Done)
	})
}

func (c *Channel) Handle() {
	c.handleOnce.Do(func() {
		go func() {
			defer func() {
				for _, client := range c.GetClients() {
					client.Cleanup()
				}
			}()

			for {
				select {
				case <-c.Done:
					return
				case data, ok := <-c.Data:
					var wg sync.WaitGroup
					for _, client := range c.GetClients() {
						if client.Direction == ChannelDirectionInput || (client.ID == data.ClientID && !client.Replay) {
							continue
						}

						wg.Add(1)
						go func() {
							defer wg.Done()
							if !ok {
								client.onceData.Do(func() {
									close(client.Data)
								})
								return
							}

							select {
							case client.Data <- data:
							case <-client.Done:
							case <-c.Done:
							}
						}()
					}
					wg.Wait()
				}
			}
		}()
	})
}

A pkg/pubsub/client.go

+52, -0

......@@ -0,0 +1,52 @@
package pubsub

import (
	"io"
	"iter"
	"sync"

	"github.com/antoniomika/syncmap"
)

func NewClient(ID string, rw io.ReadWriter, direction ChannelDirection, blockWrite, replay, keepAlive bool) *Client {
	return &Client{
		ID:         ID,
		ReadWriter: rw,
		Direction:  direction,
		Channels:   syncmap.New[string, *Channel](),
		Done:       make(chan struct{}),
		Data:       make(chan ChannelMessage),
		Replay:     replay,
		BlockWrite: blockWrite,
		KeepAlive:  keepAlive,
	}
}

/*
Client is the container for holding state between multiple devices.  A
client has a direction (input, output, inputout) as well as a way to
send data to all the associated channels.
*/
type Client struct {
	ID         string
	ReadWriter io.ReadWriter
	Channels   *syncmap.Map[string, *Channel]
	Direction  ChannelDirection
	Done       chan struct{}
	Data       chan ChannelMessage
	Replay     bool
	BlockWrite bool
	KeepAlive  bool
	once       sync.Once
	onceData   sync.Once
}

func (c *Client) GetChannels() iter.Seq2[string, *Channel] {
	return c.Channels.Range
}

func (c *Client) Cleanup() {
	c.once.Do(func() {
		close(c.Done)
	})
}

A pkg/pubsub/multicast.go

+84, -0

......@@ -0,0 +1,84 @@
package pubsub

import (
	"context"
	"errors"
	"io"
	"iter"
	"log/slog"

	"github.com/antoniomika/syncmap"
)

/*
Multicast is a flexible, bidirectional broker.

It provides the most pure version of our PubSub interface which lets
end-developers build one-to-many connections between publishers and
subscribers and vice versa.

It doesn't provide any topic filtering capabilities and is only
concerned with sending data to and from an `io.ReadWriter` via our
channels.
*/
type Multicast struct {
	Broker
	Logger *slog.Logger
}

func NewMulticast(logger *slog.Logger) *Multicast {
	return &Multicast{
		Logger: logger,
		Broker: &BaseBroker{
			Channels: syncmap.New[string, *Channel](),
			Logger:   logger.With(slog.Bool("broker", true)),
		},
	}
}

func (p *Multicast) getClients(direction ChannelDirection) iter.Seq2[string, *Client] {
	return func(yield func(string, *Client) bool) {
		for clientID, client := range p.GetClients() {
			if client.Direction == direction {
				yield(clientID, client)
			}
		}
	}
}

func (p *Multicast) GetPipes() iter.Seq2[string, *Client] {
	return p.getClients(ChannelDirectionInputOutput)
}

func (p *Multicast) GetPubs() iter.Seq2[string, *Client] {
	return p.getClients(ChannelDirectionInput)
}

func (p *Multicast) GetSubs() iter.Seq2[string, *Client] {
	return p.getClients(ChannelDirectionOutput)
}

func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool) (error, error) {
	client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)

	go func() {
		<-ctx.Done()
		client.Cleanup()
	}()

	return p.Connect(client, channels)
}

func (p *Multicast) Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, replay bool) (error, error) {
	return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false)
}

func (p *Multicast) Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, blockWrite bool) error {
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false))
}

func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, keepAlive bool) error {
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
}

var _ PubSub = (*Multicast)(nil)

A pkg/pubsub/multicast_test.go

+164, -0

......@@ -0,0 +1,164 @@
package pubsub

import (
	"bytes"
	"context"
	"fmt"
	"log/slog"
	"sync"
	"testing"
)

type Buffer struct {
	b bytes.Buffer
	m sync.Mutex
}

func (b *Buffer) Read(p []byte) (n int, err error) {
	b.m.Lock()
	defer b.m.Unlock()
	return b.b.Read(p)
}
func (b *Buffer) Write(p []byte) (n int, err error) {
	b.m.Lock()
	defer b.m.Unlock()
	return b.b.Write(p)
}
func (b *Buffer) String() string {
	b.m.Lock()
	defer b.m.Unlock()
	return b.b.String()
}

func TestMulticastSubBlock(t *testing.T) {
	orderActual := ""
	orderExpected := "sub-pub-"
	actual := new(Buffer)
	expected := "some test data"
	name := "test-channel"
	syncer := make(chan int)

	cast := NewMulticast(slog.Default())

	var wg sync.WaitGroup
	wg.Add(2)

	channel := NewChannel(name)

	go func() {
		orderActual += "sub-"
		syncer <- 0
		fmt.Println(cast.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
		wg.Done()
	}()

	<-syncer

	go func() {
		orderActual += "pub-"
		fmt.Println(cast.Pub(context.TODO(), "2", &Buffer{b: *bytes.NewBufferString(expected)}, []*Channel{channel}, true))
		wg.Done()
	}()

	wg.Wait()

	if orderActual != orderExpected {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	}
}

func TestMulticastPubBlock(t *testing.T) {
	orderActual := ""
	orderExpected := "pub-sub-"
	actual := new(Buffer)
	expected := "some test data"
	name := "test-channel"
	syncer := make(chan int)

	cast := NewMulticast(slog.Default())

	var wg sync.WaitGroup
	wg.Add(2)

	channel := NewChannel(name)

	go func() {
		orderActual += "pub-"
		syncer <- 0
		fmt.Println(cast.Pub(context.TODO(), "1", &Buffer{b: *bytes.NewBufferString(expected)}, []*Channel{channel}, true))
		wg.Done()
	}()

	<-syncer

	go func() {
		orderActual += "sub-"
		wg.Done()
		fmt.Println(cast.Sub(context.TODO(), "2", actual, []*Channel{channel}, false))
	}()

	wg.Wait()

	if orderActual != orderExpected {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	}
}

func TestMulticastMultSubs(t *testing.T) {
	orderActual := ""
	orderExpected := "sub-sub-pub-"
	actual := new(Buffer)
	actualOther := new(Buffer)
	expected := "some test data"
	name := "test-channel"
	syncer := make(chan int)

	cast := NewMulticast(slog.Default())

	var wg sync.WaitGroup
	wg.Add(3)

	channel := NewChannel(name)

	go func() {
		orderActual += "sub-"
		syncer <- 0
		fmt.Println(cast.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
		wg.Done()
	}()

	<-syncer

	go func() {
		orderActual += "sub-"
		syncer <- 0
		fmt.Println(cast.Sub(context.TODO(), "2", actualOther, []*Channel{channel}, false))
		wg.Done()
	}()

	<-syncer

	go func() {
		orderActual += "pub-"
		fmt.Println(cast.Pub(context.TODO(), "3", &Buffer{b: *bytes.NewBufferString(expected)}, []*Channel{channel}, true))
		wg.Done()
	}()

	wg.Wait()

	if orderActual != orderExpected {
		t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	}
	if actual.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	}
	if actualOther.String() != expected {
		t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther, expected)
	}
}

A pkg/pubsub/pubsub.go

+26, -0

......@@ -0,0 +1,26 @@
package pubsub

import (
	"context"
	"io"
	"iter"
)

/*
PubSub is our take on a basic publisher and subscriber interface.

It has a few notable requirements:
- Each operation must accept an array of channels
- A way to send, receive, and stream data between clients

PubSub also inherits the properties of a Broker.
*/
type PubSub interface {
	Broker
	GetPubs() iter.Seq2[string, *Client]
	GetSubs() iter.Seq2[string, *Client]
	GetPipes() iter.Seq2[string, *Client]
	Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, replay bool) (error, error)
	Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, keepAlive bool) error
	Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, blockWrite bool) error
}

...	...		@@ -45,7 +45,6 @@ require (
45	45		github.com/mmcdole/gofeed v1.3.0
46	46		github.com/neurosnap/go-exif-remove v0.0.0-20221010134343-50d1e3c35577
47	47		github.com/picosh/go-rsync-receiver v0.0.0-20250304201040-fcc11dd22d79
48		-	github.com/picosh/pubsub v0.0.0-20241114191831-ec8f16c0eb88
49	48		github.com/picosh/utils v0.0.0-20260125160622-5c3a9e231ec6
50	49		github.com/pkg/sftp v1.13.9
51	50		github.com/prometheus/client_golang v1.22.0

...	...		@@ -20,8 +20,8 @@ import (
20	20		"github.com/gorilla/feeds"
21	21		"github.com/picosh/pico/pkg/db"
22	22		"github.com/picosh/pico/pkg/pssh"
	23	+	psub "github.com/picosh/pico/pkg/pubsub"
23	24		"github.com/picosh/pico/pkg/shared"
24		-	psub "github.com/picosh/pubsub"
25	25		gossh "golang.org/x/crypto/ssh"
26	26		)
27	27

...	...		@@ -9,8 +9,8 @@ import (
9	9		"github.com/antoniomika/syncmap"
10	10		"github.com/picosh/pico/pkg/db/postgres"
11	11		"github.com/picosh/pico/pkg/pssh"
	12	+	psub "github.com/picosh/pico/pkg/pubsub"
12	13		"github.com/picosh/pico/pkg/shared"
13		-	psub "github.com/picosh/pubsub"
14	14		"golang.org/x/crypto/ssh"
15	15		)
16	16

...	...		@@ -16,8 +16,8 @@ import (
16	16		"github.com/picosh/pico/pkg/db"
17	17		"github.com/picosh/pico/pkg/db/stub"
18	18		"github.com/picosh/pico/pkg/pssh"
	19	+	psub "github.com/picosh/pico/pkg/pubsub"
19	20		"github.com/picosh/pico/pkg/shared"
20		-	psub "github.com/picosh/pubsub"
21	21		"github.com/prometheus/client_golang/prometheus"
22	22		"golang.org/x/crypto/ssh"
23	23		)

...	...		@@ -0,0 +1,58 @@
	1	+	# pubsub
	2	+
	3	+	A generic pubsub implementation for Go.
	4	+
	5	+	```go
	6	+	package main
	7	+
	8	+	import (
	9	+	"bytes"
	10	+	"context"
	11	+	"fmt"
	12	+	"log/slog"
	13	+
	14	+	"github.com/picosh/pubsub"
	15	+	)
	16	+
	17	+	func main() {
	18	+	ctx := context.TODO()
	19	+	logger := slog.Default()
	20	+	broker := pubsub.NewMulticast(logger)
	21	+
	22	+	chann := []*pubsub.Channel{
	23	+	pubsub.NewChannel("my-topic"),
	24	+	}
	25	+
	26	+	go func() {
	27	+	writer := bytes.NewBufferString("my data")
	28	+	_ = broker.Pub(ctx, "pubID", writer, chann, false)
	29	+	}()
	30	+
	31	+	reader := bytes.NewBufferString("")
	32	+	_ = broker.Sub(ctx, "subID", reader, chann, false)
	33	+
	34	+	// result
	35	+	fmt.Println("data from pub:", reader)
	36	+	}
	37	+	```
	38	+
	39	+	## pubsub over ssh
	40	+
	41	+	The simplest pubsub system for everyday automation needs.
	42	+
	43	+	Using `wish` we can integrate our pubsub system into an SSH app.
	44	+
	45	+	[![asciicast](https://asciinema.org/a/674287.svg)](https://asciinema.org/a/674287)
	46	+
	47	+	```bash
	48	+	# term 1
	49	+	mkdir ./ssh_data
	50	+	cat ~/.ssh/id_ed25519 ./ssh_data/authorized_keys
	51	+	go run ./cmd/example
	52	+
	53	+	# term 2
	54	+	ssh -p 2222 localhost sub xyz
	55	+
	56	+	# term 3
	57	+	echo "hello world" \| ssh -p 2222 localhost pub xyz
	58	+	```

...	...		@@ -0,0 +1,203 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"io"
	6	+	"iter"
	7	+	"log/slog"
	8	+	"sync"
	9	+	"time"
	10	+
	11	+	"github.com/antoniomika/syncmap"
	12	+	)
	13	+
	14	+	/*
	15	+	Broker receives published messages and dispatches the message to the
	16	+	subscribing clients. An message contains a message topic that clients
	17	+	subscribe to and brokers use these subscription lists for determining the
	18	+	clients to receive the message.
	19	+	*/
	20	+	type Broker interface {
	21	+	GetChannels() iter.Seq2[string, *Channel]
	22	+	GetClients() iter.Seq2[string, *Client]
	23	+	Connect(Client, []Channel) (error, error)
	24	+	}
	25	+
	26	+	type BaseBroker struct {
	27	+	Channels syncmap.Map[string, Channel]
	28	+	Logger *slog.Logger
	29	+	}
	30	+
	31	+	func (b *BaseBroker) Cleanup() {
	32	+	toRemove := []string{}
	33	+	for _, channel := range b.GetChannels() {
	34	+	count := 0
	35	+
	36	+	for range channel.GetClients() {
	37	+	count++
	38	+	}
	39	+
	40	+	if count == 0 {
	41	+	channel.Cleanup()
	42	+	toRemove = append(toRemove, channel.Topic)
	43	+	}
	44	+	}
	45	+
	46	+	for _, channel := range toRemove {
	47	+	b.Channels.Delete(channel)
	48	+	}
	49	+	}
	50	+
	51	+	func (b BaseBroker) GetChannels() iter.Seq2[string, Channel] {
	52	+	return b.Channels.Range
	53	+	}
	54	+
	55	+	func (b BaseBroker) GetClients() iter.Seq2[string, Client] {
	56	+	return func(yield func(string, *Client) bool) {
	57	+	for _, channel := range b.GetChannels() {
	58	+	channel.Clients.Range(yield)
	59	+	}
	60	+	}
	61	+	}
	62	+
	63	+	func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error) {
	64	+	for _, channel := range channels {
	65	+	dataChannel := b.ensureChannel(channel)
	66	+	dataChannel.Clients.Store(client.ID, client)
	67	+	client.Channels.Store(dataChannel.Topic, dataChannel)
	68	+	defer func() {
	69	+	client.Channels.Delete(channel.Topic)
	70	+	dataChannel.Clients.Delete(client.ID)
	71	+
	72	+	client.Cleanup()
	73	+
	74	+	count := 0
	75	+	for _, cl := range dataChannel.GetClients() {
	76	+	if cl.Direction == ChannelDirectionInput \|\| cl.Direction == ChannelDirectionInputOutput {
	77	+	count++
	78	+	}
	79	+	}
	80	+
	81	+	if count == 0 {
	82	+	for _, cl := range dataChannel.GetClients() {
	83	+	if !cl.KeepAlive {
	84	+	cl.Cleanup()
	85	+	}
	86	+	}
	87	+	}
	88	+
	89	+	b.Cleanup()
	90	+	}()
	91	+	}
	92	+
	93	+	var (
	94	+	inputErr error
	95	+	outputErr error
	96	+	wg sync.WaitGroup
	97	+	)
	98	+
	99	+	// Pub
	100	+	if client.Direction == ChannelDirectionInput \|\| client.Direction == ChannelDirectionInputOutput {
	101	+	wg.Add(1)
	102	+	go func() {
	103	+	defer wg.Done()
	104	+	for {
	105	+	data := make([]byte, 32*1024)
	106	+	n, err := client.ReadWriter.Read(data)
	107	+	data = data[:n]
	108	+
	109	+	channelMessage := ChannelMessage{
	110	+	Data: data,
	111	+	ClientID: client.ID,
	112	+	Direction: ChannelDirectionInput,
	113	+	}
	114	+
	115	+	if client.BlockWrite {
	116	+	mainLoop:
	117	+	for {
	118	+	count := 0
	119	+	for _, channel := range client.GetChannels() {
	120	+	for _, chanClient := range channel.GetClients() {
	121	+	if chanClient.Direction == ChannelDirectionOutput \|\| chanClient.Direction == ChannelDirectionInputOutput {
	122	+	count++
	123	+	}
	124	+	}
	125	+	}
	126	+
	127	+	if count > 0 {
	128	+	break mainLoop
	129	+	}
	130	+
	131	+	select {
	132	+	case <-client.Done:
	133	+	break mainLoop
	134	+	case <-time.After(1 * time.Millisecond):
	135	+	continue
	136	+	}
	137	+	}
	138	+	}
	139	+
	140	+	var sendwg sync.WaitGroup
	141	+
	142	+	for _, channel := range client.GetChannels() {
	143	+	sendwg.Add(1)
	144	+	go func() {
	145	+	defer sendwg.Done()
	146	+	select {
	147	+	case channel.Data <- channelMessage:
	148	+	case <-client.Done:
	149	+	case <-channel.Done:
	150	+	}
	151	+	}()
	152	+	}
	153	+
	154	+	sendwg.Wait()
	155	+
	156	+	if err != nil {
	157	+	if errors.Is(err, io.EOF) {
	158	+	return
	159	+	}
	160	+	inputErr = err
	161	+	return
	162	+	}
	163	+	}
	164	+	}()
	165	+	}
	166	+
	167	+	// Sub
	168	+	if client.Direction == ChannelDirectionOutput \|\| client.Direction == ChannelDirectionInputOutput {
	169	+	wg.Add(1)
	170	+	go func() {
	171	+	defer wg.Done()
	172	+	mainLoop:
	173	+	for {
	174	+	select {
	175	+	case data, ok := <-client.Data:
	176	+	_, err := client.ReadWriter.Write(data.Data)
	177	+	if err != nil {
	178	+	outputErr = err
	179	+	break mainLoop
	180	+	}
	181	+
	182	+	if !ok {
	183	+	break mainLoop
	184	+	}
	185	+	case <-client.Done:
	186	+	break mainLoop
	187	+	}
	188	+	}
	189	+	}()
	190	+	}
	191	+
	192	+	wg.Wait()
	193	+
	194	+	return inputErr, outputErr
	195	+	}
	196	+
	197	+	func (b BaseBroker) ensureChannel(channel Channel) *Channel {
	198	+	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	199	+	dataChannel.Handle()
	200	+	return dataChannel
	201	+	}
	202	+
	203	+	var _ Broker = (*BaseBroker)(nil)

...	...		@@ -0,0 +1,114 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"iter"
	5	+	"sync"
	6	+
	7	+	"github.com/antoniomika/syncmap"
	8	+	)
	9	+
	10	+	type ChannelDirection int
	11	+
	12	+	func (d ChannelDirection) String() string {
	13	+	return [...]string{"input", "output", "inputoutput"}[d]
	14	+	}
	15	+
	16	+	const (
	17	+	ChannelDirectionInput ChannelDirection = iota
	18	+	ChannelDirectionOutput
	19	+	ChannelDirectionInputOutput
	20	+	)
	21	+
	22	+	type ChannelAction int
	23	+
	24	+	func (d ChannelAction) String() string {
	25	+	return [...]string{"data", "close"}[d]
	26	+	}
	27	+
	28	+	const (
	29	+	ChannelActionData = iota
	30	+	ChannelActionClose
	31	+	)
	32	+
	33	+	type ChannelMessage struct {
	34	+	Data []byte
	35	+	ClientID string
	36	+	Direction ChannelDirection
	37	+	Action ChannelAction
	38	+	}
	39	+
	40	+	func NewChannel(topic string) *Channel {
	41	+	return &Channel{
	42	+	Topic: topic,
	43	+	Done: make(chan struct{}),
	44	+	Data: make(chan ChannelMessage),
	45	+	Clients: syncmap.New[string, *Client](),
	46	+	}
	47	+	}
	48	+
	49	+	/*
	50	+	Channel is a container for a topic. It holds the list of clients and
	51	+	a data channel to receive a message.
	52	+	*/
	53	+	type Channel struct {
	54	+	Topic string
	55	+	Done chan struct{}
	56	+	Data chan ChannelMessage
	57	+	Clients syncmap.Map[string, Client]
	58	+	handleOnce sync.Once
	59	+	cleanupOnce sync.Once
	60	+	}
	61	+
	62	+	func (c Channel) GetClients() iter.Seq2[string, Client] {
	63	+	return c.Clients.Range
	64	+	}
	65	+
	66	+	func (c *Channel) Cleanup() {
	67	+	c.cleanupOnce.Do(func() {
	68	+	close(c.Done)
	69	+	})
	70	+	}
	71	+
	72	+	func (c *Channel) Handle() {
	73	+	c.handleOnce.Do(func() {
	74	+	go func() {
	75	+	defer func() {
	76	+	for _, client := range c.GetClients() {
	77	+	client.Cleanup()
	78	+	}
	79	+	}()
	80	+
	81	+	for {
	82	+	select {
	83	+	case <-c.Done:
	84	+	return
	85	+	case data, ok := <-c.Data:
	86	+	var wg sync.WaitGroup
	87	+	for _, client := range c.GetClients() {
	88	+	if client.Direction == ChannelDirectionInput \|\| (client.ID == data.ClientID && !client.Replay) {
	89	+	continue
	90	+	}
	91	+
	92	+	wg.Add(1)
	93	+	go func() {
	94	+	defer wg.Done()
	95	+	if !ok {
	96	+	client.onceData.Do(func() {
	97	+	close(client.Data)
	98	+	})
	99	+	return
	100	+	}
	101	+
	102	+	select {
	103	+	case client.Data <- data:
	104	+	case <-client.Done:
	105	+	case <-c.Done:
	106	+	}
	107	+	}()
	108	+	}
	109	+	wg.Wait()
	110	+	}
	111	+	}
	112	+	}()
	113	+	})
	114	+	}

...	...		@@ -0,0 +1,52 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"io"
	5	+	"iter"
	6	+	"sync"
	7	+
	8	+	"github.com/antoniomika/syncmap"
	9	+	)
	10	+
	11	+	func 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	+	Data: make(chan ChannelMessage),
	19	+	Replay: replay,
	20	+	BlockWrite: blockWrite,
	21	+	KeepAlive: keepAlive,
	22	+	}
	23	+	}
	24	+
	25	+	/*
	26	+	Client is the container for holding state between multiple devices. A
	27	+	client has a direction (input, output, inputout) as well as a way to
	28	+	send data to all the associated channels.
	29	+	*/
	30	+	type Client struct {
	31	+	ID string
	32	+	ReadWriter io.ReadWriter
	33	+	Channels syncmap.Map[string, Channel]
	34	+	Direction ChannelDirection
	35	+	Done chan struct{}
	36	+	Data chan ChannelMessage
	37	+	Replay bool
	38	+	BlockWrite bool
	39	+	KeepAlive bool
	40	+	once sync.Once
	41	+	onceData sync.Once
	42	+	}
	43	+
	44	+	func (c Client) GetChannels() iter.Seq2[string, Channel] {
	45	+	return c.Channels.Range
	46	+	}
	47	+
	48	+	func (c *Client) Cleanup() {
	49	+	c.once.Do(func() {
	50	+	close(c.Done)
	51	+	})
	52	+	}

...	...		@@ -0,0 +1,84 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"io"
	7	+	"iter"
	8	+	"log/slog"
	9	+
	10	+	"github.com/antoniomika/syncmap"
	11	+	)
	12	+
	13	+	/*
	14	+	Multicast is a flexible, bidirectional broker.
	15	+
	16	+	It provides the most pure version of our PubSub interface which lets
	17	+	end-developers build one-to-many connections between publishers and
	18	+	subscribers and vice versa.
	19	+
	20	+	It doesn't provide any topic filtering capabilities and is only
	21	+	concerned with sending data to and from an `io.ReadWriter` via our
	22	+	channels.
	23	+	*/
	24	+	type Multicast struct {
	25	+	Broker
	26	+	Logger *slog.Logger
	27	+	}
	28	+
	29	+	func NewMulticast(logger slog.Logger) Multicast {
	30	+	return &Multicast{
	31	+	Logger: logger,
	32	+	Broker: &BaseBroker{
	33	+	Channels: syncmap.New[string, *Channel](),
	34	+	Logger: logger.With(slog.Bool("broker", true)),
	35	+	},
	36	+	}
	37	+	}
	38	+
	39	+	func (p Multicast) getClients(direction ChannelDirection) iter.Seq2[string, Client] {
	40	+	return func(yield func(string, *Client) bool) {
	41	+	for clientID, client := range p.GetClients() {
	42	+	if client.Direction == direction {
	43	+	yield(clientID, client)
	44	+	}
	45	+	}
	46	+	}
	47	+	}
	48	+
	49	+	func (p Multicast) GetPipes() iter.Seq2[string, Client] {
	50	+	return p.getClients(ChannelDirectionInputOutput)
	51	+	}
	52	+
	53	+	func (p Multicast) GetPubs() iter.Seq2[string, Client] {
	54	+	return p.getClients(ChannelDirectionInput)
	55	+	}
	56	+
	57	+	func (p Multicast) GetSubs() iter.Seq2[string, Client] {
	58	+	return p.getClients(ChannelDirectionOutput)
	59	+	}
	60	+
	61	+	func (p Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool) (error, error) {
	62	+	client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)
	63	+
	64	+	go func() {
	65	+	<-ctx.Done()
	66	+	client.Cleanup()
	67	+	}()
	68	+
	69	+	return p.Connect(client, channels)
	70	+	}
	71	+
	72	+	func (p Multicast) Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, replay bool) (error, error) {
	73	+	return p.connect(ctx, ID, rw, channels, ChannelDirectionInputOutput, false, replay, false)
	74	+	}
	75	+
	76	+	func (p Multicast) Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, blockWrite bool) error {
	77	+	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionInput, blockWrite, false, false))
	78	+	}
	79	+
	80	+	func (p Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, keepAlive bool) error {
	81	+	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
	82	+	}
	83	+
	84	+	var _ PubSub = (*Multicast)(nil)

...	...		@@ -0,0 +1,164 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"fmt"
	7	+	"log/slog"
	8	+	"sync"
	9	+	"testing"
	10	+	)
	11	+
	12	+	type Buffer struct {
	13	+	b bytes.Buffer
	14	+	m sync.Mutex
	15	+	}
	16	+
	17	+	func (b *Buffer) Read(p []byte) (n int, err error) {
	18	+	b.m.Lock()
	19	+	defer b.m.Unlock()
	20	+	return b.b.Read(p)
	21	+	}
	22	+	func (b *Buffer) Write(p []byte) (n int, err error) {
	23	+	b.m.Lock()
	24	+	defer b.m.Unlock()
	25	+	return b.b.Write(p)
	26	+	}
	27	+	func (b *Buffer) String() string {
	28	+	b.m.Lock()
	29	+	defer b.m.Unlock()
	30	+	return b.b.String()
	31	+	}
	32	+
	33	+	func TestMulticastSubBlock(t *testing.T) {
	34	+	orderActual := ""
	35	+	orderExpected := "sub-pub-"
	36	+	actual := new(Buffer)
	37	+	expected := "some test data"
	38	+	name := "test-channel"
	39	+	syncer := make(chan int)
	40	+
	41	+	cast := NewMulticast(slog.Default())
	42	+
	43	+	var wg sync.WaitGroup
	44	+	wg.Add(2)
	45	+
	46	+	channel := NewChannel(name)
	47	+
	48	+	go func() {
	49	+	orderActual += "sub-"
	50	+	syncer <- 0
	51	+	fmt.Println(cast.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
	52	+	wg.Done()
	53	+	}()
	54	+
	55	+	<-syncer
	56	+
	57	+	go func() {
	58	+	orderActual += "pub-"
	59	+	fmt.Println(cast.Pub(context.TODO(), "2", &Buffer{b: bytes.NewBufferString(expected)}, []Channel{channel}, true))
	60	+	wg.Done()
	61	+	}()
	62	+
	63	+	wg.Wait()
	64	+
	65	+	if orderActual != orderExpected {
	66	+	t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	67	+	}
	68	+	if actual.String() != expected {
	69	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	70	+	}
	71	+	}
	72	+
	73	+	func TestMulticastPubBlock(t *testing.T) {
	74	+	orderActual := ""
	75	+	orderExpected := "pub-sub-"
	76	+	actual := new(Buffer)
	77	+	expected := "some test data"
	78	+	name := "test-channel"
	79	+	syncer := make(chan int)
	80	+
	81	+	cast := NewMulticast(slog.Default())
	82	+
	83	+	var wg sync.WaitGroup
	84	+	wg.Add(2)
	85	+
	86	+	channel := NewChannel(name)
	87	+
	88	+	go func() {
	89	+	orderActual += "pub-"
	90	+	syncer <- 0
	91	+	fmt.Println(cast.Pub(context.TODO(), "1", &Buffer{b: bytes.NewBufferString(expected)}, []Channel{channel}, true))
	92	+	wg.Done()
	93	+	}()
	94	+
	95	+	<-syncer
	96	+
	97	+	go func() {
	98	+	orderActual += "sub-"
	99	+	wg.Done()
	100	+	fmt.Println(cast.Sub(context.TODO(), "2", actual, []*Channel{channel}, false))
	101	+	}()
	102	+
	103	+	wg.Wait()
	104	+
	105	+	if orderActual != orderExpected {
	106	+	t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	107	+	}
	108	+	if actual.String() != expected {
	109	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	110	+	}
	111	+	}
	112	+
	113	+	func TestMulticastMultSubs(t *testing.T) {
	114	+	orderActual := ""
	115	+	orderExpected := "sub-sub-pub-"
	116	+	actual := new(Buffer)
	117	+	actualOther := new(Buffer)
	118	+	expected := "some test data"
	119	+	name := "test-channel"
	120	+	syncer := make(chan int)
	121	+
	122	+	cast := NewMulticast(slog.Default())
	123	+
	124	+	var wg sync.WaitGroup
	125	+	wg.Add(3)
	126	+
	127	+	channel := NewChannel(name)
	128	+
	129	+	go func() {
	130	+	orderActual += "sub-"
	131	+	syncer <- 0
	132	+	fmt.Println(cast.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
	133	+	wg.Done()
	134	+	}()
	135	+
	136	+	<-syncer
	137	+
	138	+	go func() {
	139	+	orderActual += "sub-"
	140	+	syncer <- 0
	141	+	fmt.Println(cast.Sub(context.TODO(), "2", actualOther, []*Channel{channel}, false))
	142	+	wg.Done()
	143	+	}()
	144	+
	145	+	<-syncer
	146	+
	147	+	go func() {
	148	+	orderActual += "pub-"
	149	+	fmt.Println(cast.Pub(context.TODO(), "3", &Buffer{b: bytes.NewBufferString(expected)}, []Channel{channel}, true))
	150	+	wg.Done()
	151	+	}()
	152	+
	153	+	wg.Wait()
	154	+
	155	+	if orderActual != orderExpected {
	156	+	t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
	157	+	}
	158	+	if actual.String() != expected {
	159	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual, expected)
	160	+	}
	161	+	if actualOther.String() != expected {
	162	+	t.Fatalf("\nactual:(%s)\nexpected:(%s)", actualOther, expected)
	163	+	}
	164	+	}

...	...		@@ -0,0 +1,26 @@
	1	+	package pubsub
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"iter"
	7	+	)
	8	+
	9	+	/*
	10	+	PubSub is our take on a basic publisher and subscriber interface.
	11	+
	12	+	It has a few notable requirements:
	13	+	- Each operation must accept an array of channels
	14	+	- A way to send, receive, and stream data between clients
	15	+
	16	+	PubSub also inherits the properties of a Broker.
	17	+	*/
	18	+	type PubSub interface {
	19	+	Broker
	20	+	GetPubs() iter.Seq2[string, *Client]
	21	+	GetSubs() iter.Seq2[string, *Client]
	22	+	GetPipes() iter.Seq2[string, *Client]
	23	+	Pipe(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, replay bool) (error, error)
	24	+	Sub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, keepAlive bool) error
	25	+	Pub(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, blockWrite bool) error
	26	+	}

Commit 4b5c506

Commit `4b5c506`