infra

package

v0.0.0-...-9cad258 Latest Latest Go to latest Published: Sep 22, 2022 License: MIT Imports: 8 Imported by: 0

Documentation ¶

const (
	Driver = "postgres"
	DSN    = "postgres://$SVC_DB_USER_NAME:$SVC_DB_PASSWORD@$SVC_DB_HOST:$SVC_DB_PORT/$SVC_DB_DATABASE?sslmode=disable"
)

View Source

const (
	KafkaAddr    = "$SVC_KAFKA_HOST:$SVC_KAFKA_PORT"
	KafkaGroupID = "$SVC_KAFKA_GROUP_ID"
)

This section is empty.

func NewDB() (*sql.DB, func(), error)

func RunHTTPServer(ctx context.Context, addr string, handler http.Handler) error

type KafkaHandler interface {
	HandleMessage(context.Context, []byte) error
}

type KafkaHandlerFunc func(context.Context, []byte) error

func (f KafkaHandlerFunc) HandleMessage(ctx context.Context, msg []byte) error

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

func NewKafkaProcessor() (*KafkaProcessor, error)

func (c *KafkaProcessor) Close()

func (c *KafkaProcessor) RegisterHandler(topic string, handler KafkaHandler)

func (c *KafkaProcessor) RegisterHandlerFunc(topic string, handler func(context.Context, []byte) error)

func (c *KafkaProcessor) Start(ctx context.Context)

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

func NewKafkaSender() (*KafkaSender, error)

func (s *KafkaSender) Close()

func (s *KafkaSender) Send(topic string, msg []byte) error

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

func NewSimpleConsumerGroupHandler(processor KafkaHandler) *SimpleConsumerGroupHandler

func (c SimpleConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error

func (c SimpleConsumerGroupHandler) ConsumeClaim(
	session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim,
) (err error)

func (c SimpleConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error