producer

Documentation

Index

• Variables
• func StringToDuration(in string) (time.Duration, error)
• type EventMeshProducer
  • func NewEventMeshProducer(cfg *ProducerGroupConfig) (EventMeshProducer, error)
• type Processor
• type ProducerGroupConfig
• type ProducerManager
  • func NewProducerManager() (ProducerManager, error)
• type ProducerService
  • func NewProducerServiceServer(producerMgr ProducerManager) (*ProducerService, error)
  • func (p *ProducerService) BatchPublish(ctx context.Context, msg *pb.BatchMessage) (*pb.Response, error)
  • func (p *ProducerService) Publish(ctx context.Context, msg *pb.SimpleMessage) (*pb.Response, error)
  • func (p *ProducerService) RequestReply(ctx context.Context, msg *pb.SimpleMessage) (*pb.SimpleMessage, error)
• type SendMessageContext

Variables

var (
	ErrProtocolPluginNotFound = fmt.Errorf("protocol plugin not found")
)

Functions

func StringToDuration

func StringToDuration(in string) (time.Duration, error)

Types

type EventMeshProducer

type EventMeshProducer interface {
	Send(sctx SendMessageContext, callback *connector.SendCallback) error
	Request(sctx SendMessageContext, callback *connector.RequestReplyCallback, timeout time.Duration) error
	Reply(sctx SendMessageContext, callback *connector.SendCallback) error
	Start() error
	Shutdown() error
	Status() consts.ServiceState
	String() string
}

func NewEventMeshProducer

func NewEventMeshProducer(cfg *ProducerGroupConfig) (EventMeshProducer, error)

type Processor

type Processor interface {
	AsyncMessage(ctx context.Context, producerMgr ProducerManager, msg *pb.SimpleMessage) (*pb.Response, error)
	ReplyMessage(ctx context.Context, producerMgr ProducerManager, emiter emitter.EventEmitter, msg *pb.SimpleMessage) error
	RequestReplyMessage(ctx context.Context, producerMgr ProducerManager, msg *pb.SimpleMessage) (*pb.SimpleMessage, error)
	BatchPublish(ctx context.Context, producerMgr ProducerManager, msg *pb.BatchMessage) (*pb.Response, error)
}

type ProducerGroupConfig

type ProducerGroupConfig struct {
	GroupName string `json:"groupName"`
}

type ProducerManager

type ProducerManager interface {
	GetProducer(groupName string) (EventMeshProducer, error)
	CreateProducer(producerGroupConfig *ProducerGroupConfig) (EventMeshProducer, error)
	Start() error
	Shutdown() error
}

func NewProducerManager

func NewProducerManager() (ProducerManager, error)

type ProducerService

type ProducerService struct {
	pb.UnimplementedPublisherServiceServer
	// contains filtered or unexported fields
}

func NewProducerServiceServer

func NewProducerServiceServer(producerMgr ProducerManager) (*ProducerService, error)

func (*ProducerService) BatchPublish

func (p *ProducerService) BatchPublish(ctx context.Context, msg *pb.BatchMessage) (*pb.Response, error)

func (*ProducerService) Publish

func (p *ProducerService) Publish(ctx context.Context, msg *pb.SimpleMessage) (*pb.Response, error)

func (*ProducerService) RequestReply

func (p *ProducerService) RequestReply(ctx context.Context, msg *pb.SimpleMessage) (*pb.SimpleMessage, error)

type SendMessageContext

type SendMessageContext struct {
	Ctx         context.Context
	Event       *cloudv2.Event
	BizSeqNO    string
	ProducerAPI EventMeshProducer
	CreateTime  time.Time
}

SendMessageContext context in produce message