feat: tidying

docker-compose.yaml

@@ -72,4 +72,4 @@ services:
                 ports: 
                         - "${CRAWLER_PORT}:9020"
                         - "127.0.0.1:${CRAWLER_METRICS_PORT}:9080"
-                        - "127.0.0.1:${CRAWLER_SSE_PORT}:9099"
+                        - "${CRAWLER_SSE_PORT}:9099"

pkg/crawler/ethereum.go

@@ -160,6 +160,7 @@ func NewEthereumCrawler(mainCtx *cli.Context, conf config.EthereumCrawlerConfig)
 		default:
 			log.Error("untraceable gossipsub topic", top)
 			continue
+
 		}
 		topic := eth.ComposeTopic(conf.ForkDigest, top)
 		gs.JoinAndSubscribe(topic, msgHandler, conf.PersistMsgs)

pkg/events/events.go

@@ -61,8 +61,13 @@ func (f *Forwarder) Start(ctx context.Context) error {
 	var err error
 
 	f.once.Do(func() {
+		f.startWorkers()
+
 		f.subscribeDownstream(ctx)
 
+		f.server.CreateStream(TopicEthereumAttestation)
+		f.server.CreateStream(TopicTimedEthereumAttestation)
+
 		err = f.startHTTPServer()
 		if err != nil {
 			return
@@ -83,19 +88,17 @@ func (f *Forwarder) startHTTPServer() error {
 	sseMux := http.NewServeMux()
 	sseMux.HandleFunc("/events", f.server.ServeHTTP)
 
-	log.WithField("address", f.ip).WithField("port", f.port).Info("Starting SSE server")
-
-	errCh := make(chan error, 1)
+	log.WithField("address", f.ip).WithField("port", f.port).Info("Starting SSE server!")
 
 	// Start the HTTP server
 	go func() {
 		err := http.ListenAndServe(fmt.Sprintf("%s:%d", f.ip, f.port), sseMux)
 		if err != nil {
-			errCh <- err
+			log.Fatal(err)
 		}
 	}()
 
-	return <-errCh
+	return nil
 }
 
 // subscribeDownstream subscribes to downstream "internal" events