processor

Documentation

Index

• Constants
• func NewService(queue *nats.EncodedConn, repository Repository) (processingpb.ProcessingServer, error)
• type Client
  • func (c Client) Do(request interface{}, endpoint string) (string, error)
• type Command
  • func (c Command) Execute() (*processingpb.Response, error)
  • func (c Command) ProcessingRequest() *processingpb.ProcessingRequest
  • func (c Command) ShouldExecute() (bool, error)
• type Executable
• type Job
  • func (j Job) Execute() (*processingpb.Response, error)
  • func (j Job) ProcessingRequest() *processingpb.ProcessingRequest
  • func (j Job) ShouldExecute() (bool, error)
• type Processor
  • func New(logger zerolog.Logger, queue *nats.EncodedConn, ...) Processor
  • func (p Processor) Start() error
  • func (p Processor) Stop()
• type Repository
  • func NewRepository(url string) (Repository, error)

Constants

const (
	ProcessorQueueGroup           = "processor"
	MessageRequestReceivedSubject = "messagerequest.received"
	JobReceivedSubject            = "jobs"
	ReplyReceivedSubject          = "processor.reply.received"
)

const (
	DatabaseName = "treediagram_processor"
)

Functions

func NewService

func NewService(queue *nats.EncodedConn, repository Repository) (processingpb.ProcessingServer, error)

Types

type Client

type Client struct{}

func (Client) Do

func (c Client) Do(request interface{}, endpoint string) (string, error)

type Command

type Command struct {
	Request processingpb.MessageRequest `json:"request"`
	Intent  intentpb.Intent             `json:"intent"`
}

func (Command) Execute

func (c Command) Execute() (*processingpb.Response, error)

func (Command) ProcessingRequest

func (c Command) ProcessingRequest() *processingpb.ProcessingRequest

func (Command) ShouldExecute

func (c Command) ShouldExecute() (bool, error)

type Executable

type Executable interface {
	ShouldExecute() (bool, error)
	Execute() (*processingpb.Response, error)
	ProcessingRequest() *processingpb.ProcessingRequest
}

type Job

type Job struct {
	SchedulingJob schedulingpb.Job
}

func (Job) Execute

func (j Job) Execute() (*processingpb.Response, error)

func (Job) ProcessingRequest

func (j Job) ProcessingRequest() *processingpb.ProcessingRequest

func (Job) ShouldExecute

func (j Job) ShouldExecute() (bool, error)

type Processor

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

func New

func New(logger zerolog.Logger, queue *nats.EncodedConn, registry intentpb.IntentRegistryClient, userClient userpb.UserClient, repository Repository) Processor

func (Processor) Start

func (p Processor) Start() error

func (Processor) Stop

func (p Processor) Stop()

type Repository

type Repository interface {
	SaveProcessingRequest(*processingpb.ProcessingRequest) error
	ProcessingRequest(string) (*processingpb.ProcessingRequest, error)
	SaveProcessingEvent(*processingpb.ProcessingEvent) error
	ProcessingEvents(string) ([]*processingpb.ProcessingEvent, error)
	SaveMessageReply(*processingpb.MessageReply) error
	MessageReply(string) (*processingpb.MessageReply, error)

	Migrate() error
}

func NewRepository

func NewRepository(url string) (Repository, error)