_
 _ __  ___ _   _| |__
| '_ \/ __| | | | '_ \
| |_) \__ \ |_| | |_) |
| .__/|___/\__,_|_.__/
|_|

A tiny, concurrent pub/sub broker in Go. Fan-out over gRPC streams, zero dependencies to run.

---

Demo

# terminal 1: broker
$ go run ./cmd/broker
broker listening on :50051

# terminal 2: subscriber
$ go run ./cmd/consumer news
subscribing to topic: news
[news] hello psub

# terminal 3: publisher
$ go run ./cmd/publisher news "hello psub"
published

Start more consumers on the same topic and every one of them gets every message.

How it works

flowchart LR
    P1([Publisher]) -- "Publish (unary)" --> B
    P2([Publisher]) -- "Publish (unary)" --> B

    subgraph B [Broker]
        direction TB
        R["topic → subscriber channels<br/><sub>guarded by RWMutex</sub>"]
    end

    B -- "stream" --> C1([Consumer A])
    B -- "stream" --> C2([Consumer B])
    B -- "stream" --> C3([Consumer C])

Loading

sequenceDiagram
    participant C as Consumer
    participant B as Broker
    participant P as Publisher

    C->>B: Subscribe("news")
    Note over B: register buffered channel under topic
    P->>B: Publish("news", payload)
    Note over B: RLock, snapshot channels, RUnlock
    B-->>P: success
    B-->>C: stream.Send(message)
    C--xB: disconnect
    Note over B: stream context cancelled, channel removed

Loading

1. Subscribe opens a long-lived gRPC server-stream. The broker gives each stream its own buffered channel and registers it under the topic.
2. Publish is a plain unary call. The broker snapshots the topic's channels under a read lock, releases it, then pushes the message into each channel without blocking.
3. Disconnect cancels the stream's context. The broker notices, and a defer removes the subscriber. No heartbeats, no leaks.

Design

Choice	Reason
sync.RWMutex	Publishes read the registry far more often than subscribes write to it, so readers shouldn't block each other
Snapshot, then send outside the lock	Sending can be slow; the lock is never held across it
Non-blocking send (select + default)	One slow subscriber can never stall a publisher or the other subscribers
Buffered channel per subscriber	Absorbs bursts without dropping
Server-streaming RPC	gRPC has no broadcast primitive, so a stream per subscriber is the idiomatic way to push
Context-based cleanup	Disconnects are detected by gRPC itself

Delivery model: at-most-once and in-memory, in the spirit of NATS core. Subscribers receive messages published while they are connected, and a single publisher's messages arrive in order.

Project layout

psub/
├── cmd/
│   ├── broker/        # gRPC server on :50051
│   ├── publisher/     # CLI: publish a message
│   └── consumer/      # CLI: subscribe and print
└── pkg/
    ├── broker/        # registry, Publish, Subscribe
    ├── publisher/     # importable publisher client
    ├── consumer/      # importable consumer client
    └── grpcapi/       # .proto + generated code

API

service PubSubService {
  rpc Publish(PublishRequest)     returns (PublishResponse);  // unary
  rpc Subscribe(SubscribeRequest) returns (stream Message);   // server-streaming
}

Quick start

git clone https://github.com/SupriyoP09/psub.git
cd psub
go mod tidy
go run ./cmd/broker

Generated protobuf code is committed, so protoc is not needed to build or run.

Use it as a library

// publish
p, _ := publisher.New("localhost:50051")
defer p.Close()
p.Publish(ctx, "news", []byte("hello"))

// subscribe
c, _ := consumer.New("localhost:50051")
defer c.Close()
c.Subscribe(ctx, "news", func(m *pb.Message) {
    fmt.Printf("[%s] %s\n", m.Topic, m.Payload)
})

Regenerate protobuf code

Only needed after editing psub.proto:

cd pkg/grpcapi
protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       psub.proto