orderstore

Documentation

Overview

Package orderstore contains implementations of stores used to store order updates.

Index

• func DefaultReaderConfig(topic string, kafkaBrokerUrls []string) kafka.ReaderConfig
• func DefaultWriterConfig(topic string, kafkaBrokerUrls []string) kafka.WriterConfig
• type KafkaStore
  • func NewKafkaStore(kafkaReaderConfig kafka.ReaderConfig, kafkaWriterConfig kafka.WriterConfig, ...) (*KafkaStore, error)
  • func (ks *KafkaStore) Close()
  • func (ks *KafkaStore) LoadOrders(ctx context.Context, filter func(order *model.Order) bool) (map[string]*model.Order, error)
  • func (ks *KafkaStore) SubscribeToAllOrders(ctx context.Context, createdAfter time.Time, bufferSize int) (map[string]*model.Order, <-chan *model.Order, error)
  • func (ks *KafkaStore) Write(ctx context.Context, order *model.Order) error

Functions

func DefaultReaderConfig

func DefaultReaderConfig(topic string, kafkaBrokerUrls []string) kafka.ReaderConfig

func DefaultWriterConfig

func DefaultWriterConfig(topic string, kafkaBrokerUrls []string) kafka.WriterConfig

Types

type KafkaStore

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

func NewKafkaStore

func NewKafkaStore(kafkaReaderConfig kafka.ReaderConfig, kafkaWriterConfig kafka.WriterConfig,
	ownerId string) (*KafkaStore, error)

func (*KafkaStore) Close

func (ks *KafkaStore) Close()

func (*KafkaStore) LoadOrders

func (ks *KafkaStore) LoadOrders(ctx context.Context, filter func(order *model.Order) bool) (map[string]*model.Order, error)

func (*KafkaStore) SubscribeToAllOrders

func (ks *KafkaStore) SubscribeToAllOrders(ctx context.Context, createdAfter time.Time, bufferSize int) (map[string]*model.Order, <-chan *model.Order, error)

func (*KafkaStore) Write

func (ks *KafkaStore) Write(ctx context.Context, order *model.Order) error