flowprocess

Mar 2, 2023

Changes in this version

type Node

+ SetProcessorSelector func(processorSelector ProcessorSelector)

+ SubmitTask func(task Task) error

+ type ProcessorSelector interface

+ DefineProcessorInTaskChan func() (processorInTaskChanIndexes []int)

+ SelectInTaskChanIndex func(task Task) (inTaskChanIndex int)

type TaskHandler

+ func NewSortTaskHandler(taskStartId uint64, maxBlocking int) TaskHandler

+ type TraceableFlow interface

+ SetOnNewTaskCreated func(onNewTaskCreated func(task TraceableTask, nodeId int))

+ SetOnTaskFinished func(onTaskFinished func(task TraceableTask, nodeId int, err error))

+ SetTaskAt func(taskAt func(task TraceableTask, nodeId int))

Aug 28, 2022

Changes in this version

type Flow

+ SubmitTaskAfterStart func(nodeId int, task Task) error

type Node

+ Cancel func()

May 25, 2022

Changes in this version

type Flow

+ func NewFlow(taskAt func(task TraceableTask, nodeId int), ...) Flow

+ AddNodeTaskHandlers func(processorChanBufferSize int, taskHandlers ...TaskHandler)

+ type NodeKey string

+ const NODE_ID

type TaskHandler

+ OnCompleted func(dispatch func(outTask Task)) (err error)

+ type TaskHandlerAdapter struct

+ func (h TaskHandlerAdapter) OnCompleted(dispatch func(outTask Task)) (err error)

+ type TraceableTask interface

+ Inner func() Task

+ TaskId func() string

+ func ToTraceableTask(taskId string, task Task) TraceableTask

Nov 30, 2021

Changes in this version

+ type DefaultProcessor struct

+ func (p *DefaultProcessor) Proccess(inTasks InTaskChan, outTask OutTaskChan, ctx context.Context) (cancelAllProcess bool)

type Flow

+ func (p *Flow) AddMultiTaskHandlerNodeProcessors(processorChanBufferSize int, taskHandlers ...MultiTaskHandler)

+ func (p *Flow) AddTaskHandlerNodeProcessors(taskHandler TaskHandler, processorNum int, processorChanBufferSize int)

+ func (p *Flow) Result() (resultChan <-chan Task)

+ type MultiTaskHandler interface

+ GetHandleResult func() (result interface{})

+ Handle func(Task) (err error)

type Processor

+ func NewMultiTaskHandlerProcessor(multiTaskHandler MultiTaskHandler) Processor

+ func NewTaskHandlerProcessors(taskHandler TaskHandler, processorNum int) []Processor

+ type TaskHandler interface

+ Handle func(Task) (result interface{}, err error)

Sep 12, 2021

Changes in this version

type Flow

+ func (p *Flow) AddNode(node Node)

+ type Node interface

+ Done func() <-chan struct{}

+ Process func(in TaskChan) (out TaskChan)

+ func NewNode(processors []Processor, processorChanBufferSize int) Node

Sep 10, 2021

Changes in this version

+ type Flow struct

+ func (p *Flow) AddMultiChanNode(pu MultiChanNode)

+ func (p *Flow) AddNodeProcessors(processors []Processor, processorChanBufferSize int)

+ func (p *Flow) AddSingleChanNode(pu SingleChanNode)

+ func (p *Flow) Await()

+ func (p *Flow) Start()

+ type MultiChanNode interface

+ Done func() <-chan struct{}

+ Process func(ins []TaskChan) (outs []TaskChan)

+ func NewMultiChanNode(processors []Processor, processorChanBufferSize int) MultiChanNode

+ type SingleChanNode interface

+ Done func() <-chan struct{}

+ Process func(in TaskChan) (out TaskChan)

+ func NewSingleChanNode(processors []Processor, processorChanBufferSize int) SingleChanNode

Sep 3, 2021

Changes in this version

type FlowProcess

+ func (p *FlowProcess) AddMultiChanFlowUnit(pu MultiChanFlowUnit)

+ func (p *FlowProcess) AddSingleChanFlowUnitProcessors(processors []Processor, processorChanBufferSize int)

+ type InTaskChan <-chan Task

type MultiChanFlowUnit

+ func NewMultiChanFlowUnit(processors []Processor, processorChanBufferSize int) MultiChanFlowUnit

+ type OutTaskChan chan<- Task

type SingleChanFlowUnit

+ func NewSingleChanFlowUnit(processors []Processor, processorChanBufferSize int) SingleChanFlowUnit

Sep 2, 2021

Changes in this version

+ type FlowProcess struct

+ func (p *FlowProcess) Add(pu MultiChanFlowUnit)

+ func (p *FlowProcess) AddSingleChanFlowUnit(pu SingleChanFlowUnit)

+ func (p *FlowProcess) Await()

+ func (p *FlowProcess) Start()

+ type MultiChanFlowUnit interface

+ Done func() <-chan struct{}

+ UnitProcess func(ins []TaskChan) (outs []TaskChan)

+ type Processor interface

+ Proccess func(inTasks TaskChan, outTask TaskChan)

+ type SingleChanFlowUnit interface

+ Done func() <-chan struct{}

+ UnitProcess func(in TaskChan) (out TaskChan)

+ type Task interface

+ type TaskChan chan Task