How to Add Pub/Sub to a Go API

Go makes concurrent consumers easy with goroutines, so the choices in adding Pub/Sub are mostly about the broker around them.

Adding events to a Go API means picking a broker and client, defining message types both sides agree on, running subscribers alongside the HTTP server, retrying failures and making sure an event is not lost when the database write commits and the publish fails. This guide covers Encore, Watermill, NATS and SNS with SQS.

In the Go Developer Survey 2024 H2, 50% of respondents deployed Go programs to AWS, followed by self-owned or company-owned servers at 37% and Google Cloud at 30%.

What you need to decide:

• Whether you need fan-out to several subscribers or a work queue.
• How message types are shared between publisher and subscriber.
• How failures are retried and where poison messages go.
• How you keep the database write and the publish consistent.

Pub/Sub options for Go at a glance

Encore: typed topics and subscriptions in Go

Encore combines topics, subscriptions and the broker behind them into package-level declarations in your Go code. Encore for Go is a backend framework with an infrastructure SDK, and the broker runs locally under encore run and is provisioned in your cloud on deploy.

A topic is generic over its message type:

package orders

import (
	"context"

	"encore.dev/pubsub"
)

// OrderPlaced is published after an order has been stored.
type OrderPlaced struct {
	OrderID int64  `json:"order_id"`
	Email   string `json:"email"`
}

var OrderPlacedTopic = pubsub.NewTopic[*OrderPlaced]("order-placed", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

type PlaceParams struct {
	Email string `json:"email"`
}

type PlaceResponse struct {
	OrderID int64 `json:"order_id"`
}

//encore:api public method=POST path=/orders
func Place(ctx context.Context, p *PlaceParams) (*PlaceResponse, error) {
	orderID := int64(42) // stored in your database in a real app
	if _, err := OrderPlacedTopic.Publish(ctx, &OrderPlaced{OrderID: orderID, Email: p.Email}); err != nil {
		return nil, err
	}
	return &PlaceResponse{OrderID: orderID}, nil
}

A subscription in another service handles each event, retrying with backoff when the handler returns an error:

package email

import (
	"context"
	"time"

	"encore.app/orders"
	"encore.dev/pubsub"
	"encore.dev/rlog"
)

var _ = pubsub.NewSubscription(
	orders.OrderPlacedTopic, "send-confirmation",
	pubsub.SubscriptionConfig[*orders.OrderPlaced]{
		Handler: sendConfirmation,
		RetryPolicy: &pubsub.RetryPolicy{
			MinBackoff: 10 * time.Second,
			MaxRetries: 5,
		},
	},
)

func sendConfirmation(ctx context.Context, e *orders.OrderPlaced) error {
	rlog.Info("sending confirmation", "order_id", e.OrderID, "email", e.Email)
	return nil
}

Tests read back what was published without running a broker:

package orders

import (
	"context"
	"testing"

	"encore.dev/et"
)

func TestPlacePublishes(t *testing.T) {
	if _, err := Place(context.Background(), &PlaceParams{Email: "ada@example.com"}); err != nil {
		t.Fatal(err)
	}
	msgs := et.Topic(OrderPlacedTopic).PublishedMessages()
	if len(msgs) != 1 {
		t.Fatalf("got %d messages", len(msgs))
	}
}

encore run runs the topic and subscription locally, and on deploy they become SNS and SQS in your AWS account or Pub/Sub on GCP. For events that must match a database write, Encore supports a transactional outbox. A coding agent adding a subscriber gets a compile error if it reads a field OrderPlaced does not have, and it can run encore test to check the publish before opening a pull request.

Without Encore

Without Encore you pick and run a broker for each environment, create the topics and subscriptions or queues, start consumer goroutines next to the HTTP server and shut them down cleanly, and put a fake behind an interface for tests. The sections below show the common libraries.

Try Encore →

Watermill

Watermill puts one publisher and subscriber interface over Kafka, NATS, RabbitMQ, SQS, Google Pub/Sub, Postgres and an in-memory channel, with a router for handlers and middleware for retries:

package main

import (
	"context"
	"encoding/json"

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)

type OrderPlaced struct {
	OrderID int64  `json:"order_id"`
	Email   string `json:"email"`
}

func main() {
	ps := gochannel.NewGoChannel(gochannel.Config{}, watermill.NewStdLogger(false, false))
	msgs, err := ps.Subscribe(context.Background(), "order-placed")
	if err != nil {
		panic(err)
	}
	done := make(chan struct{})
	go func() {
		for m := range msgs {
			var e OrderPlaced
			_ = json.Unmarshal(m.Payload, &e)
			m.Ack()
			close(done)
			return
		}
	}()
	payload, _ := json.Marshal(OrderPlaced{OrderID: 42, Email: "ada@example.com"})
	_ = ps.Publish("order-placed", message.NewMessage(watermill.NewUUID(), payload))
	<-done
}

Good to know

Swapping gochannel for a real broker is a constructor change, and the broker itself is still yours to run in each environment.

NATS with nats.go

NATS is a lightweight broker written in Go, and nats.go is its client. Core NATS delivers at most once to connected subscribers, so events that must survive a restart go through JetStream streams and durable consumers with acknowledgements.

Good to know

NATS is quick to run locally and has a small footprint, while in production you operate a cluster or use a hosted service, and JetStream storage needs capacity planning.

SNS and SQS with the AWS SDK for Go v2

On AWS the native fan-out is an SNS topic with one SQS queue per consumer, using SNS to SQS subscriptions, with the AWS SDK for Go v2 publishing to SNS and long-polling SQS in a goroutine.

Good to know

Standard SQS queues deliver at least once and may reorder messages, so handlers should be idempotent, and FIFO queues trade throughput for ordering and deduplication. Each topic, queue, subscription and queue policy is a resource you create per environment.

How to choose

• Encore for typed topics and subscriptions that run locally, are testable without a broker and are created in your AWS or GCP account.
• Watermill if you want one API over a broker you already run.
• NATS for low-latency messaging you are prepared to operate.
• SNS and SQS if you manage AWS resources already and want the native services.

Related guides

• Message queues vs Pub/Sub
• Add a database to a Go API
• Microservices in Go
• Event-driven architecture
• How to set up a queue on AWS with an AI coding agent
• Best Go framework for event-driven systems
• Best Go Web framework for AI coding agents