upgrade goware/channel and add util.Alerter

ethmonitor/ethmonitor.go

@@ -14,6 +14,7 @@ import (
 	"github.com/0xsequence/ethkit/go-ethereum"
 	"github.com/0xsequence/ethkit/go-ethereum/common"
 	"github.com/0xsequence/ethkit/go-ethereum/core/types"
+	"github.com/0xsequence/ethkit/util"
 	"github.com/cespare/xxhash/v2"
 	"github.com/goware/breaker"
 	"github.com/goware/cachestore"
@@ -36,6 +37,7 @@ var DefaultOptions = Options{
 	LogTopics:                []common.Hash{}, // all logs
 	DebugLogging:             false,
 	CacheExpiry:              300 * time.Second,
+	Alerter:                  util.NoopAlerter(),
 }
 
 type Options struct {
@@ -84,6 +86,9 @@ type Options struct {
 	// CacheExpiry is how long to keep each record in cache
 	CacheExpiry time.Duration
 
+	// Alerter config via github.com/goware/alerter
+	Alerter util.Alerter
+
 	// DebugLogging toggle
 	DebugLogging bool
 }
@@ -102,6 +107,7 @@ type Monitor struct {
 	options Options
 
 	log      logger.Logger
+	alert    util.Alerter
 	provider ethrpc.RawInterface
 
 	chain           *Chain
@@ -129,6 +135,9 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err
 	if opts.Logger == nil {
 		return nil, fmt.Errorf("ethmonitor: logger is nil")
 	}
+	if opts.Alerter == nil {
+		opts.Alerter = util.NoopAlerter()
+	}
 
 	opts.BlockRetentionLimit += opts.TrailNumBlocksBehindHead
 
@@ -159,6 +168,7 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err
 	return &Monitor{
 		options:      opts,
 		log:          opts.Logger,
+		alert:        opts.Alerter,
 		provider:     provider,
 		chain:        newChain(opts.BlockRetentionLimit, opts.Bootstrap),
 		chainID:      chainID,
@@ -744,7 +754,10 @@ func (m *Monitor) Subscribe() Subscription {
 	defer m.mu.Unlock()
 
 	subscriber := &subscriber{
-		ch:   channel.NewUnboundedChan[Blocks](m.log, 10, 5000),
+		ch: channel.NewUnboundedChan[Blocks](10, 5000, channel.Options{
+			Logger:  m.log,
+			Alerter: m.alert,
+		}),
 		done: make(chan struct{}),
 	}
 

ethreceipts/ethreceipts.go

@@ -15,6 +15,7 @@ import (
 	"github.com/0xsequence/ethkit/go-ethereum"
 	"github.com/0xsequence/ethkit/go-ethereum/common"
 	"github.com/0xsequence/ethkit/go-ethereum/core/types"
+	"github.com/0xsequence/ethkit/util"
 	"github.com/goware/breaker"
 	"github.com/goware/cachestore"
 	"github.com/goware/cachestore/memlru"
@@ -31,6 +32,7 @@ var DefaultOptions = Options{
 	PastReceiptsCacheSize:            5_000,
 	NumBlocksToFinality:              0, // value of <=0 here will select from ethrpc.Networks[chainID].NumBlocksToFinality
 	FilterMaxWaitNumBlocks:           0, // value of 0 here means no limit, and will listen until manually unsubscribed
+	Alerter:                          util.NoopAlerter(),
 }
 
 type Options struct {
@@ -58,11 +60,15 @@ type Options struct {
 
 	// Cache backend ...
 	// CacheBackend cachestore.Backend
+
+	// Alerter config via github.com/goware/alerter
+	Alerter util.Alerter
 }
 
 type ReceiptsListener struct {
 	options  Options
 	log      logger.Logger
+	alert    util.Alerter
 	provider ethrpc.Interface
 	monitor  *ethmonitor.Monitor
 	br       *breaker.Breaker
@@ -102,6 +108,10 @@ func NewReceiptsListener(log logger.Logger, provider ethrpc.Interface, monitor *
 		opts = options[0]
 	}
 
+	if opts.Alerter == nil {
+		opts.Alerter = util.NoopAlerter()
+	}
+
 	if !monitor.Options().WithLogs {
 		return nil, fmt.Errorf("ethreceipts: ReceiptsListener needs a monitor with WithLogs enabled to function")
 	}
@@ -141,6 +151,7 @@ func NewReceiptsListener(log logger.Logger, provider ethrpc.Interface, monitor *
 	return &ReceiptsListener{
 		options:           opts,
 		log:               log,
+		alert:             opts.Alerter,
 		provider:          provider,
 		monitor:           monitor,
 		br:                breaker.New(log, 1*time.Second, 2, 4), // max 4 retries
@@ -183,8 +194,11 @@ func (l *ReceiptsListener) Subscribe(filterQueries ...FilterQuery) Subscription
 
 	subscriber := &subscriber{
 		listener: l,
-		ch:       channel.NewUnboundedChan[Receipt](l.log, 2, 5000),
-		done:     make(chan struct{}),
+		ch: channel.NewUnboundedChan[Receipt](2, 5000, channel.Options{
+			Logger:  l.log,
+			Alerter: l.alert,
+		}),
+		done: make(chan struct{}),
 		finalizer: &finalizer{
 			numBlocksToFinality: big.NewInt(int64(l.options.NumBlocksToFinality)),
 			queue:               []finalTxn{},

go.mod

@@ -1,6 +1,8 @@
 module github.com/0xsequence/ethkit
 
-go 1.20
+go 1.21
+
+toolchain go1.22.0
 
 require (
 	github.com/btcsuite/btcd v0.23.4
@@ -17,7 +19,7 @@ require (
 	github.com/goware/breaker v0.1.2
 	github.com/goware/cachestore v0.8.0
 	github.com/goware/calc v0.2.0
-	github.com/goware/channel v0.2.4
+	github.com/goware/channel v0.4.0
 	github.com/goware/logger v0.3.0
 	github.com/goware/pp v0.0.3
 	github.com/goware/superr v0.0.2

go.sum

@@ -1,6 +1,8 @@
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
+github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
 github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
+github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
 github.com/btcsuite/btcd v0.22.0-beta.0.20220111032746-97732e52810c/go.mod h1:tjmYdS6MLJ5/s0Fj4DbLgSbDHbEqLJrtnHecBFkdz5M=
 github.com/btcsuite/btcd v0.23.0/go.mod h1:0QJIIN1wwIXF/3G/m87gIwGniDMDQqjVn4SZgnFpsYY=
@@ -73,8 +75,8 @@ github.com/goware/cachestore v0.8.0 h1:NWW9nh7eXgDQfaxdhWOdopKRDc6bWH8qm5kv2w6LS
 github.com/goware/cachestore v0.8.0/go.mod h1:ikiO2RmxIt4cVqEBII6yR+V4Z7pH+y8bMQHpd1MvG1Y=
 github.com/goware/calc v0.2.0 h1:3B9qjXYpE0kgS4LhyklbM6X/0cOvZLdUZG7sdAuVCb4=
 github.com/goware/calc v0.2.0/go.mod h1:BSQUbfS6ICW9RvSV9SikDY+t6/HQKI+CUxIpjE3VD28=
-github.com/goware/channel v0.2.4 h1:ifU+wT0INGf0kdQpCnQRLAThOFUz98T0POHXmKsMat4=
-github.com/goware/channel v0.2.4/go.mod h1:R1EdaSW0bQ7A6KvEtD/FZC4ZLrnf/TMnBrzzwXVfT7M=
+github.com/goware/channel v0.4.0 h1:MfJPLJKhZi2D+seuM2MTKQxOzzjP/zIE7yAn8y47UbA=
+github.com/goware/channel v0.4.0/go.mod h1:m3JBDrah57BipyVWu8R1eBanf1YacnZRluIVSv0jF/U=
 github.com/goware/logger v0.3.0 h1:pdgnsqj2rSDXtfdu+UuAFuBuOapxeDYNETY39227LMM=
 github.com/goware/logger v0.3.0/go.mod h1:IC34c5H56R1I4/R/d51aQhzHsjSJqkQyIHyuJxOiu0w=
 github.com/goware/pp v0.0.3 h1:2Yv0IFGOpVjCDayPYzrqskCe9qmGoKBIyu6Uy//LVUU=

util/alerter.go

@@ -0,0 +1,15 @@
+package util
+
+import "context"
+
+type Alerter interface {
+	Alert(ctx context.Context, format string, v ...interface{})
+}
+
+func NoopAlerter() Alerter {
+	return noopAlerter{}
+}
+
+type noopAlerter struct{}
+
+func (noopAlerter) Alert(ctx context.Context, format string, v ...interface{}) {}