streaming: generation-qualified stream lifecycle and exact publication

examples/rmap/basics/main.go

@@ -14,7 +14,7 @@ func main() {
 	ctx := context.Background()
 
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 
 	// Make sure Redis is up and running and we can connect to it
 	if err := rdb.Ping(ctx).Err(); err != nil {

examples/rmap/multi-nodes/main.go

@@ -18,7 +18,7 @@ func main() {
 	ctx := context.Background()
 
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 
 	// Make sure Redis is up and running and we can connect to it
 	if err := rdb.Ping(ctx).Err(); err != nil {

examples/streaming/README.md

@@ -10,30 +10,34 @@ To run the examples, follow these steps:
 
 2. Open a terminal or command prompt.
 
+   The examples read `REDIS_ADDR` and `REDIS_PASSWORD`; from the repository
+   root, `source .env` selects the default local Redis configuration.
+
 3. Clone the `goadesign/pulse` repository by running the following command:
-   ```
+   ```bash
    git clone https://github.com/goadesign/pulse.git
    ```
 
 4. Change into the example directory (e.g. `examples/streaming/single-reader`):
-   ```
+   ```bash
    cd pulse/examples/streaming/single-reader
    ```
-   ```
 
-5. Install the required dependencies by running the following command:
-   ```
-   go get github.com/redis/go-redis/v9 goa.design/pulse/rmap
+5. Download the repository's pinned dependencies:
+   ```bash
+   go mod download
    ```
 
 6. Build the Go program by executing the following command:
-   ```
+   ```bash
    go build
    ```
 
 7. Run the program using the following command:
-   ```
+   ```bash
    ./single-reader
    ```
 
 This will execute the program and demonstrate the basic operations on streaming.
+The [`exact-publication`](exact-publication/main.go) example shows
+deadline-owned `AddOnce` retries and read-only snapshots.

examples/streaming/exact-publication/main.go

@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"goa.design/pulse/streaming"
+	"goa.design/pulse/streaming/options"
+)
+
+func main() {
+	ctx := context.Background()
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     os.Getenv("REDIS_ADDR"),
+		Password: os.Getenv("REDIS_PASSWORD"),
+	})
+	if err := rdb.Ping(ctx).Err(); err != nil {
+		panic(err)
+	}
+
+	deadline := time.Now().Add(time.Minute).Truncate(time.Millisecond)
+	stream, err := streaming.NewStream(
+		"exact-publication",
+		rdb,
+		options.WithStreamDeadline(deadline),
+		options.WithStreamMaxLen(1_000),
+	)
+	if err != nil {
+		panic(err)
+	}
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
+
+	first, err := stream.AddOnce(
+		ctx,
+		"facility-42:alarm-7",
+		"alarm-opened",
+		[]byte("high discharge pressure"),
+		options.WithTopic("alarms"),
+	)
+	if err != nil {
+		panic(err)
+	}
+	retry, err := stream.AddOnce(
+		ctx,
+		"facility-42:alarm-7",
+		"alarm-opened",
+		[]byte("high discharge pressure"),
+		options.WithTopic("alarms"),
+	)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("first=%s retry=%s\n", first, retry)
+
+	events, err := stream.Snapshot(ctx)
+	if err != nil {
+		panic(err)
+	}
+	for _, event := range events {
+		fmt.Printf("%s %s: %s\n", event.ID(), event.EventName(), event.Payload())
+	}
+}

examples/streaming/multi-readers/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it

examples/streaming/multi-sinks/main.go

@@ -14,7 +14,7 @@ import (
 // NOTE: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it
@@ -29,7 +29,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Write 2 events to the stream
 	id1, err := stream.Add(ctx, "event 1", []byte("payload 1"))
@@ -54,7 +58,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink1.Close(ctx)
+	defer func() {
+		if err := sink1.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read and acknowlege event
 	ev := <-sink1.Subscribe()
@@ -70,12 +78,16 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer sink2.Close(ctx)
+	defer func() {
+		if err := sink2.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read second event
 	ev = <-sink2.Subscribe()
 	fmt.Printf("sink 2, event: %s, payload: %s\n", ev.EventName, ev.Payload)
-	if sink2.Ack(ctx, ev); err != nil {
+	if err := sink2.Ack(ctx, ev); err != nil {
 		panic(err)
 	}
 }

examples/streaming/multi-streams/main.go

@@ -14,7 +14,7 @@ import (
 // Note: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 
 	// Make sure Redis is up and running and we can connect to it
@@ -29,7 +29,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream1.Destroy(ctx)
+	defer func() {
+		if err := stream1.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Create sink
 	sink, err := stream1.NewSink(ctx, "multistreams-sink",
@@ -40,7 +44,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Subscribe to events
 	c := sink.Subscribe()
@@ -57,7 +65,11 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer stream2.Destroy(ctx)
+	defer func() {
+		if err := stream2.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add stream to sink
 	err = sink.AddStream(ctx, stream2)

examples/streaming/pub-sub/main.go

@@ -14,7 +14,7 @@ import (
 // NOTE: the example below does not handle errors for brevity.
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -27,7 +27,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event to topic "my-topic"
 	id1, err := stream.Add(ctx,
@@ -56,7 +60,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Read both events
 	c := sink.Subscribe()

examples/streaming/single-reader/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -26,7 +26,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event
 	id, err := stream.Add(ctx, "event", []byte("payload"))

examples/streaming/single-sink/main.go

@@ -13,7 +13,7 @@ import (
 
 func main() {
 	// Create Redis client
-	rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", Password: os.Getenv("REDIS_PASSWORD")})
+	rdb := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_ADDR"), Password: os.Getenv("REDIS_PASSWORD")})
 	ctx := context.Background()
 	if err := rdb.Ping(ctx).Err(); err != nil {
 		panic(err)
@@ -26,7 +26,11 @@ func main() {
 	}
 
 	// Don't forget to destroy the stream when done
-	defer stream.Destroy(ctx)
+	defer func() {
+		if err := stream.Destroy(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Add a new event
 	id, err := stream.Add(ctx, "event", []byte("payload"))
@@ -45,7 +49,11 @@ func main() {
 	}
 
 	// Don't forget to close the sink when done
-	defer sink.Close(ctx)
+	defer func() {
+		if err := sink.Close(ctx); err != nil {
+			panic(err)
+		}
+	}()
 
 	// Consume event
 	ev := <-sink.Subscribe()

pool/node_test.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -992,14 +993,15 @@ func TestStaleNodeStreamCleanup(t *testing.T) {
 		rdb      = ptesting.NewRedisClient(t)
 		node1    = newFastCleanupTestNode(t, ctx, rdb, testName)
 		node2    = newFastCleanupTestNode(t, ctx, rdb, testName)
-		numJobs  = 0
+		numJobs  atomic.Int64
 	)
 	defer ptesting.CleanupRedis(t, rdb, false, testName)
 
-	// Configure nodes to send jobs to specific workers
+	// Configure nodes to send jobs to specific workers. The hasher is shared
+	// by both nodes and called from concurrent routing and rebalance
+	// goroutines, so its state must be synchronized.
 	node1.h = &ptesting.Hasher{IndexFunc: func(key string, numBuckets int64) int64 {
-		numJobs++
-		if numJobs > 2 {
+		if numJobs.Add(1) > 2 {
 			return 0 // to avoid panics on cleanup where jobs get requeued
 		}
 		if key == "job1" {