consuming

Documentation

Index

• func New(nodeID string, dispatcher Dispatcher, configs []ConsumerConfig) ([]service.Service, error)
• type ConsumerConfig
• type Dispatcher
• type JSONRawOrString
  • func (j JSONRawOrString) MarshalJSON() ([]byte, error)
  • func (j *JSONRawOrString) UnmarshalJSON(data []byte) error
• type KafkaConfig
• type KafkaConsumer
  • func NewKafkaConsumer(name string, nodeID string, dispatcher Dispatcher, config KafkaConfig, ...) (*KafkaConsumer, error)
  • func (c *KafkaConsumer) Run(ctx context.Context) error
• type KafkaJSONEvent
• type PostgresConfig
• type PostgresConsumer
  • func NewPostgresConsumer(name string, dispatcher Dispatcher, config PostgresConfig, ...) (*PostgresConsumer, error)
  • func (c *PostgresConsumer) Run(ctx context.Context) error
• type PostgresEvent

Functions

func New

func New(nodeID string, dispatcher Dispatcher, configs []ConsumerConfig) ([]service.Service, error)

Types

type ConsumerConfig

type ConsumerConfig = configtypes.Consumer

type Dispatcher

type Dispatcher interface {
	Dispatch(ctx context.Context, method string, data []byte) error
	Publish(ctx context.Context, req *apiproto.PublishRequest) error
	Broadcast(ctx context.Context, req *apiproto.BroadcastRequest) error
}

type JSONRawOrString

type JSONRawOrString json.RawMessage

JSONRawOrString can decode payload from bytes and from JSON string. This gives us better interoperability. For example, JSONB field is encoded as JSON string in Debezium PostgreSQL connector.

func (JSONRawOrString) MarshalJSON

func (j JSONRawOrString) MarshalJSON() ([]byte, error)

MarshalJSON returns m as the JSON encoding of m.

func (*JSONRawOrString) UnmarshalJSON

func (j *JSONRawOrString) UnmarshalJSON(data []byte) error

type KafkaConfig

type KafkaConfig = configtypes.KafkaConsumerConfig

type KafkaConsumer

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

func NewKafkaConsumer

func NewKafkaConsumer(
	name string, nodeID string, dispatcher Dispatcher, config KafkaConfig, metrics *commonMetrics,
) (*KafkaConsumer, error)

func (*KafkaConsumer) Run

func (c *KafkaConsumer) Run(ctx context.Context) error

type KafkaJSONEvent

type KafkaJSONEvent struct {
	Method  string          `json:"method"`
	Payload JSONRawOrString `json:"payload"`
}

type PostgresConfig

type PostgresConfig = configtypes.PostgresConsumerConfig

type PostgresConsumer

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

func NewPostgresConsumer

func NewPostgresConsumer(
	name string, dispatcher Dispatcher, config PostgresConfig, metrics *commonMetrics,
) (*PostgresConsumer, error)

func (*PostgresConsumer) Run

func (c *PostgresConsumer) Run(ctx context.Context) error

type PostgresEvent

type PostgresEvent struct {
	ID        int64
	Method    string
	Payload   []byte
	Partition int64
}