engine

type AutoSQLTransform struct {
	Name  string
	Table string `aql: "STAGING_TABLE, optional"`

	StagingSQLConnString string `aql: "STAGING_CONNECTION_STRING, optional"`
	Query                string
	ParameterTable       *ParameterTable
	ParameterNames       []string
	// contains filtered or unexported fields
}

type Condition func(msg map[string]interface{}, eof bool) bool

type ConsoleDestination struct {
	Name         string
	FormatAsJSON bool
	Writer       io.Writer
	// contains filtered or unexported fields
}

type ConsoleLogger struct {
	MinLevel LogLevel
	// contains filtered or unexported fields
}

type Coordinator interface {
	RegisterHooks(...interface{}) //arguments should be SourceHook, TransformHook or DestinationHook
	AddSource(name string, alias string, s Source) error
	AddDestination(name string, alias string, d Destination) error
	AddTest(node string, name string, desc string, c Condition) error
	AddTransform(name string, alias string, t Transform) error
	AddConstraint(before, after string) error
	Connect(from string, to string) error
	UseContext(ctx context.Context)
	Compile() error
	Execute() error
	Stop()
}

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

type Destination interface {

	//Ping checks that the destination is available. It is used to verify
	//the destination at runtime.
	Ping() error

	//Open gives the destination a stream to start pulling from and an error stream
	Open(Stream, Logger, Stopper)
}

type DestinationHook func(string, Destination) (Destination, error)

type DevNull struct {
	Name string
}

type Event struct {
	Time    time.Time
	Source  string
	Level   LogLevel
	Message string
}

type ExcelDestination struct {
	Name      string
	Filename  string `aql:"FILE"`
	Overwrite bool   `aql:"OVERWRITE, optional"`
	Template  string `aql:"TEMPLATE, optional"`
	Sheet     string `aql:"SHEET"`
	Range     ExcelRange
	Alias     string
	Transpose bool     `aql:"TRANSPOSE, optional"`
	Cols      []string `aql:"COLUMNS, optional"`
	// contains filtered or unexported fields
}

type ExcelRange struct {
	X1 int
	Y1 int
	X2 ExcelRangePoint
	Y2 ExcelRangePoint
}

type ExcelRangePoint struct {
	N bool //wildcard
	P int
}

type ExcelSource struct {
	Name                 string
	Filename             string `aql:"FILE"`
	Sheet                string `aql:"SHEET"`
	Range                ExcelRange
	RangeIncludesColumns bool
	Dateformat           string
	Cols                 []string `aql:"COLUMNS, optional"`
	// contains filtered or unexported fields
}

type GenericLogger struct {
	MinLevel LogLevel

	Writer io.Writer
	// contains filtered or unexported fields
}

type GraphNode interface {
	Name() string
	Type() string
}

type HTTPSource struct {
	Name string

	URL                  string            `aql:"URL"` //URL of request
	Headers              map[string]string //Headers to add to request, optional
	JSONPath             string            `aql:"JSON_PATH, optional"` //Path to object containing array of rows, optional
	NoColumnNames        bool              //If response has array of primitive types rather than objects with column names, eg. ["bob",2] instead of {"name": "bob", "age": 2}
	ColumnNames          []string          `aql:"COLUMNS, optional"`                     //if NoColumnNames is true, this should be provided
	PaginationLimitName  string            `aql:"PAGINATION_LIMIT_PARAMETER, optional"`  //query parameter for pagination limit (optional)
	PaginationOffsetName string            `aql:"PAGINATION_OFFSET_PARAMETER, optional"` //query parameter for pagination offset (optional)
	PageSize             int               `aql:"PAGE_SIZE, optional"`                   //size of page for pagination
	// contains filtered or unexported fields
}

type LiteralSource struct {
	Name    string
	Content string
	Columns []string
	Format  LiteralSourceFormat
	// contains filtered or unexported fields
}

type LiteralSourceFormat int

type LogLevel int

type Logger interface {
	//  Chan returns a chan that can be used to log events
	Chan() chan<- Event

	//  Error returns the latest error that has been logged. The logger must keep track of this.
	Error() error

	// Wait should block until the logger is done processing messages in its chan. The sender should close the chan before calling this or it will deadlock
	Wait()
}

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

type MandrillDestination struct {
	Name       string
	APIKey     string `aql:"API_KEY"`
	Sender     *MandrillPrincipal
	Recipients []MandrillPrincipal
	SplitByRow bool   `aql:"SPLIT, optional"`
	Template   string `aql:"TEMPLATE"`
	Subject    string `aql:"SUBJECT, optional"`
	// contains filtered or unexported fields
}

type MandrillPrincipal struct {
	Name  string
	Email string
}

type Message struct {
	Source      string
	Destination string
	Data        []interface{}
}

type Middleware func(Stream) Stream

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

type ParameterTable struct {
	sync.Mutex
	// contains filtered or unexported fields
}

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

type Passthrough struct {
	sync.Mutex
	// contains filtered or unexported fields
}

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

type SQLDestination struct {
	Name             string
	Driver           string
	ConnectionString string
	Table            string `aql:"TABLE"`
	Tx               *sql.Tx

	RowsPerBatch int  `aql:"ROWS_PER_BATCH,optional"`
	DropNulls    bool `aql:"DROP_NULLS,optional"`

	TxUseFunc     func() (*sql.Tx, error)
	TxReleaseFunc func()
	Alias         string
	// contains filtered or unexported fields
}

type SQLInserter interface {
	New() SQLInserter

	//Initialize with connection details and database.
	Initialize(l Logger, tableName string, db *sql.DB, cols []string) error

	//Insert a single batch
	InsertBatch(tx *sql.Tx, msgs []Message) error

	//Hook that is called before the transaction manager/etc commits/rollbacks the transaction
	PreCommit() error
}

type SQLSource struct {
	Name             string
	Driver           string
	ConnectionString string
	Query            string
	ParameterTable   *ParameterTable
	Tx               *sql.Tx

	ExecOnly       bool
	ParameterNames []string
	TxReleaseFunc  func()
	TxUseFunc      func() (*sql.Tx, error)
	// contains filtered or unexported fields
}

type Sequenceable interface {
	Sequence(seq []string)
}

type SequenceableTransform interface {
	Transform
	Sequenceable
}

type Sequencer interface {
	Wait(task string)
	Done(task string)
}

type SlackOpts struct {
	Channel    string `aql:"SLACK_CHANNEL, optional"`
	Emoji      string `aql:"SLACK_EMOJI, optional"`
	Username   string `aql:"SLACK_USER, optional"`
	WebhookURL string `aql:"SLACK_WEBHOOK_URL"`
	MinLevel   string `aql:"SLACK_LOG_LEVEL"`
	Script     string `aql:"SLACK_NAME, optional"`
}

type SliceDestination struct {
	Alias string
	sync.Mutex
	// contains filtered or unexported fields
}

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

type Source interface {
	//SetName sets the name (or alias) of the source for outgoing messages
	SetName(name string)

	//Ping attempts to connect to the source without creating a stream.
	//This is used to check that the source is valid at run-time.
	Ping() error

	//Get connects to the source and returns a stream of data.
	Open(Stream, Logger, Stopper)
}

type SourceHook func(string, Source) (Source, error)

type Stopper interface {
	//Stopped checks if the stopper is stopped
	Stopped() bool

	//Stops. This is irreversible.
	Stop()
}

type Stream interface {
	//Columns returns a slice of column names
	Columns() []string

	//SetColumns sets the destination columns. destination can be a wildcard.
	SetColumns(destination string, cols []string) error

	//Chan is the channel for the stream. It will be closed by the sender when the stream is at an end.
	Chan(destination string) chan Message
}

type TransactionManager interface {
	//  Register makes the connection known to the connection manager. It does
	//  NOT begin a new transaction.
	Register(aql.Connection) error

	//  Use will begin a new transaction (if none exists) or re-use the existing
	//  transaction, locking it so that no one may concurrently use it.
	Tx(connection string) (*sql.Tx, error)

	//  Release the transaction so that it may be used by others. It panics if the
	//  connection has not been registered.
	Release(connection string)

	//  Commit commits ALL transactions. It is an error to call Use() or Register()
	//  after Commit().
	Commit() error

	//  Rollback rolls back ALL transactions. It is an error to call Use() or Register()
	//  after Commit().
	Rollback() error
}

type Transform interface {
	//SetName sets the alias of the transform for outgoing messages
	SetName(name string)

	//Open gives the transform a stream to start pulling from
	Open(source Stream, dest Stream, logger Logger, stop Stopper)
}

type TransformHook func(string, Transform) (Transform, error)