sse

sup/sse

sup/sse is a high-reliability Server-Sent Events (SSE) client implementation for Go, built on top of the sup actor library. It provides thread-safe, supervised, and observable way to interact with Server-Sent Events (SSE) streams.

Features

• Actor-Based Concurrency: Thread-safe access to hardware. Multiple goroutines can call the actor safely; the actor handles the queue.
• Supervised Lifecycle: Designed to run under a sup.Supervisor. If the connection drops, the actor returns a fatal error, allowing the supervisor to handle reconnection.
• Last-Event-ID: Automatically tracks the last event ID and includes it in the Last-Event-ID header for reconnections.

Quick start

package main

import (
    "context"
    "fmt"
    "time"

		"github.com/webermarci/sup"
    "github.com/webermarci/sup/sse"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	
	handler := func(sse.Event event) {
		fmt.Println(event)
	}
	
	actor := sse.NewActor("actor", "http://localhost:8080", handler,
		sse.WithTimeout(10 * time.Second),
	)

	supervisor := sup.NewSupervisor("root",
		sup.WithActor(actor),
		sup.WithPolicy(sup.Permanent),
		sup.WithRestartDelay(time.Second),
		sup.WithRestartLimit(5, 10 * time.Second),
	)
	
	supervisor.Run(ctx)
	supervisor.Wait()
}

Using it with pubsub

package main

import (
    "context"
    "fmt"
    "time"

    "github.com/webermarci/pubsub"
    "github.com/webermarci/sup"
    "github.com/webermarci/sup/sse"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	pubsub := pubsub.New[string, sse.Event](10)

	handler := func(sse.Event event) {
		pubsub.Publish("sse", event)
	}
  
	actor := sse.NewActor("actor", "http://localhost:8080", handler,
		sse.WithTimeout(10 * time.Second),
	)

	supervisor := sup.NewSupervisor("root",
		sup.WithActor(actor),
		sup.WithPolicy(sup.Permanent),
		sup.WithRestartDelay(time.Second),
		sup.WithRestartLimit(5, 10 * time.Second),
	)
  
	go supervisor.Run(ctx)
  
	events := pubsub.Subscribe(ctx, "sse")
  
	go func() {
		for event := range events {
			fmt.Println(event)
		}
	}()
  
	supervisor.Wait()
	pubsub.Close()
}