components

type Action uint32

type ChannelBridgeConfig struct {
	Log            logger.Logger
	Name           string
	StepWatcher    *stats.StepWatcher
	WaitCounter    ComponentWaiter
	PanicHandlerFn PanicHandlerFunc
}

type ChannelCombinerConfig struct {
	Log            logger.Logger
	Name           string
	Chan1          chan stream.Record
	Chan2          chan stream.Record
	StepWatcher    *s.StepWatcher
	WaitCounter    ComponentWaiter
	PanicHandlerFn PanicHandlerFunc
}

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

type ComponentStep struct {
	Type string            `json:"type" errorTxt:"step type" mandatory:"yes"`
	Data map[string]string `json:"data" errorTxt:"step data" mandatory:"yes"`
}

type ComponentWaiter interface {
	Add()
	Done()
}

type ControlAction struct {
	Action       Action
	ResponseChan chan error // channel to send a response channel on.
}

type CopyFilesToS3Config struct {
	Log               logger.Logger
	Name              string
	InputChan         chan stream.Record // the input channel of rows containing files (with full paths) to copy/move to S3.
	FileNameChanField string             // name of the field in InputChan that contains the files to move.
	BucketName        string             // target bucket
	BucketPrefix      string
	Region            string
	RemoveInputFiles  bool // true to delete the input files after successful copy to s3.
	StepWatcher       *stats.StepWatcher
	WaitCounter       ComponentWaiter
	PanicHandlerFn    PanicHandlerFunc
}

type CqnDownstreamSnowflakeSync struct {
	// Snowflake specific downstream components:
	// For CsvFileWriterConfig
	CsvFileNamePrefix string
	CsvHeaderFields   []string // the slice of key names to be found in InputChan that will be used as the CSV header.
	CsvMaxFileRows    int
	CsvMaxFileBytes   int
	// For CopyFilesToS3Config
	BucketPrefix string
	BucketName   string
	Region       string
	// For SnowflakeLoaderConfig
	StageName string
}

type CqnDownstreamSyncer interface {
	GetDownstreamSyncFunc(cfg *CqnWithArgsConfig) func(ctx context.Context, inputChan chan stream.Record, tgtSqlQuery string) (outputChan chan stream.Record, controlChan chan ControlAction)
}

type CqnDownstreamTableSync struct {
	BatchSize int // commit interval for internal TableSync components
}

type CqnWithArgsConfig struct {
	Log                  logger.Logger
	Name                 string                   // component instance name for better logging
	SrcCqnConnection     shared.OracleCqnExecutor // relloyd/go-oci8 -> CqnConn{} database connection for CQN source query
	SrcDBConnector       shared.Connector         // another connection matching the source CQN connection. Used for standard SQL queries; passed to the cqnHandler used by this component
	TgtDBConnector       shared.Connector         // target database connector
	SrcSqlQuery          string                   // SQL statement to execute CQN // we can force ORDER BY using the MergeDiffJoinKeys which are expected to be the primary keys
	TgtSqlQuery          string                   // SQL statement to fetch target table data for comparison (do we need ORDER BY on this?)
	SrcRowIdKey          string                   // field names holding Oracle row ID values
	TgtRowIdKey          string                   // field names holding Oracle row ID values
	TgtSchema            string                   // provided to perform a TableSync  // TODO: can we do this without need duplicate info?
	TgtTable             string                   // provided to perform a TableSync
	TgtKeyCols           *om.OrderedMap           // primary key fields to sync - ordered map of: key = chan field name; value = target table column name
	TgtOtherCols         *om.OrderedMap           // other fields to sync - ordered map of: key = chan field name; value = target table column name
	MergeDiffJoinKeys    *om.OrderedMap           // primary keys to join src and tgt result sets for comparison (map entries should be of type map[source]=target)
	MergeDiffCompareKeys *om.OrderedMap           // keys upon which to compare old vs new (tgt vs src) rows (map entries should be of type map[source]=target)
	StepWatcher          *s.StepWatcher           // optional ptr to object that can gather step stats
	WaitCounter          ComponentWaiter
	PanicHandlerFn       PanicHandlerFunc
	DownstreamHandler    CqnDownstreamSyncer
}

type CsvFileWriterConfig struct {
	Log                               logger.Logger
	Name                              string
	InputChan                         chan stream.Record // the input channel of rows to write to an output CSV file.
	OutputDir                         string             // set to empty string to use a system generated sub directory in OS temp space.
	FileNamePrefix                    string
	FileNameSuffixAppendCreationStamp bool
	FileNameSuffixDateFormat          string
	FileNameExtension                 string
	UseGzip                           bool
	MaxFileRows                       int
	MaxFileBytes                      int
	HeaderFields                      []string // the slice of key names to be found in InputChan that will be used as the CSV header.
	OutputChanField4FilePath          string   // the field on outputChan that will contain the file name.
	StepWatcher                       *s.StepWatcher
	WaitCounter                       ComponentWaiter
	PanicHandlerFn                    PanicHandlerFunc
}

type DateRangeGeneratorConfig struct {
	Log                         log.Logger
	Name                        string
	InputChan                   chan stream.Record // input channel containing time.Time
	InputChanFieldName4FromDate string             // name of the field on InputChan which contains the FromDate values expected to be of type time.Time.
	InputChanFieldName4ToDate   string             // name of the field on InputChan which contains the ToDate values expected to be of type time.Time. This takes precedence over use of field ToDateRFC3339orNow.
	ToDateRFC3339orNow          string             // either supply "now" or a date in RFC3339 format which includes a time zone offset. If used, "now" will be truncated to the nearest second.
	UseUTC                      bool               // if true then the date generated by supplying "now" will be in UTC; else we expect local times.
	IntervalSizeSeconds         int                // number of seconds to split the duration between FromDate and ToDate into.
	OutputChanFieldName4LowDate string
	OutputChanFieldName4HiDate  string
	PassInputFieldsToOutput     bool
	StepWatcher                 *stats.StepWatcher
	WaitCounter                 ComponentWaiter
	PanicHandlerFn              PanicHandlerFunc
}

type FieldMapperConfig struct {
	Log            logger.Logger
	Name           string
	InputChan      chan stream.Record // input channel containing time.Time
	Steps          []ComponentStep
	StepWatcher    *stats.StepWatcher
	WaitCounter    ComponentWaiter
	PanicHandlerFn PanicHandlerFunc
}

type FilterMetadata string

type FilterRowsConfig struct {
	Log            log.Logger
	Name           string
	InputChan      chan stream.Record // input channel containing time.Time
	FilterType     FilterType         // one of the keys in the filterTypes map.
	FilterMetadata FilterMetadata     // the field found in stream.StreamRecordIface data map to operate on.
	StepWatcher    *stats.StepWatcher
	WaitCounter    ComponentWaiter
	PanicHandlerFn PanicHandlerFunc
}

type FilterType string

type GenerateRowsConfig struct {
	Log                    logger.Logger
	Name                   string
	FieldName4Sequence     string // optional field name to hold 1-based sequence number on the outputChan.
	MapFieldNamesValuesCSV string // optional CSV string of fieldName:fieldValue tokens to use for row generation.
	NumRows                int    // number of rows to generate on outputChan.
	SleepIntervalSeconds   int    // number of seconds to sleep between emitting rows.
	StepWatcher            *stats.StepWatcher
	WaitCounter            ComponentWaiter
	PanicHandlerFn         PanicHandlerFunc
}

type ManifestWriterConfig struct {
	Log                                       logger.Logger
	Name                                      string
	InputChan                                 chan stream.Record // the input channel of rows to write to an output CSV file.
	InputChanField4FilePath                   string
	OutputDir                                 string // set to empty string to use a system generated sub directory in OS temp space.
	ManifestFileNamePrefix                    string
	ManifestFileNameSuffixAppendCreationStamp bool
	ManifestFileNameSuffixDateFormat          string // golang Time format to be appended to Prefix. If not supplied, the default value is constants.TimeFormatYearSeconds.
	ManifestFileNameExtension                 string
	OutputChanField4ManifestDir               string
	OutputChanField4ManifestName              string
	OutputChanField4ManifestFullPath          string
	StepWatcher                               *s.StepWatcher
	WaitCounter                               ComponentWaiter
	PanicHandlerFn                            PanicHandlerFunc
}

type MergeDiffConfig struct {
	Log                 logger.Logger
	Name                string
	ChanOld             chan stream.Record
	ChanNew             chan stream.Record
	JoinKeys            *om.OrderedMap
	CompareKeys         *om.OrderedMap
	ResultFlagKeyName   string
	OutputIdenticalRows bool
	StepWatcher         *s.StepWatcher
	WaitCounter         ComponentWaiter
	PanicHandlerFn      PanicHandlerFunc
}

type MergeNChannelsConfig struct {
	Log                 logger.Logger
	Name                string
	InputChannels       []chan stream.Record
	AllowFieldOverwrite bool
	StepWatcher         *stats.StepWatcher
	WaitCounter         ComponentWaiter
	PanicHandlerFn      PanicHandlerFunc
}

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

type NumberRangeGeneratorConfig struct {
	Log                         log.Logger
	Name                        string
	InputChan                   chan stream.Record // input channel containing low and high numbers to split by IntervalSize
	InputChanFieldName4LowNum   string             // name of the field on InputChan which contains the Low value ,expected to be of type int
	InputChanFieldName4HighNum  string             // name of the field on InputChan which contains the Hi value, expected to be of type int
	IntervalSize                float64            // number of units to split the difference between LowNum and HighNum into
	OutputLeftPaddedNumZeros    int
	OutputChanFieldName4LowNum  string
	OutputChanFieldName4HighNum string
	PassInputFieldsToOutput     bool
	StepWatcher                 *stats.StepWatcher
	WaitCounter                 ComponentWaiter
	PanicHandlerFn              PanicHandlerFunc
}

type PanicHandlerFunc func()

type S3BucketListerConfig struct {
	Log                               logger.Logger
	Name                              string
	Region                            string // AWS region for the bucket.
	BucketName                        string // AWS bucket name.
	BucketPrefix                      string // AWS bucket prefix.
	ObjectNamePrefix                  string // list files where the beginning of their names matches this string (this is not the AWS bucket prefix). This is given to S3 list command and a dumb filter.
	ObjectNameRegexp                  string // used to further filter the list of files fetched using the ObjectNamePrefix.
	OutputField4FileName              string // the map key on outputChan that contains the file names found in the S3 bucket. If this is an empty string then default to value found in this package var, Defaults.
	OutputField4FileNameWithoutPrefix string
	OutputField4BucketName            string             // the map key on outputChan that contains the bucket name. If this is an empty string then default to value found in this package var, Defaults.
	OutputField4BucketPrefix          string             // the map key on outputChan that contains the bucket prefix. If this is an empty string then default to value found in this package var, Defaults.
	OutputField4BucketRegion          string             // the map key on outputChan that contains the bucket region. If this is an empty string then default to value found in this package var, Defaults.
	StepWatcher                       *stats.StepWatcher // supply a StepWatcher or nil.
	WaitCounter                       ComponentWaiter
	PanicHandlerFn                    PanicHandlerFunc
}

type S3ManifestReaderConfig struct {
	Log                          logger.Logger
	Name                         string
	InputChan                    chan stream.Record // the input channel of rows to write to an output CSV file.
	InputChanField4ManifestName  string             // path to manifest files (s3:// or file://)
	BucketName                   string             // bucket containing manifest files
	BucketPrefix                 string
	Region                       string
	OutputChanField4DataFileName string // outputChan field to produce file names onto
	StepWatcher                  *stats.StepWatcher
	WaitCounter                  ComponentWaiter
	PanicHandlerFn               PanicHandlerFunc
}

type SnowflakeLoaderConfig struct {
	Log                     logger.Logger
	Name                    string
	InputChan               chan stream.Record
	Db                      shared.Connector        // connection to target snowflake database abstracted via interface.
	InputChanField4FileName string                  // the field name found on InputChan that contains the file name to load.
	StageName               string                  // the external stage that can access the files to load.
	TargetSchemaTableName   rdbms.SchemaTable       // the [schema.]table to load into.
	DeleteAll               bool                    // set to true to SQL DELETE all table rows before loading begins (set Use1Transaction = true for a safe reload of data).
	FnGetSnowflakeSqlSlice  SnowflakeSqlBuilderFunc // func that will be used by NewSnowflakeLoader to fetch a slice of SQL statements to execute per input row.
	CommitSequenceKeyName   string                  // the field name added by this component to the outputChan record, incremented when a batch is committed; used by downstream components - see also TableSync component.
	StepWatcher             *stats.StepWatcher
	WaitCounter             ComponentWaiter
	PanicHandlerFn          PanicHandlerFunc
}

type SnowflakeMergeConfig struct {
	Log                     logger.Logger
	Name                    string
	InputChan               chan stream.Record
	Db                      shared.Connector  // connection to target snowflake database abstracted via interface.
	InputChanField4FileName string            // the field name found on InputChan that contains the file name to load.
	StageName               string            // the external stage that can access the files to load.
	TargetSchemaTableName   rdbms.SchemaTable // the [schema.]table to load into.
	CommitSequenceKeyName   string            // the field name added by this component to the outputChan record, incremented when a batch is committed; used by downstream components - see also TableSync component.
	TargetKeyCols           *om.OrderedMap    // ordered map of: key = chan field name; value = target table column name
	TargetOtherCols         *om.OrderedMap    // ordered map of: key = chan field name; value = target table column name
	StepWatcher             *stats.StepWatcher
	WaitCounter             ComponentWaiter
	PanicHandlerFn          PanicHandlerFunc
}

type SnowflakeMergeSqlConfig struct {
	TargetKeyCols   *om.OrderedMap // ordered map of: key = chan field name; value = target table column name
	TargetOtherCols *om.OrderedMap // ordered map of: key = chan field name; value = target table column name
}

type SnowflakeSqlBuilderFunc func(tableName rdbms.SchemaTable, stageName string, fileName string, force bool) []string

type SnowflakeSyncConfig struct {
	Log                     logger.Logger
	Name                    string
	InputChan               chan stream.Record
	Db                      shared.Connector  // connection to target snowflake database abstracted via interface.
	InputChanField4FileName string            // the field name found on InputChan that contains the file name to load.
	StageName               string            // the external stage that can access the files to load.
	TargetSchemaTableName   rdbms.SchemaTable // the [schema.]table to load into.
	CommitSequenceKeyName   string            // the field name added by this component to the outputChan record, incremented when a batch is committed; used by downstream components - see also TableSync component.
	TargetKeyCols           *om.OrderedMap    // ordered map of: key = chan field name; value = target table column name
	TargetOtherCols         *om.OrderedMap    // ordered map of: key = chan field name; value = target table column name
	FlagField               string
	StepWatcher             *stats.StepWatcher
	WaitCounter             ComponentWaiter
	PanicHandlerFn          PanicHandlerFunc
}

type SnowflakeSyncSqlConfig struct {
	TargetKeyCols   *om.OrderedMap // ordered map of: key = chan field name; value = target table column name
	TargetOtherCols *om.OrderedMap // ordered map of: key = chan field name; value = target table column name
	FlagField       string
}

type SqlExecConfig struct {
	Log                      logger.Logger
	Name                     string
	InputChan                chan stream.Record
	SqlQueryFieldName        string
	SqlRowsAffectedFieldName string
	OutputDb                 shared.Connector
	StepWatcher              *s.StepWatcher
	WaitCounter              ComponentWaiter
	PanicHandlerFn           PanicHandlerFunc
}

type SqlQueryWithArgsConfig struct {
	Log            logger.Logger
	Name           string
	Db             shared.Connector
	StepWatcher    *s.StepWatcher // optional ptr to object that can gather step stats.
	WaitCounter    ComponentWaiter
	Sqltext        string
	Args           []interface{}
	PanicHandlerFn PanicHandlerFunc
}

type SqlQueryWithChanConfig struct {
	Log             logger.Logger
	Name            string
	Db              shared.Connector
	StepWatcher     *s.StepWatcher // optional ptr to object that can gather step stats.
	WaitCounter     ComponentWaiter
	Sqltext         string
	InputChan       chan stream.Record // optional input channel from which values for bind variables are fetched. If omitted, then Sqltext must not use binds.
	InputChanFields []string           // list of field names in the input channel for which to use as bind variables.
	PanicHandlerFn  PanicHandlerFunc
}

type SqlQueryWithReplace struct {
	Log            logger.Logger
	Name           string
	Db             shared.Connector
	StepWatcher    *s.StepWatcher // optional ptr to object that can gather step stats.
	WaitCounter    ComponentWaiter
	Sqltext        string
	Args           []interface{}
	Replacements   map[string]string
	PanicHandlerFn PanicHandlerFunc
}

type StdOutPassThroughConfig struct {
	Log             logger.Logger
	Name            string
	InputChan       chan stream.Record
	Writer          io.Writer // the write to output records to, usually STDOUT
	OutputFields    []string  // the list of fields to write to the Writer (leave empty for all fields)
	AbortAfterCount int64
	StepWatcher     *stats.StepWatcher
	WaitCounter     ComponentWaiter
	PanicHandlerFn  PanicHandlerFunc
}

type TableMergeConfig struct {
	Log                                logger.Logger // TODO: do we need to find a way to stub this out?
	Name                               string
	InputChan                          chan stream.Record // input rows to write to database table.
	OutputDb                           shared.Connector   // target database connection for writes.
	ExecBatchSize                      int                // commit interval in num rows
	CommitBatchSize                    int
	shared.SqlStatementGeneratorConfig // config for target database table
	StepWatcher                        *s.StepWatcher
	WaitCounter                        ComponentWaiter
	PanicHandlerFn                     PanicHandlerFunc
}

type TableSyncConfig struct {
	Log             logger.Logger
	Name            string
	InputChan       chan stream.Record // input rows to write to database table.
	OutputDb        shared.Connector   // target database connection for writes.
	CommitBatchSize int                // commit interval in num rows
	TxtBatchNumRows int                // number of rows in a single SQL statement.
	// outputRowsAfterCommit bool                 // FEATURE NOT USED YET - this component will forward rows to its outputChan as they are processed (False) or after each transaction is committed (True). The latter means that extra memory is used to buffer rows the amount of which matches the batch size before they are released downstream.
	FlagKeyName                        string // name of the key in channel inputChan that contains values "N", "C", "D" (see constants for actual values) that can be used to distinguish, "new", "changed" and "deleted" rows, which resolve to database INSERTs/UPDATEs/DELETEs respectively.
	CommitSequenceKeyName              string // the field name added by this component to the outputChan record, incremented when a batch is committed.
	shared.SqlStatementGeneratorConfig        // config for target database table
	StepWatcher                        *s.StepWatcher
	WaitCounter                        ComponentWaiter
	PanicHandlerFn                     PanicHandlerFunc
}