core

type Bgpipe struct {
	zerolog.Logger
	Version string

	Ctx    context.Context
	Cancel context.CancelCauseFunc

	F      *pflag.FlagSet // global flags
	K      *koanf.Koanf   // global config
	Pipe   *pipe.Pipe     // bgpfix pipe
	Stages []*StageBase   // pipe stages
	// contains filtered or unexported fields
}

type NewStage func(base *StageBase) Stage

type Stage interface {
	// Attach is run before the pipe starts.
	// It should check the config and attach to the bgpfix pipe.
	Attach() error

	// Prepare is called when the stage starts, but before Run, callbacks, and handlers.
	// It should prepare required I/O, eg. files, network connections, etc.
	//
	// If no error is returned, the stage emits a "READY" event, all callbacks and handlers
	// are enabled, and Run is called when all stages starting in parallel are ready too.
	Prepare() error

	// Run runs the stage and returns after all work has finished.
	// It must respect StageBase.Ctx. Returning a non-nil error different
	// than ErrStopped results in a fatal error that stops the whole pipe.
	//
	// Emits "START" just before, and "STOP" after stage operation is finished.
	Run() error

	// Stop is called when the stage is requested to stop (by an event),
	// or after Run() exits (in order to clean-up).
	// It should safely finish all I/O and make Run return if it's still running.
	Stop() error
}

type StageBase struct {
	zerolog.Logger // logger with stage name
	Stage          // the real implementation

	Ctx    context.Context         // stage context
	Cancel context.CancelCauseFunc // cancel to stop the stage

	B *Bgpipe      // parent
	P *pipe.Pipe   // bgpfix pipe
	K *koanf.Koanf // integrated config (args / config file / etc)

	Index   int          // stage index (zero means internal)
	Cmd     string       // stage command name
	Name    string       // human-friendly stage name
	Flags   []string     // consumed flags
	Args    []string     // consumed args
	Options StageOptions // stage options

	FilterIn  []*filter.Filter // message filter for pipe callbacks (input to stage)
	FilterOut []*filter.Filter // message filter for pipe inputs (output from stage)

	IsFirst bool    // is the first stage in pipe? (the L peer)
	IsLast  bool    // is the last stage in pipe? (the R peer)
	IsRight bool    // write L->R msgs + capture L->R msgs?
	IsLeft  bool    // write R->L msgs + capture R->L msgs?
	IsBidir bool    // true iff IsRight && IsLeft
	Dir     dir.Dir // target direction (IsLeft/IsRight translated, can be DIR_LR)
	// contains filtered or unexported fields
}

type StageOptions struct {
	Descr  string            // one-line description
	Flags  *pflag.FlagSet    // CLI flags
	Usage  string            // usage string
	Args   []string          // required argument names
	Events map[string]string // event names and descriptions

	IsProducer bool // produces messages? (= writes Line.Input?)
	IsConsumer bool // consumes messages? (= reads Line.Out?)
	IsStdin    bool // reads from stdin?
	IsStdout   bool // writes to stdout?
	Bidir      bool // allow -LR (bidir mode)?
	FilterIn   bool // allow stage input filtering? (must have callbacks)
	FilterOut  bool // allow stage output filtering? (must have inputs)
}