stream

Documentation

Index

• Variables
• type BellatrixHeaderWithSlot
  • func (b BellatrixHeaderWithSlot) Header() structs.ExecutionPayloadHeader
  • func (b BellatrixHeaderWithSlot) Slot() uint64
• type CapellaHeaderWithSlot
  • func (b CapellaHeaderWithSlot) Header() structs.ExecutionPayloadHeader
  • func (b CapellaHeaderWithSlot) Slot() uint64
• type Client
  • func NewClient(ps Pubsub, st State, cfg StreamConfig) *Client
  • func (s *Client) AttachMetrics(m *metrics.Metrics)
  • func (s *Client) BlockCache() <-chan structs.BlockAndTraceExtended
  • func (s *Client) BuilderBid() <-chan structs.BuilderBidExtended
  • func (s *Client) PublishBlockCache(ctx context.Context, block structs.BlockAndTraceExtended) error
  • func (s *Client) PublishBuilderBid(ctx context.Context, bid structs.BuilderBidExtended) error
  • func (s *Client) PublishSlotDelivered(ctx context.Context, slot structs.Slot) error
  • func (s *Client) RunBlockCacheSubscriber(ctx context.Context) error
  • func (s *Client) RunBuilderBidSubscriber(ctx context.Context) error
  • func (s *Client) RunSlotDeliveredSubscriber(ctx context.Context) error
  • func (s *Client) RunSubscriberParallel(ctx context.Context, num uint) error
• type EncoderSSZ
• type ForkVersionFormat
• type HeaderWithSlot
• type JsonItem
  • func (d JsonItem) Data() []byte
  • func (d JsonItem) Meta() Metadata
• type Metadata
• type Pubsub
• type State
• type StreamConfig
• type StreamData
• type StreamMetrics

Variables

var (
	CacheTopic         = "/block/cache"
	BidTopic           = "/block/bid"
	SlotDeliveredTopic = "/slot/delivered"
)

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

Types

type BellatrixHeaderWithSlot

type BellatrixHeaderWithSlot struct {
	ExecutionPayloadHeader bellatrix.ExecutionPayloadHeader
	HeaderSlot             uint64
}

func (BellatrixHeaderWithSlot) Header

func (b BellatrixHeaderWithSlot) Header() structs.ExecutionPayloadHeader

func (BellatrixHeaderWithSlot) Slot

func (b BellatrixHeaderWithSlot) Slot() uint64

type CapellaHeaderWithSlot

type CapellaHeaderWithSlot struct {
	ExecutionPayloadHeader capella.ExecutionPayloadHeader
	HeaderSlot             uint64
}

func (CapellaHeaderWithSlot) Header

func (b CapellaHeaderWithSlot) Header() structs.ExecutionPayloadHeader

func (CapellaHeaderWithSlot) Slot

func (b CapellaHeaderWithSlot) Slot() uint64

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) BlockCache

func (s *Client) BlockCache() <-chan structs.BlockAndTraceExtended

func (*Client) BuilderBid

func (s *Client) BuilderBid() <-chan structs.BuilderBidExtended

func (*Client) PublishBlockCache

func (s *Client) PublishBlockCache(ctx context.Context, block structs.BlockAndTraceExtended) error

func (*Client) PublishBuilderBid

func (s *Client) PublishBuilderBid(ctx context.Context, bid structs.BuilderBidExtended) error

func (*Client) PublishSlotDelivered

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

func (*Client) RunBlockCacheSubscriber

func (s *Client) RunBlockCacheSubscriber(ctx context.Context) error

func (*Client) RunBuilderBidSubscriber

func (s *Client) RunBuilderBidSubscriber(ctx context.Context) error

func (*Client) RunSlotDeliveredSubscriber

func (s *Client) RunSlotDeliveredSubscriber(ctx context.Context) error

func (*Client) RunSubscriberParallel

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

type EncoderSSZ

type EncoderSSZ interface {
	MarshalSSZ() ([]byte, error)
}

type ForkVersionFormat

type ForkVersionFormat uint64

const (
	Unknown ForkVersionFormat = iota
	AltairJson
	BellatrixJson
	CapellaJson
	CapellaSSZ
)

type HeaderWithSlot

type HeaderWithSlot struct {
	ExecutionPayloadHeader structs.ExecutionPayloadHeader
	HeaderSlot             uint64
}

type JsonItem

type JsonItem struct {
	StreamMeta Metadata
	StreamData []byte
}

func (JsonItem) Data

func (d JsonItem) Data() []byte

func (JsonItem) Meta

func (d JsonItem) Meta() Metadata

type Metadata

type Metadata struct {
	Source       string
	ForkEncoding ForkVersionFormat
}

type Pubsub

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

type State

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

type StreamConfig

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

type StreamData

type StreamData interface {
	Data() []byte
	Meta() Metadata
}

type StreamMetrics

type StreamMetrics struct {
	RecvCounter    *prometheus.CounterVec
	Timing         *prometheus.HistogramVec
	PublishSize    *prometheus.HistogramVec
	PublishCounter *prometheus.CounterVec
}