handler

Documentation

Index

• type KafkaMessage
• type KafkaMessageHandler
  • func NewKafkaMessageHandler(productRepo repository.ProductRepository, priceRepo repository.PriceRepository, ...) *KafkaMessageHandler
  • func (h *KafkaMessageHandler) OnFail(msg *kafka.Message, err error)
  • func (h *KafkaMessageHandler) OnMessage(msg *kafka.Message)
• type TableName

Types

type KafkaMessage

type KafkaMessage struct {
	Payload struct {
		After  json.RawMessage `json:"after"`
		Before json.RawMessage `json:"before"`
		Source struct {
			Version      string `json:"version"`
			Connector    string `json:"connector"`
			Name         string `json:"name"`
			TimeInMillis int64  `json:"ts_ms"`
			Database     string `json:"db"`
			Schema       string `json:"schema"`
			Table        string `json:"table"`
		} `json:"source"`
		Transaction bool `json:"transaction"`
	} `json:"payload"`
}

type KafkaMessageHandler

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

func NewKafkaMessageHandler

func NewKafkaMessageHandler(
	productRepo repository.ProductRepository,
	priceRepo repository.PriceRepository,
	identifiersRepo repository.IdentifierRepository,
	tagsRepo repository.TagRepository,
	descriptionRepo repository.DescriptionRepository,
) *KafkaMessageHandler

func (*KafkaMessageHandler) OnFail

func (h *KafkaMessageHandler) OnFail(msg *kafka.Message, err error)

func (*KafkaMessageHandler) OnMessage

func (h *KafkaMessageHandler) OnMessage(msg *kafka.Message)

type TableName

type TableName string

const (
	ProductsTableName     TableName = "products"
	PricesTableName       TableName = "prices"
	IdentifiersTableName  TableName = "identifiers"
	DescriptionsTableName TableName = "descriptions"
	TagsTableName         TableName = "tags"
)