stream

Documentation

Index

• Variables
• type BellatrixStreamBlock
  • func (b BellatrixStreamBlock) BlockBidAndTrace() structs.BlockBidAndTrace
  • func (b BellatrixStreamBlock) CompleteBlock() (structs.CompleteBlockstruct, error)
  • func (b BellatrixStreamBlock) IsCache() bool
  • func (b BellatrixStreamBlock) Loggable() map[string]any
  • func (b BellatrixStreamBlock) Source() string
• type CapellaStreamBlock
  • func (b CapellaStreamBlock) BlockBidAndTrace() structs.BlockBidAndTrace
  • func (b CapellaStreamBlock) CompleteBlock() (structs.CompleteBlockstruct, error)
  • func (b CapellaStreamBlock) IsCache() bool
  • func (b CapellaStreamBlock) Loggable() map[string]any
  • func (b CapellaStreamBlock) Source() string
• type Client
  • func NewClient(ps Pubsub, st State, cfg StreamConfig) *Client
  • func (s *Client) AttachMetrics(m *metrics.Metrics)
  • func (s *Client) PublishBlockSubmission(ctx context.Context, block structs.BlockBidAndTrace) error
  • func (s *Client) PublishCacheBlock(ctx context.Context, block structs.BlockBidAndTrace) error
  • func (s *Client) PublishSlotDelivered(ctx context.Context, slot structs.Slot) error
  • func (s *Client) RunBlockSubscriber(ctx context.Context, ds Datastore, blocks chan []byte) error
  • func (s *Client) RunSlotDeliveredSubscriber(ctx context.Context, slots chan []byte) error
  • func (s *Client) RunSubscriberParallel(ctx context.Context, ds Datastore, num uint) error
  • func (s *Client) SlotDeliveredChan() <-chan structs.Slot
• type Datastore
• type ForkVersionFormat
• type GenericStreamBlock
• type Pubsub
• type SlotDelivered
• type State
• type StreamBlock
• type StreamConfig
• type StreamMetrics

Variables

var (
	ErrDecodeVarint = errors.New("error decoding varint value")
)

Types

type BellatrixStreamBlock

type BellatrixStreamBlock struct {
	Block        bellatrix.BlockBidAndTrace
	IsBlockCache bool
	StreamSource string
}

func (BellatrixStreamBlock) BlockBidAndTrace

func (b BellatrixStreamBlock) BlockBidAndTrace() structs.BlockBidAndTrace

func (BellatrixStreamBlock) CompleteBlock

func (b BellatrixStreamBlock) CompleteBlock() (structs.CompleteBlockstruct, error)

func (BellatrixStreamBlock) IsCache

func (b BellatrixStreamBlock) IsCache() bool

func (BellatrixStreamBlock) Loggable

func (b BellatrixStreamBlock) Loggable() map[string]any

func (BellatrixStreamBlock) Source

func (b BellatrixStreamBlock) Source() string

type CapellaStreamBlock

type CapellaStreamBlock struct {
	Block        capella.BlockBidAndTrace
	IsBlockCache bool
	StreamSource string
}

func (CapellaStreamBlock) BlockBidAndTrace

func (b CapellaStreamBlock) BlockBidAndTrace() structs.BlockBidAndTrace

func (CapellaStreamBlock) CompleteBlock

func (b CapellaStreamBlock) CompleteBlock() (structs.CompleteBlockstruct, error)

func (CapellaStreamBlock) IsCache

func (b CapellaStreamBlock) IsCache() bool

func (CapellaStreamBlock) Loggable

func (b CapellaStreamBlock) Loggable() map[string]any

func (CapellaStreamBlock) Source

func (b CapellaStreamBlock) Source() string

type Client

type Client struct {
	Pubsub Pubsub

	Config StreamConfig
	Logger log.Logger
	// contains filtered or unexported fields
}

func NewClient

func NewClient(ps Pubsub, st State, cfg StreamConfig) *Client

func (*Client) AttachMetrics

func (s *Client) AttachMetrics(m *metrics.Metrics)

func (*Client) PublishBlockSubmission

func (s *Client) PublishBlockSubmission(ctx context.Context, block structs.BlockBidAndTrace) error

func (*Client) PublishCacheBlock

func (s *Client) PublishCacheBlock(ctx context.Context, block structs.BlockBidAndTrace) error

func (*Client) PublishSlotDelivered

func (s *Client) PublishSlotDelivered(ctx context.Context, slot structs.Slot) error

func (*Client) RunBlockSubscriber

func (s *Client) RunBlockSubscriber(ctx context.Context, ds Datastore, blocks chan []byte) error

func (*Client) RunSlotDeliveredSubscriber

func (s *Client) RunSlotDeliveredSubscriber(ctx context.Context, slots chan []byte) error

func (*Client) RunSubscriberParallel

func (s *Client) RunSubscriberParallel(ctx context.Context, ds Datastore, num uint) error

func (*Client) SlotDeliveredChan

func (s *Client) SlotDeliveredChan() <-chan structs.Slot

type Datastore

type Datastore interface {
	PutPayload(context.Context, structs.PayloadKey, structs.BlockBidAndTrace, time.Duration) error
	CacheBlock(context.Context, structs.PayloadKey, *structs.CompleteBlockstruct) error
}

type ForkVersionFormat

type ForkVersionFormat uint64

const (
	Unknown ForkVersionFormat = iota
	AltairJson
	BellatrixJson
	CapellaJson
)

type GenericStreamBlock

type GenericStreamBlock struct {
	Block        structs.BlockBidAndTrace
	IsBlockCache bool
	StreamSource string
}

type Pubsub

type Pubsub interface {
	Publish(context.Context, string, []byte) error
	Subscribe(context.Context, string) chan []byte
}

type SlotDelivered

type SlotDelivered struct {
	Slot uint64
}

type State

type State interface {
	ForkVersion(epoch structs.Slot) structs.ForkVersion
	HeadSlot() structs.Slot
}

type StreamBlock

type StreamBlock interface {
	BlockBidAndTrace() structs.BlockBidAndTrace
	CompleteBlock() (structs.CompleteBlockstruct, error)
	IsCache() bool
	Source() string

	Loggable() map[string]any
}

type StreamConfig

type StreamConfig struct {
	Logger          log.Logger
	ID              string
	TTL             time.Duration
	PubsubTopic     string // pubsub topic name for block submissions
	StreamQueueSize int
}

type StreamMetrics

type StreamMetrics struct {
	StreamRecvCounter *prometheus.CounterVec
	Timing            *prometheus.HistogramVec
}