replace receive slot with event stream

api/client/beacon/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "checkpoint.go",
         "client.go",
         "doc.go",
+        "health.go",
     ],
     importpath = "github.com/prysmaticlabs/prysm/v5/api/client/beacon",
     visibility = ["//visibility:public"],
@@ -36,6 +37,7 @@ go_test(
     srcs = [
         "checkpoint_test.go",
         "client_test.go",
+        "health_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

api/client/beacon/health.go

@@ -0,0 +1,47 @@
+package beacon
+
+import (
+	"sync"
+)
+
+type NodeHealth struct {
+	isHealthy  bool
+	healthChan chan bool
+	sync.RWMutex
+}
+
+func NewNodeHealth(initialStatus bool) *NodeHealth {
+	return &NodeHealth{
+		isHealthy:  initialStatus,
+		healthChan: make(chan bool, 1),
+	}
+}
+
+// HealthUpdates provides a read-only channel for health updates.
+func (n *NodeHealth) HealthUpdates() <-chan bool {
+	return n.healthChan
+}
+
+func (n *NodeHealth) IsHealthy() bool {
+	n.RLock()
+	defer n.RUnlock()
+	return n.isHealthy
+}
+
+func (n *NodeHealth) UpdateNodeHealth(newStatus bool) {
+	n.RLock()
+	isStatusChanged := newStatus != n.isHealthy
+	n.RUnlock()
+
+	if isStatusChanged {
+		n.Lock()
+		// Double-check the condition to ensure it hasn't changed since the first check.
+		if newStatus != n.isHealthy {
+			n.isHealthy = newStatus
+			n.Unlock() // It's better to unlock as soon as the protected section is over.
+			n.healthChan <- newStatus
+		} else {
+			n.Unlock()
+		}
+	}
+}

api/client/beacon/health_test.go

@@ -0,0 +1,102 @@
+package beacon
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestNodeHealth_IsHealthy(t *testing.T) {
+	tests := []struct {
+		name      string
+		isHealthy bool
+		want      bool
+	}{
+		{"initially healthy", true, true},
+		{"initially unhealthy", false, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			n := &NodeHealth{
+				isHealthy:  tt.isHealthy,
+				healthChan: make(chan bool, 1),
+			}
+			if got := n.IsHealthy(); got != tt.want {
+				t.Errorf("IsHealthy() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestNodeHealth_UpdateNodeHealth(t *testing.T) {
+	tests := []struct {
+		name       string
+		initial    bool // Initial health status
+		newStatus  bool // Status to update to
+		shouldSend bool // Should a message be sent through the channel
+	}{
+		{"healthy to unhealthy", true, false, true},
+		{"unhealthy to healthy", false, true, true},
+		{"remain healthy", true, true, false},
+		{"remain unhealthy", false, false, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			n := NewNodeHealth(tt.initial)
+			n.isHealthy = tt.initial // Set initial health status
+			n.UpdateNodeHealth(tt.newStatus)
+
+			// Check if health status was updated
+			if n.IsHealthy() != tt.newStatus {
+				t.Errorf("UpdateNodeHealth() failed to update isHealthy from %v to %v", tt.initial, tt.newStatus)
+			}
+
+			select {
+			case status := <-n.HealthUpdates():
+				if !tt.shouldSend {
+					t.Errorf("UpdateNodeHealth() unexpectedly sent status %v to HealthCh", status)
+				} else if status != tt.newStatus {
+					t.Errorf("UpdateNodeHealth() sent wrong status %v, want %v", status, tt.newStatus)
+				}
+			default:
+				if tt.shouldSend {
+					t.Error("UpdateNodeHealth() did not send any status to HealthCh when expected")
+				}
+			}
+		})
+	}
+}
+
+func TestNodeHealth_Concurrency(t *testing.T) {
+	n := NewNodeHealth(true)
+	var wg sync.WaitGroup
+
+	// Number of goroutines to spawn for both reading and writing
+	numGoroutines := 6
+
+	go func() {
+		for range n.HealthUpdates() {
+			// Consume values to avoid blocking on channel send.
+		}
+	}()
+
+	wg.Add(numGoroutines * 2) // for readers and writers
+
+	// Concurrently update health status
+	for i := 0; i < numGoroutines; i++ {
+		go func() {
+			defer wg.Done()
+			n.UpdateNodeHealth(false) // Flip health status
+			n.UpdateNodeHealth(true)  // And flip it back
+		}()
+	}
+
+	// Concurrently read health status
+	for i := 0; i < numGoroutines; i++ {
+		go func() {
+			defer wg.Done()
+			_ = n.IsHealthy() // Just read the value
+		}()
+	}
+
+	wg.Wait() // Wait for all goroutines to finish
+}

api/client/errors.go

@@ -21,6 +21,9 @@ var ErrNotFound = errors.Wrap(ErrNotOK, "recv 404 NotFound response from API")
 // ErrInvalidNodeVersion indicates that the /eth/v1/node/version API response format was not recognized.
 var ErrInvalidNodeVersion = errors.New("invalid node version response")
 
+// ErrConnectionIssue represents a connection problem.
+var ErrConnectionIssue = errors.New("could not connect")
+
 // Non200Err is a function that parses an HTTP response to handle responses that are not 200 with a formatted error.
 func Non200Err(response *http.Response) error {
 	bodyBytes, err := io.ReadAll(response.Body)

api/client/event/BUILD.bazel

@@ -0,0 +1,24 @@
+load("@prysm//tools/go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["event_stream.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v5/api/client/event",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api:go_default_library",
+        "//api/client:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["event_stream_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//testing/require:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)

api/client/event/event_stream.go

@@ -0,0 +1,148 @@
+package event
+
+import (
+	"bufio"
+	"context"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v5/api"
+	"github.com/prysmaticlabs/prysm/v5/api/client"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	EventHead                        = "head"
+	EventBlock                       = "block"
+	EventAttestation                 = "attestation"
+	EventVoluntaryExit               = "voluntary_exit"
+	EventBlsToExecutionChange        = "bls_to_execution_change"
+	EventProposerSlashing            = "proposer_slashing"
+	EventAttesterSlashing            = "attester_slashing"
+	EventFinalizedCheckpoint         = "finalized_checkpoint"
+	EventChainReorg                  = "chain_reorg"
+	EventContributionAndProof        = "contribution_and_proof"
+	EventLightClientFinalityUpdate   = "light_client_finality_update"
+	EventLightClientOptimisticUpdate = "light_client_optimistic_update"
+	EventPayloadAttributes           = "payload_attributes"
+	EventBlobSidecar                 = "blob_sidecar"
+	EventError                       = "error"
+	EventConnectionError             = "connection_error"
+)
+
+var (
+	_ = EventStreamClient(&EventStream{})
+)
+
+var DefaultEventTopics = []string{EventHead}
+
+type EventStreamClient interface {
+	Subscribe(eventsChannel chan<- *Event)
+}
+
+type Event struct {
+	EventType string
+	Data      []byte
+}
+
+// EventStream is responsible for subscribing to the Beacon API events endpoint
+// and dispatching received events to subscribers.
+type EventStream struct {
+	ctx        context.Context
+	httpClient *http.Client
+	host       string
+	topics     []string
+}
+
+func NewEventStream(ctx context.Context, httpClient *http.Client, host string, topics []string) (*EventStream, error) {
+	// Check if the host is a valid URL
+	_, err := url.ParseRequestURI(host)
+	if err != nil {
+		return nil, err
+	}
+	if len(topics) == 0 {
+		return nil, errors.New("no topics provided")
+	}
+
+	return &EventStream{
+		ctx:        ctx,
+		httpClient: httpClient,
+		host:       host,
+		topics:     topics,
+	}, nil
+}
+
+func (h *EventStream) Subscribe(eventsChannel chan<- *Event) {
+	allTopics := strings.Join(h.topics, ",")
+	log.WithField("topics", allTopics).Info("Listening to Beacon API events")
+	fullUrl := h.host + "/eth/v1/events?topics=" + allTopics
+	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, fullUrl, nil)
+	if err != nil {
+		eventsChannel <- &Event{
+			EventType: EventConnectionError,
+			Data:      []byte(errors.Wrap(err, "failed to create HTTP request").Error()),
+		}
+	}
+	req.Header.Set("Accept", api.EventStreamMediaType)
+	req.Header.Set("Connection", api.KeepAlive)
+	resp, err := h.httpClient.Do(req)
+	if err != nil {
+		eventsChannel <- &Event{
+			EventType: EventConnectionError,
+			Data:      []byte(errors.Wrap(err, client.ErrConnectionIssue.Error()).Error()),
+		}
+	}
+
+	defer func() {
+		if closeErr := resp.Body.Close(); closeErr != nil {
+			log.WithError(closeErr).Error("Failed to close events response body")
+		}
+	}()
+	// Create a new scanner to read lines from the response body
+	scanner := bufio.NewScanner(resp.Body)
+
+	var eventType, data string // Variables to store event type and data
+
+	// Iterate over lines of the event stream
+	for scanner.Scan() {
+		select {
+		case <-h.ctx.Done():
+			log.Info("Context canceled, stopping event stream")
+			close(eventsChannel)
+			return
+		default:
+			line := scanner.Text() // TODO: scanner does not handle /r and does not fully adhere to https://html.spec.whatwg.org/multipage/server-sent-events.html#the-eventsource-interface
+			// Handle the event based on your specific format
+			if line == "" {
+				// Empty line indicates the end of an event
+				if eventType != "" && data != "" {
+					// Process the event when both eventType and data are set
+					eventsChannel <- &Event{EventType: eventType, Data: []byte(data)}
+				}
+
+				// Reset eventType and data for the next event
+				eventType, data = "", ""
+				continue
+			}
+			et, ok := strings.CutPrefix(line, "event: ")
+			if ok {
+				// Extract event type from the "event" field
+				eventType = et
+			}
+			d, ok := strings.CutPrefix(line, "data: ")
+			if ok {
+				// Extract data from the "data" field
+				data = d
+			}
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		eventsChannel <- &Event{
+			EventType: EventConnectionError,
+			Data:      []byte(errors.Wrap(err, errors.Wrap(client.ErrConnectionIssue, "scanner failed").Error()).Error()),
+		}
+	}
+}