Quickstart | SQLStreams Docs

This quickstart will walk you through how to get everything set up to run a basic producer and consumer using SQLStreams.

If you just want the end result, clone this.

0. Prerequisites

Go 1.27+
Docker Compose for running Postgres (15 to 18 supported)

1. Prepare the workspace

Create a directory and install the library:

mkdir sqlstreams-quickstart
cd sqlstreams-quickstart
go mod init example.com/sqlstreams-quickstart
go get github.com/allegedlyreliable/sqlstreams@v0.1.6
mkdir -p cmd/produce cmd/consume

Next, let’s create our Docker Compose file for running Postgres.

touch compose.yaml

# compose.yaml
name: quickstart

services:
  db:
    image: postgres:18-alpine
    environment:
      POSTGRES_USER: username
      POSTGRES_PASSWORD: password
      POSTGRES_DB: quickstart
    volumes:
      - pgdata:/var/lib/postgresql
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U username -d quickstart"]
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  pgdata:

Spin it up

docker compose up -d --wait

2. Produce a message

Save this as cmd/produce/main.go:

// cmd/produce/main.go
package main

import (
  "fmt"
  "os"

  sqlstreams "github.com/allegedlyreliable/sqlstreams/client"
)

type WelcomeEmailV1 struct {
  UserId string `json:"user_id"`
}

func (WelcomeEmailV1) SchemaVersion() int { return 1 } // increment on breaking changes

func main() {
  if err := run(); err != nil {
    fmt.Fprintln(os.Stderr, err)
    os.Exit(1)
  }
}

func run() error {
  ctx, stop := sqlstreams.LifecycleContext(nil)
  defer stop()

  pool, err := sqlstreams.NewPostgresPool(ctx, "username", "password", "localhost", "quickstart", nil)
  if err != nil {
    return err
  }
  defer pool.Close()

  client, err := sqlstreams.NewClient(ctx, pool, nil)
  if err != nil {
    return err
  }

  emails := client.Stream[WelcomeEmailV1]("signup.welcome-email")
  if _, err := emails.Register(ctx, nil); err != nil {
    return err
  }

  producer, err := emails.Producer().Register(ctx, nil)
  if err != nil {
    return err
  }

  produced, err := producer.Produce(ctx, &WelcomeEmailV1{UserId: "user-123"}, nil)
  if err != nil {
    return err
  }
  fmt.Printf("produced message id=%d\n", produced.Id)
  return nil
}

Run it:

go run ./cmd/produce

The output should look something like this:

produced message id=1

You can run this as many times as you’d like and the message id will increment with each run.

3. Consume the message

Save this as cmd/consume/main.go:

// cmd/consume/main.go
package main

import (
  "context"
  "errors"
  "fmt"
  "os"

  sqlstreams "github.com/allegedlyreliable/sqlstreams/client"
)

type WelcomeEmailV1 struct {
  UserId string `json:"user_id"`
}

func (WelcomeEmailV1) SchemaVersion() int { return 1 } // increment on breaking changes

func main() {
  if err := run(); err != nil {
    fmt.Fprintln(os.Stderr, err)
    os.Exit(1)
  }
}

func run() error {
  ctx, stop := sqlstreams.LifecycleContext(nil)
  defer stop()

  pool, err := sqlstreams.NewPostgresPool(ctx, "username", "password", "localhost", "quickstart", nil)
  if err != nil {
    return err
  }
  defer pool.Close()

  client, err := sqlstreams.NewClient(ctx, pool, nil)
  if err != nil {
    return err
  }

  emails := client.Stream[WelcomeEmailV1]("signup.welcome-email")
  registered, err := emails.Get(ctx)
  if err != nil {
    return err
  }
  if registered == nil {
    return errors.New("stream not registered -- run the producer first")
  }

  sender := emails.Consumer("email-sender")
  consumer, err := sender.Register(ctx, nil)
  if err != nil {
    return err
  }

  return consumer.Consume(ctx, receiveEmail, nil)
}

func receiveEmail(ctx context.Context, email *WelcomeEmailV1) error {
  fmt.Printf("received welcome email request for %s\n", email.UserId)
  return nil
}

Run it:

go run ./cmd/consume

The output should look something like this:

received welcome email request for user-123

And that’s it. You can have a separate terminal session run the producer and watch the consumer consume the produced messages in real time.

4. Continue with your application

Produce messages in a database transaction saves an application change and its message together.
Consumer groups explains shared work and independent progress.
Retries and failed messages explains what happens when processing fails.
Register a consumer group lists group configuration fields and defaults.
Consume messages lists session options and consumer handler results.