fix: multicast

feat: docker
chore: ci

.github/docker-compose.gha.yml

@@ -0,0 +1,7 @@
+services:
+  pubsub:
+    build:
+      cache_from:
+        - type=gha
+      cache_to:
+        - type=gha,mode=max

.github/workflows/build.yml

@@ -0,0 +1,33 @@
+name: Build
+
+on:
+  push:
+    branches:
+      - main
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - name: Expose GitHub Runtime
+        uses: crazy-max/ghaction-github-runtime@v3
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+        with:
+          version: latest
+      - name: Login to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - name: Build and push Docker image
+        run: |
+          docker buildx bake \
+            -f docker-compose.yml \
+            -f .github/docker-compose.gha.yml \
+            --set *.platform=linux/arm64,linux/amd64,linux/arm/v7 \
+            --push pubsub

Dockerfile

@@ -0,0 +1,30 @@
+FROM --platform=$BUILDPLATFORM golang:1.23-alpine AS builder
+
+ENV CGO_ENABLED=0
+
+WORKDIR /usr/src/app
+
+COPY go.* .
+
+RUN --mount=type=cache,target=/go/pkg/,rw \
+  --mount=type=cache,target=/root/.cache/,rw \
+  go mod download
+
+COPY . .
+
+ARG TARGETOS
+ARG TARGETARCH
+
+ENV GOOS=${TARGETOS} GOARCH=${TARGETARCH}
+
+RUN --mount=type=cache,target=/go/pkg/,rw \
+  --mount=type=cache,target=/root/.cache/,rw \
+  go build -o pubsub ./cmd/authorized_keys
+
+FROM alpine
+
+RUN apk add --no-cache curl
+
+COPY --from=builder /usr/src/app/pubsub /pubsub
+
+ENTRYPOINT [ "/pubsub" ]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 pico.sh LLC
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -4,8 +4,9 @@ Pubsub over ssh.
 
 ```bash
 # term 1
+mkdir ./ssh_data
 cat ~/.ssh/id_ed25519 ./ssh_data/authorized_keys
-go run cmd/authorized_keys
+go run ./cmd/authorized_keys
 
 # term 2
 ssh -p 2222 localhost sub xyz

cmd/authorized_keys/main.go

@@ -44,29 +44,29 @@ type PubSub interface {
 	GetSubs() []*Subscriber
 	Sub(l *Subscriber) error
 	UnSub(l *Subscriber) error
-	Pub(msg *Msg) []error
+	Pub(msg *Msg) error
 }
 
-type BasicPubSub struct {
+type PubSubMulticast struct {
 	logger *slog.Logger
 	subs   []*Subscriber
 }
 
-func (b *BasicPubSub) GetSubs() []*Subscriber {
+func (b *PubSubMulticast) GetSubs() []*Subscriber {
 	b.logger.Info("getsubs")
 	return b.subs
 }
 
-func (b *BasicPubSub) Sub(sub *Subscriber) error {
-	b.logger.Info("sub", "channel", sub.Name)
+func (b *PubSubMulticast) Sub(sub *Subscriber) error {
 	id := uuid.New()
 	sub.ID = id.String()
+	b.logger.Info("sub", "channel", sub.Name, "id", id)
 	b.subs = append(b.subs, sub)
 	return sub.Wait()
 }
 
-func (b *BasicPubSub) UnSub(rm *Subscriber) error {
-	b.logger.Info("unsub", "channel", rm.Name)
+func (b *PubSubMulticast) UnSub(rm *Subscriber) error {
+	b.logger.Info("unsub", "channel", rm.Name, "id", rm.ID)
 	next := []*Subscriber{}
 	for _, sub := range b.subs {
 		if sub.ID != rm.ID {
@@ -77,22 +77,31 @@ func (b *BasicPubSub) UnSub(rm *Subscriber) error {
 	return nil
 }
 
-func (b *BasicPubSub) Pub(msg *Msg) []error {
-	b.logger.Info("pub", "channel", msg.Name)
-	errs := []error{}
+func (b *PubSubMulticast) Pub(msg *Msg) error {
+	log := b.logger.With("channel", msg.Name)
+	log.Info("pub")
+
+	matches := []*Subscriber{}
+	writers := []io.Writer{}
 	for _, sub := range b.subs {
-		if sub.Name != msg.Name {
-			continue
+		if sub.Name == msg.Name {
+			matches = append(matches, sub)
+			writers = append(writers, sub.Session)
 		}
+	}
 
-		_, err := io.Copy(sub.Session, msg.Reader)
-		if err != nil {
-			errs = append(errs, err)
-		}
+	log.Info("copying data")
+	writer := io.MultiWriter(writers...)
+	_, err := io.Copy(writer, msg.Reader)
+	if err != nil {
+		log.Error("pub", "err", err)
+	}
+	for _, sub := range matches {
 		sub.Chan <- err
 		b.UnSub(sub)
 	}
-	return errs
+
+	return err
 }
 
 type Cfg struct {
@@ -116,12 +125,12 @@ func PubSubMiddleware(cfg *Cfg) wish.Middleware {
 			if cmd == "help" {
 				wish.Println(sesh, "USAGE: ssh send.pico.sh (sub|pub) {channel}")
 			} else if cmd == "sub" {
-				listener := &Subscriber{
+				sub := &Subscriber{
 					Name:    channel,
 					Session: sesh,
 					Chan:    make(chan error),
 				}
-				err := cfg.PubSub.Sub(listener)
+				err := cfg.PubSub.Sub(sub)
 				if err != nil {
 					wish.Errorln(sesh, err)
 				}
@@ -136,12 +145,8 @@ func PubSubMiddleware(cfg *Cfg) wish.Middleware {
 					Name:   channel,
 					Reader: sesh,
 				}
-				errs := cfg.PubSub.Pub(msg)
-				if errs != nil {
-					for _, err := range errs {
-						wish.Errorln(sesh, err)
-					}
-				}
+				err := cfg.PubSub.Pub(msg)
+				wish.Errorln(sesh, err)
 			}
 
 			next(sesh)
@@ -156,7 +161,7 @@ func main() {
 	keyPath := GetEnv("SSH_AUTHORIZED_KEYS", "./ssh_data/authorized_keys")
 	cfg := &Cfg{
 		Logger: logger,
-		PubSub: &BasicPubSub{logger: logger},
+		PubSub: &PubSubMulticast{logger: logger},
 	}
 
 	s, err := wish.NewServer(