Redis Streams on S2

import os

import redis

basin = os.environ["S2_BASIN"]
host = f"{basin}.r.s2.dev"

r = redis.Redis(
    host=host,
    port=6380,
    ssl=True,
    password=os.environ["S2_ACCESS_TOKEN"],
    decode_responses=True,
    socket_timeout=30,
)

stream = "events"

for i in range(3):
    print("appended", r.xadd(stream, {"kind": "example", "n": str(i)}))

# Read the full stream, one page at a time.
last_id = None
start = "-"
while True:
    page = r.xrange(stream, min=start, max="+", count=100)
    if not page:
        break
    for entry_id, fields in page:
        print(entry_id, fields)
    last_id = page[-1][0]
    start = f"({last_id}"

# Tail from the last entry read. The block time must be shorter than socket_timeout.
cursor = last_id or "0-0"
while True:
    reply = r.xread({stream: cursor}, count=100, block=10_000)
    for _, entries in reply:
        for entry_id, fields in entries:
            print(entry_id, fields)
            cursor = entry_id

import { createClient } from "redis";

const basin = process.env.S2_BASIN;
const host = `${basin}.r.s2.dev`;

const client = createClient({
  socket: { host, port: 6380, tls: true, servername: host },
  password: process.env.S2_ACCESS_TOKEN,
});
client.on("error", (err) => console.error(err));
await client.connect();

const stream = "events";

for (let i = 0; i < 3; i++) {
  console.log("appended", await client.xAdd(stream, "*", { kind: "example", n: String(i) }));
}

// Read the full stream, one page at a time.
let lastId = null;
let start = "-";
while (true) {
  const page = await client.xRange(stream, start, "+", { COUNT: 100 });
  if (page.length === 0) break;
  for (const { id, message } of page) console.log(id, message);
  lastId = page.at(-1).id;
  start = `(${lastId}`;
}

// Blocking reads hold the connection, so tail on a dedicated one.
const tail = client.duplicate();
tail.on("error", (err) => console.error(err));
await tail.connect();

let cursor = lastId ?? "0-0";
while (true) {
  const reply = await tail.xRead({ key: stream, id: cursor }, { COUNT: 100, BLOCK: 10_000 });
  for (const { messages } of reply ?? []) {
    for (const { id, message } of messages) {
      console.log(id, message);
      cursor = id;
    }
  }
}

import Redis from "ioredis";

const basin = process.env.S2_BASIN;
const host = `${basin}.r.s2.dev`;

const redis = new Redis({
  host,
  port: 6380,
  tls: { servername: host },
  password: process.env.S2_ACCESS_TOKEN,
});

const stream = "events";

for (let i = 0; i < 3; i++) {
  console.log("appended", await redis.xadd(stream, "*", "kind", "example", "n", String(i)));
}

// Read the full stream, one page at a time.
let lastId = null;
let start = "-";
while (true) {
  const page = await redis.xrange(stream, start, "+", "COUNT", 100);
  if (page.length === 0) break;
  for (const [id, fields] of page) console.log(id, fields);
  lastId = page.at(-1)[0];
  start = `(${lastId}`;
}

// Blocking reads hold the connection, so tail on a dedicated one.
const tail = redis.duplicate();

let cursor = lastId ?? "0-0";
while (true) {
  const reply = await tail.xread("COUNT", 100, "BLOCK", 10_000, "STREAMS", stream, cursor);
  for (const [, entries] of reply ?? []) {
    for (const [id, fields] of entries) {
      console.log(id, fields);
      cursor = id;
    }
  }
}

package main

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/redis/go-redis/v9"
)

func main() {
	ctx := context.Background()
	basin := os.Getenv("S2_BASIN")
	host := basin + ".r.s2.dev"

	rdb := redis.NewClient(&redis.Options{
		Addr:      host + ":6380",
		Password:  os.Getenv("S2_ACCESS_TOKEN"),
		TLSConfig: &tls.Config{ServerName: host, MinVersion: tls.VersionTLS13},
	})
	defer rdb.Close()

	stream := "events"

	for i := range 3 {
		id, err := rdb.XAdd(ctx, &redis.XAddArgs{
			Stream: stream,
			Values: map[string]any{"kind": "example", "n": i},
		}).Result()
		if err != nil {
			panic(err)
		}
		fmt.Println("appended", id)
	}

	// Read the full stream, one page at a time.
	lastID := ""
	start := "-"
	for {
		page, err := rdb.XRangeN(ctx, stream, start, "+", 100).Result()
		if err != nil {
			panic(err)
		}
		if len(page) == 0 {
			break
		}
		for _, msg := range page {
			fmt.Println(msg.ID, msg.Values)
		}
		lastID = page[len(page)-1].ID
		start = "(" + lastID
	}

	// Tail from the last entry read. go-redis extends the read timeout by Block.
	cursor := lastID
	if cursor == "" {
		cursor = "0-0"
	}
	for {
		streams, err := rdb.XRead(ctx, &redis.XReadArgs{
			Streams: []string{stream, cursor},
			Count:   100,
			Block:   10 * time.Second,
		}).Result()
		if errors.Is(err, redis.Nil) {
			continue
		}
		if err != nil {
			panic(err)
		}
		for _, s := range streams {
			for _, msg := range s.Messages {
				fmt.Println(msg.ID, msg.Values)
				cursor = msg.ID
			}
		}
	}
}