replace receive slot with event stream

api/client/event/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["eventStream.go"],
+    importpath = "github.com/prysmaticlabs/prysm/v4/api/client/event",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api:go_default_library",
+        "@com_github_pkg_errors//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
+    ],
+)

api/client/event/eventStream.go

@@ -0,0 +1,141 @@
+package event
+
+import (
+	"bufio"
+	"context"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v4/api"
+	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"
+)
+
+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.Info("Starting listening to Beacon API events on topics: " + allTopics)
+	fullUrl := h.host + "/eth/v1/events?topics=" + allTopics
+	req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, fullUrl, nil)
+	if err != nil {
+		eventsChannel <- &Event{
+			EventType: EventError,
+			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: EventError,
+			Data:      []byte(errors.Wrap(err, "Failed to perform HTTP request").Error()),
+		}
+	}
+
+	defer func() {
+		if closeErr := resp.Body.Close(); closeErr != nil {
+			log.WithError(closeErr).Error("Failed to close events response body")
+			close(eventsChannel)
+		}
+	}()
+	// 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()
+			// 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
+			}
+
+			if line[0:7] == "event: " {
+				// Extract event type from the "event" field
+				eventType = line[7:]
+			} else if line[0:6] == "data: " {
+				// Extract data from the "data" field
+				data = line[6:]
+			}
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		eventsChannel <- &Event{
+			EventType: EventError,
+			Data:      []byte(errors.Wrap(err, "Error reading response body").Error()),
+		}
+	}
+}

validator/client/BUILD.bazel

@@ -25,12 +25,14 @@ go_library(
         "//validator:__subpackages__",
     ],
     deps = [
+        "//api/client/event:go_default_library",
         "//api/grpc:go_default_library",
         "//async:go_default_library",
         "//async/event:go_default_library",
         "//beacon-chain/builder:go_default_library",
         "//beacon-chain/core/altair:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
+        "//beacon-chain/rpc/eth/events:go_default_library",
         "//cache/lru:go_default_library",
         "//cmd:go_default_library",
         "//config/features:go_default_library",

validator/client/beacon-api/BUILD.bazel

@@ -15,7 +15,6 @@ go_library(
         "domain_data.go",
         "doppelganger.go",
         "duties.go",
-        "event_handler.go",
         "genesis.go",
         "get_beacon_block.go",
         "index.go",
@@ -40,11 +39,11 @@ go_library(
     visibility = ["//validator:__subpackages__"],
     deps = [
         "//api:go_default_library",
+        "//api/client/event:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
         "//beacon-chain/core/signing:go_default_library",
         "//beacon-chain/rpc/eth/beacon:go_default_library",
         "//beacon-chain/rpc/eth/config:go_default_library",
-        "//beacon-chain/rpc/eth/events:go_default_library",
         "//beacon-chain/rpc/eth/node:go_default_library",
         "//beacon-chain/rpc/eth/shared:go_default_library",
         "//beacon-chain/rpc/eth/validator:go_default_library",
@@ -64,6 +63,7 @@ go_library(
         "@com_github_golang_protobuf//ptypes/empty",
         "@com_github_pkg_errors//:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
+        "@io_opencensus_go//trace:go_default_library",
         "@org_golang_google_grpc//:go_default_library",
         "@org_golang_google_protobuf//types/known/timestamppb:go_default_library",
     ],

validator/client/beacon-api/beacon_api_validator_client.go

@@ -7,28 +7,24 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/golang/protobuf/ptypes/empty"
 	"github.com/pkg/errors"
+	"github.com/prysmaticlabs/prysm/v4/api/client/event"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v4/validator/client/iface"
+	"go.opencensus.io/trace"
 )
 
 type ValidatorClientOpt func(*beaconApiValidatorClient)
 
-func WithEventHandler(h *EventHandler) ValidatorClientOpt {
-	return func(c *beaconApiValidatorClient) {
-		c.eventHandler = h
-	}
-}
-
 type beaconApiValidatorClient struct {
 	genesisProvider         GenesisProvider
 	dutiesProvider          dutiesProvider
 	stateValidatorsProvider StateValidatorsProvider
 	jsonRestHandler         JsonRestHandler
-	eventHandler            *EventHandler
 	beaconBlockConverter    BeaconBlockConverter
 	prysmBeaconChainCLient  iface.PrysmBeaconChainClient
+	isEventStreamRunning    bool
 }
 
 func NewBeaconApiValidatorClient(jsonRestHandler JsonRestHandler, opts ...ValidatorClientOpt) iface.ValidatorClient {
@@ -42,6 +38,7 @@ func NewBeaconApiValidatorClient(jsonRestHandler JsonRestHandler, opts ...Valida
 			nodeClient:      &beaconApiNodeClient{jsonRestHandler: jsonRestHandler},
 			jsonRestHandler: jsonRestHandler,
 		},
+		isEventStreamRunning: false,
 	}
 	for _, o := range opts {
 		o(c)
@@ -114,10 +111,6 @@ func (c *beaconApiValidatorClient) ProposeExit(ctx context.Context, in *ethpb.Si
 	return c.proposeExit(ctx, in)
 }
 
-func (c *beaconApiValidatorClient) StreamSlots(ctx context.Context, in *ethpb.StreamSlotsRequest) (ethpb.BeaconNodeValidator_StreamSlotsClient, error) {
-	return c.streamSlots(ctx, in, time.Second), nil
-}
-
 func (c *beaconApiValidatorClient) StreamBlocksAltair(ctx context.Context, in *ethpb.StreamBlocksRequest) (ethpb.BeaconNodeValidator_StreamBlocksAltairClient, error) {
 	return c.streamBlocks(ctx, in, time.Second), nil
 }
@@ -163,15 +156,22 @@ func (c *beaconApiValidatorClient) WaitForChainStart(ctx context.Context, _ *emp
 	return c.waitForChainStart(ctx)
 }
 
-func (c *beaconApiValidatorClient) StartEventStream(ctx context.Context) error {
-	if c.eventHandler != nil {
-		if err := c.eventHandler.get(ctx, []string{"head"}); err != nil {
-			return errors.Wrapf(err, "could not invoke event handler")
+func (c *beaconApiValidatorClient) StartEventStream(ctx context.Context, topics []string, eventsChannel chan<- *event.Event) {
+	ctx, span := trace.StartSpan(ctx, "validator.jsonClient.StartEventStream")
+	defer span.End()
+	eventStream, err := event.NewEventStream(ctx, c.jsonRestHandler.GetHttpClient(), c.jsonRestHandler.GetHost(), topics)
+	if err != nil {
+		eventsChannel <- &event.Event{
+			EventType: event.EventError,
+			Data:      []byte(errors.Wrap(err, "failed to start event stream").Error()),
 		}
+		return
 	}
-	return nil
+	c.isEventStreamRunning = true
+	eventStream.Subscribe(eventsChannel)
+	c.isEventStreamRunning = false
 }
 
 func (c *beaconApiValidatorClient) EventStreamIsRunning() bool {
-	return c.eventHandler.running
+	return c.isEventStreamRunning
 }