core

Documentation

Index

• Constants
• Variables
• func IsAddr(v string) bool
• func IsBind(v string) bool
• func IsFile(v string) bool
• func ParseEvents(src []string, stage_defaults ...string) []string
• func ParseTypes(src []string, dst []msg.Type) ([]msg.Type, error)
• type Bgpipe
  • func NewBgpipe(repo ...map[string]NewStage) *Bgpipe
  • func (b *Bgpipe) AddRepo(cmds map[string]NewStage)
  • func (b *Bgpipe) AddStage(idx int, cmd string) (*StageBase, error)
  • func (b *Bgpipe) AttachStages() error
  • func (b *Bgpipe) Configure() error
  • func (b *Bgpipe) KillEvent(ev *pipe.Event) bool
  • func (b *Bgpipe) LogEvent(ev *pipe.Event) bool
  • func (b *Bgpipe) NewStage(cmd string) *StageBase
  • func (b *Bgpipe) Run() error
  • func (b *Bgpipe) StageCount() int
  • func (b *Bgpipe) StageDump(d dir.Dir, w io.Writer) (total int)
• type NewStage
• type Stage
• type StageBase
  • func (s *StageBase) Attach() error
  • func (s *StageBase) Errorf(format string, a ...any) error
  • func (s *StageBase) Event(et string, args ...any) *pipe.Event
  • func (s *StageBase) Prepare() error
  • func (s *StageBase) Run() error
  • func (s *StageBase) Running() bool
  • func (s *StageBase) Stop() error
  • func (s *StageBase) String() string
  • func (s *StageBase) StringLR() string
• type StageOptions

Constants

const (
	StyleNone      = ""
	StyleBlack     = "\033[30m"
	StyleRed       = "\033[31m"
	StyleGreen     = "\033[32m"
	StyleYellow    = "\033[33m"
	StyleBlue      = "\033[34m"
	StyleMagenta   = "\033[35m"
	StyleCyan      = "\033[36m"
	StyleWhite     = "\033[37m"
	StyleBold      = "\033[1m"
	StyleUnderline = "\033[4m"
	StyleReset     = "\033[0m"
)

Variables

var (
	ErrStageCmd     = errors.New("invalid stage command")
	ErrStageDiff    = errors.New("already defined but different")
	ErrStageStopped = errors.New("stage stopped")
	ErrFirstOrLast  = errors.New("must be either the first or the last stage")
	ErrInject       = errors.New("invalid --in option value")
	ErrLR           = errors.New("select either --left or --right, not both")
)

Functions

func IsAddr

func IsAddr(v string) bool

func IsBind

func IsBind(v string) bool

func IsFile

func IsFile(v string) bool

func ParseEvents

func ParseEvents(src []string, stage_defaults ...string) []string

ParseEvents parses events in src and returns the result, or nil. If stage_defaults is given, events like "foobar" are translated to "foobar/stage_defaults[:]".

func ParseTypes

func ParseTypes(src []string, dst []msg.Type) ([]msg.Type, error)

Types

type Bgpipe

type Bgpipe struct {
	zerolog.Logger

	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
}

Bgpipe represents a BGP pipeline consisting of several stages, built on top of bgpfix.Pipe

func NewBgpipe

func NewBgpipe(repo ...map[string]NewStage) *Bgpipe

NewBgpipe creates a new bgpipe instance using given repositories of stage commands

func (*Bgpipe) AddRepo

func (b *Bgpipe) AddRepo(cmds map[string]NewStage)

AddRepo adds mapping between stage commands and their NewStageFunc

func (*Bgpipe) AddStage

func (b *Bgpipe) AddStage(idx int, cmd string) (*StageBase, error)

AddStage adds and returns a new stage at idx for cmd, or returns an existing instance if it's for the same cmd.

func (*Bgpipe) AttachStages

func (b *Bgpipe) AttachStages() error

AttachStages attaches all stages to pipe

func (*Bgpipe) Configure

func (b *Bgpipe) Configure() error

Configure configures bgpipe

func (*Bgpipe) KillEvent

func (b *Bgpipe) KillEvent(ev *pipe.Event) bool

KillEvent brutally kills the session because of given event ev

func (*Bgpipe) LogEvent

func (b *Bgpipe) LogEvent(ev *pipe.Event) bool

LogEvent logs given event

func (*Bgpipe) NewStage

func (b *Bgpipe) NewStage(cmd string) *StageBase

NewStage returns new stage for given cmd, or nil on error

func (*Bgpipe) Run

func (b *Bgpipe) Run() error

Run configures and runs the bgpipe

func (*Bgpipe) StageCount

func (b *Bgpipe) StageCount() int

StageCount returns the number of stages added to the pipe

func (*Bgpipe) StageDump

func (b *Bgpipe) StageDump(d dir.Dir, w io.Writer) (total int)

StageDump prints all stages in dir direction in textual form to w (by default stdout)

type NewStage

type NewStage func(base *StageBase) Stage

NewStage returns a new Stage for given parent base. It should modify base.Options.

type 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
}

Stage implements a bgpipe stage

type StageBase

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

	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
}

StageBase represents a bgpipe stage base

func (*StageBase) Attach

func (s *StageBase) Attach() error

Attach is the default Stage implementation that does nothing.

func (*StageBase) Errorf

func (s *StageBase) Errorf(format string, a ...any) error

Errorf wraps fmt.Errorf and adds a prefix with the stage name

func (*StageBase) Event

func (s *StageBase) Event(et string, args ...any) *pipe.Event

Event sends an event, prefixing et with stage name + slash

func (*StageBase) Prepare

func (s *StageBase) Prepare() error

Prepare is the default Stage implementation that does nothing.

func (*StageBase) Run

func (s *StageBase) Run() error

Run is the default Stage implementation that just waits for the context and returns its cancel cause

func (*StageBase) Running

func (s *StageBase) Running() bool

Running returns true if the stage is in Run(), false otherwise.

func (*StageBase) Stop

func (s *StageBase) Stop() error

Stop is the default Stage implementation that does nothing.

func (*StageBase) String

func (s *StageBase) String() string

String returns stage "[index] name" or "name" if index is 0

func (*StageBase) StringLR

func (s *StageBase) StringLR() string

StringDir returns eg. "-LR [FIRST]" depending on stage direction

type StageOptions

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 to Line input)
	IsConsumer bool // consumes messages? (reads from Line output)
	IsStdin    bool // reads from stdin?
	IsStdout   bool // writes to stdout?
	Bidir      bool // allow -LR (bidir mode)?
}

StageOptions describe high-level settings of a stage