transports

Aug 25, 2023

Changes in this version

type KafkaProducer

+ func (kp *KafkaProducer) PurgeReplayCache()

type Producer

+ PurgeReplayCache func()

Apr 3, 2023

Changes in this version

type KafkaConsumer

+ func (kc *KafkaConsumer) ProducerCount(d time.Duration) uint

type KafkaProducer

+ func (kp *KafkaProducer) ProducerCount(d time.Duration) uint

+ func (kp *KafkaProducer) SetHealth(b bool)

type Producer

+ SetHealth func(bool)

+ type ProducerCounter interface

+ ProducerCount func(time.Duration) uint

May 20, 2022

Changes in this version

type Producer

+ func NewWebsocketProducer(wsurl string, resumer StreamsResumption) (Producer, error)

+ func ResolveMuxProducer(brokerParams []ProducerBrokerParams, resumer StreamsResumption) (Producer, error)

+ func ResolveProducerWithResumer(brokerURL, defaultTopic string, schema map[string]string, ...) (Producer, error)

+ type ProducerBrokerParams struct

+ DefaultTopic string

+ Schema map[string]string

+ URL string

+ type StreamsResumption interface

+ BlocksFrom func(ctx context.Context, block uint64, hash types.Hash) (chan *delivery.PendingBatch, error)

+ GetBlock func(ctx context.Context, block uint64) *delivery.PendingBatch

+ type TransportBatch struct

+ Batches map[string]types.Hash

+ Deletes []string

+ Hash types.Hash

+ Number hexutil.Uint64

+ ParentHash types.Hash

+ Values map[string]hexutil.Bytes

+ Weight *hexutil.Big

+ func (tb *TransportBatch) ToPendingBatch() *delivery.PendingBatch

Jan 6, 2022

Changes in this version

+ func ResumptionForTimestamp(brokerURL string, topics []string, timestamp int64) ([]byte, error)

Dec 8, 2021

Changes in this version

+ type BrokerParams struct

+ DefaultTopic string

+ Rollback int64

+ Topics []string

+ URL string

type Consumer

+ func ResolveConsumer(brokerURL, defaultTopic string, topics []string, resumption []byte, ...) (Consumer, error)

+ func ResolveMuxConsumer(brokerParams []BrokerParams, resumption []byte, lastNumber int64, ...) (Consumer, error)

type Producer

+ func ResolveProducer(brokerURL, defaultTopic string, schema map[string]string) (Producer, error)

Sep 29, 2021

Changes in this version

+ var SHA256 scram.HashGeneratorFcn = func() hash.Hash

+ var SHA512 scram.HashGeneratorFcn = func() hash.Hash

+ func CreateTopicIfDoesNotExist(brokerAddr, topic string, numPartitions int32, ...) error

+ func ParseKafkaURL(brokerURL string) ([]string, *sarama.Config)

+ type Consumer interface

+ Close func()

+ Ready func() <-chan struct{}

+ Start func() error

+ Subscribe func(ch interface{}) types.Subscription

+ SubscribeReorg func(ch chan<- map[int64]types.Hash) types.Subscription

+ WhyNotReady func(types.Hash) string

+ func NewKafkaConsumer(brokerURL, defaultTopic string, topics []string, resumption []byte, ...) (Consumer, error)

+ func NewNullConsumer() Consumer

+ type KafkaConsumer struct

+ func (kc *KafkaConsumer) Close()

+ func (kc *KafkaConsumer) Ready() <-chan struct{}

+ func (kc *KafkaConsumer) Start() error

+ func (kc *KafkaConsumer) Subscribe(ch interface{}) types.Subscription

+ func (kc *KafkaConsumer) SubscribeReorg(ch chan<- map[int64]types.Hash) types.Subscription

+ func (kc *KafkaConsumer) WhyNotReady(hash types.Hash) string

+ type KafkaProducer struct

+ func (kp *KafkaProducer) AddBlock(number int64, hash, parentHash types.Hash, weight *big.Int, ...) error

+ func (kp *KafkaProducer) LatestBlockFromFeed() (int64, error)

+ func (kp *KafkaProducer) Reorg(number int64, hash types.Hash) (func(), error)

+ func (kp *KafkaProducer) SendBatch(batchid types.Hash, delete []string, update map[string][]byte) error

+ type KafkaResumptionMessage struct

+ func (m *KafkaResumptionMessage) Key() []byte

+ func (m *KafkaResumptionMessage) Offset() int64

+ func (m *KafkaResumptionMessage) Source() string

+ func (m *KafkaResumptionMessage) Time() time.Time

+ func (m *KafkaResumptionMessage) Value() []byte

+ type Producer interface

+ AddBlock func(number int64, hash, parentHash types.Hash, weight *big.Int, ...) error

+ LatestBlockFromFeed func() (int64, error)

+ Reorg func(number int64, hash types.Hash) (func(), error)

+ SendBatch func(batchid types.Hash, delete []string, update map[string][]byte) error

+ func NewKafkaProducer(brokerURL, defaultTopic string, schema map[string]string) (Producer, error)

+ type XDGSCRAMClient struct

+ func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

+ func (x *XDGSCRAMClient) Done() bool

+ func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)