consumer

Documentation

Index

• Constants
• Variables
• type Blob
• type BlockData
• type Commitment
• type Consumer
  • func NewConsumer(config ConsumerConfig, logger logging.Logger) *Consumer
  • func (consumer *Consumer) Close() error
  • func (consumer *Consumer) EnableMetrics(registry *prometheus.Registry) error
  • func (consumer *Consumer) GetBlockStream() <-chan BlockData
  • func (consumer *Consumer) ResetChannel(ctx context.Context, conn *rmq.Connection) bool
  • func (consumer *Consumer) Start(ctx context.Context, addr string)
• type ConsumerConfig
• type Consumerer
• type EventListener
  • func MakeConsumerMetrics(registry *prometheus.Registry) (EventListener, error)
• type Namespace
• type PublishPayload
• type QueuesListener
  • func NewQueuesListener(receivedBlocksC chan<- BlockData, eventListener EventListener, ...) *QueuesListener
  • func (l *QueuesListener) Add(ctx context.Context, rollupId uint32, rollupDataC <-chan rmq.Delivery) error
  • func (l *QueuesListener) Remove(rollupId uint32)
• type SelectiveListener
  • func (l *SelectiveListener) OnArrival()
  • func (l *SelectiveListener) OnFormatError()
• type ShareVersion
• type SubmitRequest
• type TransactionId

Constants

const (
	RECONNECT_DELAY = 3 * time.Second
	RECHANNEL_DELAY = 2 * time.Second
	EXCHANGE_NAME   = "rollup_exchange"
)

const ConsumerSubsystem = "consumer"

const OperatorNamespace = "sffl_operator"

Variables

var (
	AlreadyClosedError = errors.New("Consumer connection is already closed")
)

var (
	QueueExistsError = errors.New("Queue already exists")
)

Types

type Blob

type Blob struct {
	Namespace    Namespace
	ShareVersion ShareVersion
	Commitment   Commitment
	Data         []byte
}

type BlockData

type BlockData struct {
	RollupId      uint32
	Commitment    Commitment
	TransactionId TransactionId
	Block         *types.Block
}

type Commitment

type Commitment = [32]byte

type Consumer

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

func NewConsumer

func NewConsumer(config ConsumerConfig, logger logging.Logger) *Consumer

func (*Consumer) Close

func (consumer *Consumer) Close() error

func (*Consumer) EnableMetrics

func (consumer *Consumer) EnableMetrics(registry *prometheus.Registry) error

func (*Consumer) GetBlockStream

func (consumer *Consumer) GetBlockStream() <-chan BlockData

func (*Consumer) ResetChannel

func (consumer *Consumer) ResetChannel(ctx context.Context, conn *rmq.Connection) bool

func (*Consumer) Start

func (consumer *Consumer) Start(ctx context.Context, addr string)

type ConsumerConfig

type ConsumerConfig struct {
	RollupIds []uint32
	Id        string
}

type Consumerer

type Consumerer interface {
	Reconnect(addr string, ctx context.Context)
	ResetChannel(conn *rmq.Connection, ctx context.Context) bool
	Close() error
	GetBlockStream() <-chan BlockData
}

type EventListener

type EventListener interface {
	OnArrival()
	OnFormatError()
}

func MakeConsumerMetrics

func MakeConsumerMetrics(registry *prometheus.Registry) (EventListener, error)

type Namespace

type Namespace struct {
	Version uint8
	Id      uint32
}

type PublishPayload

type PublishPayload struct {
	TransactionId TransactionId
	Data          []byte
}

Type reflection of MQ format

type QueuesListener

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

func NewQueuesListener

func NewQueuesListener(receivedBlocksC chan<- BlockData, eventListener EventListener, logger logging.Logger) *QueuesListener

func (*QueuesListener) Add

func (l *QueuesListener) Add(ctx context.Context, rollupId uint32, rollupDataC <-chan rmq.Delivery) error

func (*QueuesListener) Remove

func (l *QueuesListener) Remove(rollupId uint32)

type SelectiveListener

type SelectiveListener struct {
	OnArrivalCb     func()
	OnFormatErrorCb func()
}

func (*SelectiveListener) OnArrival

func (l *SelectiveListener) OnArrival()

func (*SelectiveListener) OnFormatError

func (l *SelectiveListener) OnFormatError()

type ShareVersion

type ShareVersion = uint32

Type reflections of NEAR DA client submission format

type SubmitRequest

type SubmitRequest struct {
	Blobs []Blob
}

type TransactionId

type TransactionId = [32]byte