move receiversAllowAllSources logic into processor only

Author: metachris

cmd/collect/main.go

@@ -172,22 +172,21 @@ func runCollector(cCtx *cli.Context) error {
 
 	// Start service components
 	collector := collector.New(collector.CollectorOpts{
-		Log:                      log,
-		UID:                      uid,
-		Location:                 location,
-		OutDir:                   outDir,
-		CheckNodeURI:             checkNodeURI,
-		ClickhouseDSN:            clickhouseDSN,
-		Nodes:                    nodeURIs,
-		BloxrouteAuth:            blxAuth,
-		EdenAuth:                 edenAuth,
-		ChainboundAuth:           chainboundAuth,
-		Receivers:                receivers,
-		ReceiversAllowedSources:  receiversAllowedSources,
-		ReceiversAllowAllSources: len(receiversAllowedSources) == 1 && receiversAllowedSources[0] == "all",
-		APIListenAddr:            apiListenAddr,
-		MetricsListenAddr:        metricsListenAddr,
-		EnablePprof:              enablePprof,
+		Log:                     log,
+		UID:                     uid,
+		Location:                location,
+		OutDir:                  outDir,
+		CheckNodeURI:            checkNodeURI,
+		ClickhouseDSN:           clickhouseDSN,
+		Nodes:                   nodeURIs,
+		BloxrouteAuth:           blxAuth,
+		EdenAuth:                edenAuth,
+		ChainboundAuth:          chainboundAuth,
+		Receivers:               receivers,
+		ReceiversAllowedSources: receiversAllowedSources,
+		APIListenAddr:           apiListenAddr,
+		MetricsListenAddr:       metricsListenAddr,
+		EnablePprof:             enablePprof,
 	})
 	collector.Start()
 

collector/collector.go

@@ -27,9 +27,8 @@ type CollectorOpts struct {
 	EdenAuth       []string
 	ChainboundAuth []string
 
-	Receivers                []string
-	ReceiversAllowedSources  []string
-	ReceiversAllowAllSources bool // if true, allows all sources for receivers
+	Receivers               []string
+	ReceiversAllowedSources []string
 
 	APIListenAddr     string
 	MetricsListenAddr string
@@ -57,16 +56,15 @@ func (c *Collector) Start() {
 
 	// Initialize the transaction processor, which is the brain of the collector
 	c.processor = NewTxProcessor(TxProcessorOpts{
-		Log:                      c.log,
-		UID:                      c.opts.UID,
-		Location:                 c.opts.Location,
-		OutDir:                   c.opts.OutDir,
-		CheckNodeURI:             c.opts.CheckNodeURI,
-		ClickhouseDSN:            c.opts.ClickhouseDSN,
-		HTTPReceivers:            c.opts.Receivers,
-		ReceiversAllowedSources:  c.opts.ReceiversAllowedSources,
-		ReceiversAllowAllSources: c.opts.ReceiversAllowAllSources,
-		APIServer:                apiServer,
+		Log:                     c.log,
+		UID:                     c.opts.UID,
+		Location:                c.opts.Location,
+		OutDir:                  c.opts.OutDir,
+		CheckNodeURI:            c.opts.CheckNodeURI,
+		ClickhouseDSN:           c.opts.ClickhouseDSN,
+		HTTPReceivers:           c.opts.Receivers,
+		ReceiversAllowedSources: c.opts.ReceiversAllowedSources,
+		APIServer:               apiServer,
 	})
 
 	// Start the transaction processor, which kicks off background goroutines

collector/tx_processor.go

@@ -34,16 +34,15 @@ var (
 )
 
 type TxProcessorOpts struct {
-	Log                      *zap.SugaredLogger
-	OutDir                   string // if empty no files will be written
-	UID                      string
-	Location                 string // location of the collector, will be stored in sourcelogs
-	CheckNodeURI             string
-	ClickhouseDSN            string
-	HTTPReceivers            []string
-	ReceiversAllowedSources  []string
-	ReceiversAllowAllSources bool // if true, allows all sources for receivers
-	APIServer                *api.Server
+	Log                     *zap.SugaredLogger
+	OutDir                  string // if empty no files will be written
+	UID                     string
+	Location                string // location of the collector, will be stored in sourcelogs
+	CheckNodeURI            string
+	ClickhouseDSN           string
+	HTTPReceivers           []string
+	ReceiversAllowedSources []string
+	APIServer               *api.Server
 }
 
 type TxProcessor struct {
@@ -69,7 +68,7 @@ type TxProcessor struct {
 
 	receivers                []TxReceiver
 	receiversAllowedSources  []string
-	receiversAllowAllSources bool // if true, all sources are allowed to send transactions to the receivers
+	receiversAllowAllSources bool
 
 	lastHealthCheckCall time.Time
 
@@ -113,7 +112,7 @@ func NewTxProcessor(opts TxProcessorOpts) *TxProcessor {
 
 		receivers:                receivers,
 		receiversAllowedSources:  opts.ReceiversAllowedSources,
-		receiversAllowAllSources: opts.ReceiversAllowAllSources,
+		receiversAllowAllSources: len(opts.ReceiversAllowedSources) == 1 && opts.ReceiversAllowedSources[0] == "all",
 	}
 }