wf

Documentation

Overview

Package wf contains functions that implement node processing workflow based on RabbitMQ messages

Index

• Constants
• func SafeProcessBatch(envConfig *env.EnvConfig, logger *l.CapiLogger, ...) (wfmodel.NodeBatchStatusType, proc.BatchStats, error)
• func UpdateNodeStatusFromBatches(logger *l.CapiLogger, pCtx *ctx.MessageProcessingContext) (wfmodel.NodeBatchStatusType, bool, error)
• func UpdateRunStatusFromNodes(logger *l.CapiLogger, pCtx *ctx.MessageProcessingContext) error
• type DaemonCmdType
  • func AmqpFullReconnectCycle(envConfig *env.EnvConfig, logger *l.CapiLogger, osSignalChannel chan os.Signal) DaemonCmdType
  • func ProcessDataBatchMsg(envConfig *env.EnvConfig, logger *l.CapiLogger, msgTs int64, ...) DaemonCmdType
  • func (daemonCmd DaemonCmdType) ToString() string

Constants

const DlxSuffix string = "_dlx"

Functions

func SafeProcessBatch

func SafeProcessBatch(envConfig *env.EnvConfig, logger *l.CapiLogger, pCtx *ctx.MessageProcessingContext, readerNodeRunId int16, lookupNodeRunId int16) (wfmodel.NodeBatchStatusType, proc.BatchStats, error)

func UpdateNodeStatusFromBatches

func UpdateNodeStatusFromBatches(logger *l.CapiLogger, pCtx *ctx.MessageProcessingContext) (wfmodel.NodeBatchStatusType, bool, error)

func UpdateRunStatusFromNodes

func UpdateRunStatusFromNodes(logger *l.CapiLogger, pCtx *ctx.MessageProcessingContext) error

Types

type DaemonCmdType

type DaemonCmdType int8

const (
	DaemonCmdNone                DaemonCmdType = 0 // Should never see this
	DaemonCmdAckSuccess          DaemonCmdType = 2 // Best case
	DaemonCmdRejectAndRetryLater DaemonCmdType = 3 // Node dependencies are not ready, wait with proessing this node
	DaemonCmdReconnectDb         DaemonCmdType = 4 // Db workflow error, try to reconnect
	DaemonCmdQuit                DaemonCmdType = 5 // Shutdown command was received
	DaemonCmdAckWithError        DaemonCmdType = 6 // There was a processing error: either some serious biz logic re-trying will not help, or it was a data table error (we consider it persistent), so ack it
	DaemonCmdReconnectQueue      DaemonCmdType = 7 // Queue error, try to reconnect
)

func AmqpFullReconnectCycle

func AmqpFullReconnectCycle(envConfig *env.EnvConfig, logger *l.CapiLogger, osSignalChannel chan os.Signal) DaemonCmdType

func ProcessDataBatchMsg

func ProcessDataBatchMsg(envConfig *env.EnvConfig, logger *l.CapiLogger, msgTs int64, dataBatchInfo *wfmodel.MessagePayloadDataBatch) DaemonCmdType

func (DaemonCmdType) ToString

func (daemonCmd DaemonCmdType) ToString() string