Feature: add gas_price websocket channel (#268)

* Feature: add gas_price websocket channel

* Uncomment

cmd/api/bus/dispatcher.go

@@ -89,7 +89,7 @@ func (d *Dispatcher) listen(ctx context.Context) {
  return
  }
  if notification == nil {
- log.Warn().Str("channel", notification.Channel).Msg("nil notification")
+ log.Warn().Msg("nil notification")
  continue
  }
  if err := d.handleNotification(ctx, notification); err != nil {

cmd/api/gas/tracker.go

@@ -29,6 +29,8 @@ var (
  maxBlockSize = coreTypes.MaxDataBytesNoEvidence(1974272, 100)
 )
 
+type ComputeHandler func(ctx context.Context, gasState GasPrice) error
+
 type Tracker struct {
  state storage.IState
  stats storage.IBlockStats
@@ -39,6 +41,8 @@ type Tracker struct {
  gasState GasPrice
  q *queue
  g workerpool.Group
+
+ computeHandler ComputeHandler
 }
 
 func NewTracker(
@@ -64,6 +68,10 @@ func NewTracker(
  }
 }
 
+func (tracker *Tracker) SubscribeOnCompute(handler ComputeHandler) {
+ tracker.computeHandler = handler
+}
+
 func (tracker *Tracker) Start(ctx context.Context) {
  tracker.g.GoCtx(ctx, tracker.listen)
 }
@@ -90,6 +98,14 @@ func (tracker *Tracker) listen(ctx context.Context) {
  if err := tracker.computeMetrics(); err != nil {
  log.Err(err).Msg("compute metrics")
  }
+
+ if tracker.computeHandler != nil {
+ tracker.g.GoCtx(ctx, func(ctx context.Context) {
+ if err := tracker.computeHandler(ctx, tracker.gasState); err != nil {
+ log.Err(err).Msg("error in compute handler of gas tracker")
+ }
+ })
+ }
  }
  }
 }

cmd/api/handler/websocket/client.go

@@ -82,6 +82,8 @@ func (c *Client) ApplyFilters(msg Subscribe) error {
  c.filters.head = true
  case ChannelBlocks:
  c.filters.blocks = true
+ case ChannelGasPrice:
+ c.filters.gasPrice = true
  default:
  return errors.Wrap(ErrUnknownChannel, msg.Channel)
  }
@@ -97,6 +99,8 @@ func (c *Client) DetachFilters(msg Unsubscribe) error {
  c.filters.head = false
  case ChannelBlocks:
  c.filters.blocks = false
+ case ChannelGasPrice:
+ c.filters.gasPrice = false
  default:
  return errors.Wrap(ErrUnknownChannel, msg.Channel)
  }

cmd/api/handler/websocket/filters.go

@@ -37,7 +37,21 @@ func (f HeadFilter) Filter(c client, msg Notification[*responses.State]) bool {
  return fltrs.head
 }
 
+type GasPriceFilter struct{}
+
+func (f GasPriceFilter) Filter(c client, msg Notification[*responses.GasPrice]) bool {
+ if msg.Body == nil {
+ return false
+ }
+ fltrs := c.Filters()
+ if fltrs == nil {
+ return false
+ }
+ return fltrs.gasPrice
+}
+
 type Filters struct {
- head bool
- blocks bool
+ head bool
+ blocks bool
+ gasPrice bool
 }

cmd/api/handler/websocket/manager.go

@@ -12,6 +12,7 @@ import (
  sdkSync "github.com/dipdup-net/indexer-sdk/pkg/sync"
 
  "github.com/celenium-io/celestia-indexer/cmd/api/bus"
+ "github.com/celenium-io/celestia-indexer/cmd/api/gas"
  "github.com/celenium-io/celestia-indexer/cmd/api/handler/responses"
  "github.com/celenium-io/celestia-indexer/internal/storage"
  "github.com/gorilla/websocket"
@@ -25,8 +26,9 @@ type Manager struct {
  clients *sdkSync.Map[uint64, *Client]
  observer *bus.Observer
 
- blocks *Channel[storage.Block, *responses.Block]
- head *Channel[storage.State, *responses.State]
+ blocks *Channel[storage.Block, *responses.Block]
+ head *Channel[storage.State, *responses.State]
+ gasPrice *Channel[gas.GasPrice, *responses.GasPrice]
 
  g workerpool.Group
 }
@@ -56,6 +58,11 @@ func NewManager(observer *bus.Observer) *Manager {
  HeadFilter{},
  )
 
+ manager.gasPrice = NewChannel(
+ gasPriceProcessor,
+ GasPriceFilter{},
+ )
+
  return manager
 }
 
@@ -131,6 +138,8 @@ func (manager *Manager) AddClientToChannel(channel string, client *Client) {
  manager.head.AddClient(client)
  case ChannelBlocks:
  manager.blocks.AddClient(client)
+ case ChannelGasPrice:
+ manager.gasPrice.AddClient(client)
  default:
  log.Error().Str("channel", channel).Msg("unknown channel name")
  }
@@ -142,7 +151,13 @@ func (manager *Manager) RemoveClientFromChannel(channel string, client *Client)
  manager.head.RemoveClient(client.id)
  case ChannelBlocks:
  manager.blocks.RemoveClient(client.id)
+ case ChannelGasPrice:
+ manager.gasPrice.RemoveClient(client.id)
  default:
  log.Error().Str("channel", channel).Msg("unknown channel name")
  }
 }
+
+func (manager *Manager) GasTrackerHandler(_ context.Context, state gas.GasPrice) error {
+ return manager.gasPrice.processMessage(state)
+}