broker

Documentation

Overview

Package broker implements a broker server.

Index

• type Handler
  • func NewHandler(s Scheduler) *Handler
  • func (s *Handler) SendDelayMessage(ctx context.Context, req *pb.SendDelayMessageRequest) (*emptypb.Empty, error)
• type Scheduler
  • func New(config *configs.BrokerConfig, delay storage.DelayStorage, ...) (Scheduler, error)

Types

type Handler

type Handler struct {
	pb.UnimplementedSchedulerServiceServer
	// contains filtered or unexported fields
}

Handler implements the broker grpc server.

func NewHandler

func NewHandler(s Scheduler) *Handler

NewHandler create a broker grpc server.

func (*Handler) SendDelayMessage

func (s *Handler) SendDelayMessage(ctx context.Context, req *pb.SendDelayMessageRequest) (*emptypb.Empty, error)

SendDelayMessage send a delay message to the broker.

type Scheduler

type Scheduler interface {
	// Run start the scheduler
	Run()
	// Produce produce a Message to DelayStorage
	Produce(ctx context.Context, msg *types.Message) error
	// NewConsumer consume Messages from the RealTime Connector queue
	NewConsumer(topic, group string, batchSize int, fn type2.HandleMessage) error
	// Close close the scheduler
	Close() error
}

Scheduler interface

func New

func New(config *configs.BrokerConfig, delay storage.DelayStorage,
	realtime connector.RealTimeConnector) (Scheduler, error)

New create a new scheduler