subscription

Documentation

Index

• func ConvertBatchHeader(head *common.BatchHeader) *types.Header
• func ForwardFromChannels[R any](inputChannels []chan R, onMessage func(R) error, onBackendDisconnect func(), ...)
• func HandleUnsubscribe(connectionSub *rpc.Subscription, onUnsub func())
• func HandleUnsubscribeErrChan(errChan []<-chan error, onUnsub func())
• type Manager
  • func NewManager[T any]() *Manager[T]
  • func (m *Manager[T]) Subscribe(subscriber T) func()
  • func (m *Manager[T]) Subscribers() []T
• type NewHeadsService
  • func NewNewHeadsService(connect func() (chan *common.BatchHeader, <-chan error, error), ...) *NewHeadsService
  • func (nhs *NewHeadsService) HealthStatus(context.Context) host.HealthStatus
  • func (nhs *NewHeadsService) RegisterNotifier(notifier *rpc.Notifier, subscription *rpc.Subscription)
  • func (nhs *NewHeadsService) Start() error
  • func (nhs *NewHeadsService) Stop() error

Functions

func ConvertBatchHeader

func ConvertBatchHeader(head *common.BatchHeader) *types.Header

func ForwardFromChannels

func ForwardFromChannels[R any](inputChannels []chan R, onMessage func(R) error, onBackendDisconnect func(), backendDisconnected *atomic.Bool, stopped *atomic.Bool, timeoutInterval time.Duration, logger gethlog.Logger)

ForwardFromChannels - reads messages from all input channels, and calls the `onMessage` callback. Exits when the "stopped" flag is true or when the connection times out. Must be called as a go routine!

func HandleUnsubscribe

func HandleUnsubscribe(connectionSub *rpc.Subscription, onUnsub func())

HandleUnsubscribe - when the client calls "unsubscribe" or the subscription times out, it calls `onSub` Must be called as a go routine!

func HandleUnsubscribeErrChan

func HandleUnsubscribeErrChan(errChan []<-chan error, onUnsub func())

HandleUnsubscribeErrChan - when the client calls "unsubscribe" or the subscription times out, it calls `onSub` Must be called as a go routine!

Types

type Manager

type Manager[T any] struct {
	// contains filtered or unexported fields
}

Manager is a subscription manager - allows for thread-safe registering/unregistering subscribers of type T

func NewManager

func NewManager[T any]() *Manager[T]

NewManager creates a new subscription manager

func (*Manager[T]) Subscribe

func (m *Manager[T]) Subscribe(subscriber T) func()

Subscribe adds a new subscriber, returns an unsubscribe function

func (*Manager[T]) Subscribers

func (m *Manager[T]) Subscribers() []T

Subscribers returns a list of all subscribers

type NewHeadsService

type NewHeadsService struct {
	// contains filtered or unexported fields
}

NewHeadsService multiplexes new batch header messages from an input channel into multiple subscribers also handles unsubscribe Note: this is a service which must be Started and Stopped

func NewNewHeadsService

func NewNewHeadsService(connect func() (chan *common.BatchHeader, <-chan error, error), convertToEthHeader bool, logger gethlog.Logger, onMessage func(*common.BatchHeader) error) *NewHeadsService

connect - function that returns the input channel

func (*NewHeadsService) HealthStatus

func (nhs *NewHeadsService) HealthStatus(context.Context) host.HealthStatus

func (*NewHeadsService) RegisterNotifier

func (nhs *NewHeadsService) RegisterNotifier(notifier *rpc.Notifier, subscription *rpc.Subscription)

func (*NewHeadsService) Start

func (nhs *NewHeadsService) Start() error

func (*NewHeadsService) Stop

func (nhs *NewHeadsService) Stop() error