processor

Documentation

Overview

Package processor defines the document processing unit interface

Index

• func NewWorkerProcessorSpec(ctx context.T, workerLimit int, assignedDocType contracts.DocumentType, ...) *workerProcessorSpec
• type EngineProcessor
  • func NewEngineProcessor(ctx context.T, startWorker *workerProcessorSpec, ...) *EngineProcessor
  • func (p *EngineProcessor) Cancel(docState contracts.DocumentState) (errorCode ErrorCode)
  • func (p *EngineProcessor) InitialProcessing(skipDocumentIfExpired bool) (err error)
  • func (p *EngineProcessor) Start() (resChan chan contracts.DocumentResult, err error)
  • func (p *EngineProcessor) Stop()
  • func (p *EngineProcessor) Submit(docState contracts.DocumentState) (errorCode ErrorCode)
• type ErrorCode
• type ExecuterCreator
• type Processor

Functions

func NewWorkerProcessorSpec

func NewWorkerProcessorSpec(ctx context.T, workerLimit int, assignedDocType contracts.DocumentType, bufferLimit int) *workerProcessorSpec

NewWorkerProcessorSpec return new worker processor specification object reference

Types

type EngineProcessor

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

EngineProcessor defines methods to process the incoming document by pushing to the executor using JobPools

func NewEngineProcessor

func NewEngineProcessor(ctx context.T, startWorker *workerProcessorSpec, cancelWorker *workerProcessorSpec) *EngineProcessor

NewEngineProcessor returns the newly initiated EngineProcessor TODO worker pool should be triggered in the Start() function supported document types indicate the domain of the documents the Processor with run upon. There'll be race-conditions if there're multiple Processors in a certain domain.

func (*EngineProcessor) Cancel

func (p *EngineProcessor) Cancel(docState contracts.DocumentState) (errorCode ErrorCode)

Cancel pushes the command to CancelThread which is responsible for submitting to cancelCommandPool

func (*EngineProcessor) InitialProcessing

func (p *EngineProcessor) InitialProcessing(skipDocumentIfExpired bool) (err error)

func (*EngineProcessor) Start

func (p *EngineProcessor) Start() (resChan chan contracts.DocumentResult, err error)

func (*EngineProcessor) Stop

func (p *EngineProcessor) Stop()

Stop set the cancel flags of all the running jobs, which are to be captured by the command worker and shutdown gracefully

func (*EngineProcessor) Submit

func (p *EngineProcessor) Submit(docState contracts.DocumentState) (errorCode ErrorCode)

Submit submits to the pool a document in form of docState object, results will be streamed back from the channel returned by Start()

type ErrorCode

type ErrorCode string

ErrorCode represents processor related error codes

const (

	// CommandBufferFull denotes that the cancel command buffer is full
	CommandBufferFull ErrorCode = "CommandBufferFull"

	// ClosedProcessor denotes that the processor is closed
	ClosedProcessor ErrorCode = "ClosedProcessor"

	// UnsupportedDocType represents unsupported doc type
	UnsupportedDocType ErrorCode = "UnsupportedDocType"

	// DuplicateCommand represents duplicate command in the buffer
	DuplicateCommand ErrorCode = "DuplicateCommand"

	// InvalidDocumentId represents invalid document id
	InvalidDocumentId ErrorCode = "InvalidDocumentId"

	// ConversionFailed represents conversion from pool to processor error code failed
	ConversionFailed ErrorCode = "ConversionFailed"

	// SubmissionPanic represents panic during submission to the pool
	SubmissionPanic ErrorCode = "SubmissionPanic"
)

type ExecuterCreator

type ExecuterCreator func(ctx context.T) executer.Executer

type Processor

type Processor interface {
	//Start activate the Processor and pick up the leftover document in the last run, it returns a channel to caller to gather DocumentResult
	Start() (chan contracts.DocumentResult, error)
	//InitialProcessing processes any initial documents loaded from file directory. This should be run after Start().
	InitialProcessing(skipDocumentIfExpired bool) error
	//Stop the processor, save the current state to resume later
	Stop()
	//Submit to the pool a document in form of docState object, results will be streamed back from the central channel returned by Start()
	Submit(docState contracts.DocumentState) ErrorCode
	//Cancel cancels processing of the given document
	Cancel(docState contracts.DocumentState) ErrorCode
}