stream

Documentation

Index

• Constants
• Variables
• func CChainNotReady(err error) bool
• func ChainNotReady(err error) bool
• func GetTopicName(networkID uint32, chainID string, eventType EventType) string
• func IndexNotReady(err error) bool
• func NewMessage(id string, chainID string, body []byte, timestamp int64, nanosecond int64) services.Consumable
• func NewProducerCChain(sc *servicesctrl.Control, conf cfg.Config) utils.ListenCloser
• func UpdateTxPool(ctxTimeout time.Duration, conns *utils.Connections, persist db.Persist, ...) error
• type EventType
• type IndexType
  • func (t IndexType) String() string
• type IndexedChain
  • func (t IndexedChain) String() string
• type Message
  • func (m *Message) Body() []byte
  • func (m *Message) ChainID() string
  • func (m *Message) ID() string
  • func (m *Message) Nanosecond() int64
  • func (m *Message) Timestamp() int64
• type ProcessorDB
• type ProcessorFactoryChainDB
  • func NewConsumerDBFactory(factory serviceConsumerFactory, eventType EventType) ProcessorFactoryChainDB
• type ProcessorFactoryInstDB
  • func NewConsumerCChainDB() ProcessorFactoryInstDB
• type ProducerCChain
  • func (p *ProducerCChain) Close() error
  • func (p *ProducerCChain) Failure()
  • func (p *ProducerCChain) ID() string
  • func (p *ProducerCChain) Listen() error
  • func (p *ProducerCChain) Success()
• type ProducerChain
  • func NewProducerChain(sc *servicesctrl.Control, conf cfg.Config, chainID string, eventType EventType, ...) (*ProducerChain, error)
  • func (p *ProducerChain) Close() error
  • func (p *ProducerChain) Failure()
  • func (p *ProducerChain) ID() string
  • func (p *ProducerChain) Listen() error
  • func (p *ProducerChain) Success()

Constants

const (
	IndexerTimeout = 3 * time.Minute
	MaxTxRead      = 500
)

Variables

var (
	ErrUnknownVM = errors.New("unknown VM")

	ErrInvalidTopicName    = errors.New("invalid topic name")
	ErrWrongTopicEventType = errors.New("wrong topic event type")
	ErrWrongTopicNetworkID = errors.New("wrong topic networkID")
)

var (

	// ErrNoMessage is no message
	ErrNoMessage = errors.New("no message")
)

Functions

func CChainNotReady

func CChainNotReady(err error) bool

func ChainNotReady

func ChainNotReady(err error) bool

func GetTopicName

func GetTopicName(networkID uint32, chainID string, eventType EventType) string

func IndexNotReady

func IndexNotReady(err error) bool

func NewMessage

func NewMessage(id string,
	chainID string,
	body []byte,
	timestamp int64,
	nanosecond int64,
) services.Consumable

func NewProducerCChain

func NewProducerCChain(sc *servicesctrl.Control, conf cfg.Config) utils.ListenCloser

func UpdateTxPool

func UpdateTxPool(
	ctxTimeout time.Duration,
	conns *utils.Connections,
	persist db.Persist,
	txPool *db.TxPool,
	sc *servicesctrl.Control,
) error

Types

type EventType

type EventType string

const (
	EventTypeConsensus EventType = "consensus"
	EventTypeDecisions EventType = "decisions"
)

type IndexType

type IndexType byte

const (
	IndexTypeTransactions IndexType = iota
	IndexTypeVertices
	IndexTypeBlocks
)

func (IndexType) String

func (t IndexType) String() string

type IndexedChain

type IndexedChain byte

const (
	IndexXChain IndexedChain = iota
	IndexPChain
	IndexCChain
)

func (IndexedChain) String

func (t IndexedChain) String() string

type Message

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

Message is a message on the event stream

func (*Message) Body

func (m *Message) Body() []byte

func (*Message) ChainID

func (m *Message) ChainID() string

func (*Message) ID

func (m *Message) ID() string

func (*Message) Nanosecond

func (m *Message) Nanosecond() int64

func (*Message) Timestamp

func (m *Message) Timestamp() int64

type ProcessorDB

type ProcessorDB interface {
	Process(*utils.Connections, *db.TxPool) error
	Close() error
	ID() string
	Topic() []string
}

type ProcessorFactoryChainDB

type ProcessorFactoryChainDB func(*servicesctrl.Control, cfg.Config, string, string) (ProcessorDB, error)

func NewConsumerDBFactory

func NewConsumerDBFactory(factory serviceConsumerFactory, eventType EventType) ProcessorFactoryChainDB

NewConsumerFactory returns a processorFactory for the given service consumer

type ProcessorFactoryInstDB

type ProcessorFactoryInstDB func(*servicesctrl.Control, cfg.Config) (ProcessorDB, error)

func NewConsumerCChainDB

func NewConsumerCChainDB() ProcessorFactoryInstDB

type ProducerCChain

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

func (*ProducerCChain) Close

func (p *ProducerCChain) Close() error

func (*ProducerCChain) Failure

func (p *ProducerCChain) Failure()

func (*ProducerCChain) ID

func (p *ProducerCChain) ID() string

func (*ProducerCChain) Listen

func (p *ProducerCChain) Listen() error

func (*ProducerCChain) Success

func (p *ProducerCChain) Success()

type ProducerChain

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

func NewProducerChain

func NewProducerChain(sc *servicesctrl.Control, conf cfg.Config, chainID string, eventType EventType, indexerType IndexType, indexerChain IndexedChain) (*ProducerChain, error)

func (*ProducerChain) Close

func (p *ProducerChain) Close() error

func (*ProducerChain) Failure

func (p *ProducerChain) Failure()

func (*ProducerChain) ID

func (p *ProducerChain) ID() string

func (*ProducerChain) Listen

func (p *ProducerChain) Listen() error

func (*ProducerChain) Success

func (p *ProducerChain) Success()