pipe

package

Go to main page

Versions in this module

v0

v0.4.0

Oct 4, 2024

Changes in this version

type Context

+ func GetContext(m *msg.Msg) *Context

type Pipe

+ func (p *Pipe) ParseMsg(m *msg.Msg) error

v0.3.0

May 22, 2024

Changes in this version

+ var EVENT_EOR = "bgpfix/pipe.EOR"

+ var EVENT_EOR_AF = "bgpfix/pipe.EOR_AF"

type Line

+ EoR *xsync.MapOf[af.AF, int64]

v0.2.0

Apr 12, 2024

Changes in this version

+ func HasContext(m *msg.Msg) bool

+ func HasTags(m *msg.Msg) bool

+ func MsgTags(m *msg.Msg) map[string]string

type Action

+ const ACTION_MASK

+ func (ac *Action) Accept()

+ func (ac *Action) Borrow()

+ func (ac *Action) Drop()

+ func (ac *Action) FromJSON(src []byte) error

+ func (ac Action) IsAccept() bool

+ func (ac Action) IsBorrow() bool

+ func (ac Action) IsDrop() bool

+ func (ac Action) ToJSON(dst []byte) []byte

type Callback

+ Dropped bool

+ func (cb *Callback) Disable() bool

+ func (cb *Callback) Drop()

+ func (cb *Callback) Enable() bool

+ func (cb *Callback) String() string

type Context

+ func (mx *Context) DropTags()

+ func (mx *Context) HasTags() bool

type Event

+ Action Action

+ Handler *Handler

type Handler

+ Dropped bool

+ func (h *Handler) Disable() bool

+ func (h *Handler) Drop()

+ func (h *Handler) Enable() bool

+ func (h *Handler) String() string

type Input

+ func (in *Input) WriteFunc(src []byte, cb CallbackFunc) (int, error)

type Line

+ func (l *Line) WriteOutput(m *msg.Msg) (write_error error)

v0.1.7

Mar 22, 2024

Changes in this version

+ func ActionAccept(m *msg.Msg) *msg.Msg

+ func ActionBorrow(m *msg.Msg) *msg.Msg

+ func ActionClear(m *msg.Msg) *msg.Msg

+ func ActionDrop(m *msg.Msg) *msg.Msg

+ func ActionIsAccept(m *msg.Msg) bool

+ func ActionIsBorrow(m *msg.Msg) bool

+ func ActionIsDrop(m *msg.Msg) bool

type Context

+ func (mx *Context) ActionAccept() *Context

+ func (mx *Context) ActionBorrow() *Context

+ func (mx *Context) ActionClear() *Context

+ func (mx *Context) ActionDrop() *Context

+ func (mx *Context) ActionIsAccept() bool

+ func (mx *Context) ActionIsBorrow() bool

+ func (mx *Context) ActionIsDrop() bool

type Pipe

+ func (p *Pipe) GetMsg() (m *msg.Msg)

+ func (p *Pipe) PutMsg(m *msg.Msg)

v0.1.6

Feb 19, 2024

Changes in this version

type Context

+ Action Action

+ Callback *Callback

+ Input *Proc

+ Pipe *Pipe

+ func MsgContext(m *msg.Msg) *Context

+ func (pc *Context) Clear()

+ func (pc *Context) FromJSON(src []byte) error

+ func (pc *Context) GetTag(tag string) string

+ func (pc *Context) HasTag(tag string) bool

+ func (pc *Context) Reset()

+ func (pc *Context) SetTag(tag string, val ...string)

+ func (pc *Context) Tags() map[string]string

+ func (pc *Context) ToJSON(dst []byte) []byte

type Options

+ Procs []*Proc

+ func (o *Options) AddProc(dir msg.Dir, tpl ...*Proc) *Proc

type Pipe

+ func (p *Pipe) LineFor(dst msg.Dir) *Line

+ type Proc struct

+ CallbackFilter FilterMode

+ Dir msg.Dir

+ FilterValue any

+ Id int

+ In chan *msg.Msg

+ Line *Line

+ Name string

+ Pipe *Pipe

+ Reverse bool

+ Stats struct{ ... }

+ func (in *Proc) Close()

+ func (in *Proc) Wait()

+ func (in *Proc) Write(src []byte) (n int, err error)

+ func (in *Proc) WriteMsg(m *msg.Msg) (write_error error)

v0.1.4

Oct 30, 2023

Changes in this version

type Action

+ const ACTION_OK

v0.1.3

Oct 26, 2023

v0.1.2

Oct 26, 2023

v0.1.1

Oct 26, 2023

Changes in this version

+ var DefaultOptions = Options

+ var EVENT_ALIVE = "bgpfix/pipe.ALIVE"

+ var EVENT_ESTABLISHED = "bgpfix/pipe.ESTABLISHED"

+ var EVENT_OPEN = "bgpfix/pipe.OPEN"

+ var EVENT_PARSE = "bgpfix/pipe.PARSE"

+ var EVENT_START = "bgpfix/pipe.START"

+ var EVENT_STOP = "bgpfix/pipe.STOP"

+ var EVENT_UPDATE = "bgpfix/pipe.UPDATE"

+ var ErrInClosed = errors.New("input channel closed")

+ var ErrOutClosed = errors.New("output channel closed")

+ var ErrStopped = errors.New("pipe stopped")

+ type Action byte

+ const ACTION_ACCEPT

+ const ACTION_BORROW

+ const ACTION_CONTINUE

+ const ACTION_DROP

+ func (ac *Action) Add(a Action)

+ func (ac *Action) Clear()

+ func (ac Action) Is(a Action) bool

+ func (ac Action) Not(a Action) bool

+ type Callback struct

+ Dir msg.Dir

+ Enabled *atomic.Bool

+ Func CallbackFunc

+ Id int

+ Name string

+ Order int

+ Post bool

+ Pre bool

+ Raw bool

+ Types []msg.Type

+ type CallbackFunc func(m *msg.Msg) (add_action Action)

+ type Event struct

+ Dir msg.Dir

+ Error error

+ Msg *msg.Msg

+ Pipe *Pipe

+ Seq uint64

+ Time time.Time

+ Type string

+ Value any

+ func (ev *Event) String() string

+ func (ev *Event) Wait()

+ type FilterMode = int

+ const FILTER_ALL

+ const FILTER_EQ

+ const FILTER_GE

+ const FILTER_GT

+ const FILTER_LE

+ const FILTER_LT

+ const FILTER_NE

+ const FILTER_NONE

+ type Handler struct

+ Dir msg.Dir

+ Enabled *atomic.Bool

+ Func HandlerFunc

+ Id int

+ Name string

+ Order int

+ Post bool

+ Pre bool

+ Types []string

+ type HandlerFunc func(ev *Event) (keep_event bool)

+ type Input struct

+ CallbackFilter FilterMode

+ Dir msg.Dir

+ FilterValue any

+ Id int

+ In chan *msg.Msg

+ Line *Line

+ Name string

+ Pipe *Pipe

+ Reverse bool

+ Stats struct{ ... }

+ func (in *Input) Close()

+ func (in *Input) Wait()

+ func (in *Input) Write(src []byte) (n int, err error)

+ func (in *Input) WriteMsg(m *msg.Msg) (write_error error)

+ type Line struct

+ Dir msg.Dir

+ In chan *msg.Msg

+ LastAlive atomic.Int64

+ LastOpen atomic.Int64

+ LastUpdate atomic.Int64

+ Open atomic.Pointer[msg.Open]

+ Out chan *msg.Msg

+ Pipe *Pipe

+ func (l *Line) Close()

+ func (l *Line) CloseOutput()

+ func (l *Line) Read(dst []byte) (int, error)

+ func (l *Line) Wait()

+ func (l *Line) WriteOut(m *msg.Msg) (write_error error)

+ func (l *Line) WriteTo(w io.Writer) (int64, error)

+ type Options struct

+ Callbacks []*Callback

+ Caps bool

+ Handlers []*Handler

+ Inputs []*Input

+ Logger *zerolog.Logger

+ MsgPool *sync.Pool

+ func (o *Options) AddCallback(cbf CallbackFunc, tpl ...*Callback) *Callback

+ func (o *Options) AddHandler(hdf HandlerFunc, tpl ...*Handler) *Handler

+ func (o *Options) AddInput(dir msg.Dir, tpl ...*Input) *Input

+ func (o *Options) OnEstablished(hdf HandlerFunc) *Handler

+ func (o *Options) OnEvent(hdf HandlerFunc, types ...string) *Handler

+ func (o *Options) OnEventPost(hdf HandlerFunc, types ...string) *Handler

+ func (o *Options) OnEventPre(hdf HandlerFunc, types ...string) *Handler

+ func (o *Options) OnMsg(cbf CallbackFunc, dir msg.Dir, types ...msg.Type) *Callback

+ func (o *Options) OnMsgPost(cbf CallbackFunc, dir msg.Dir, types ...msg.Type) *Callback

+ func (o *Options) OnMsgPre(cbf CallbackFunc, dir msg.Dir, types ...msg.Type) *Callback

+ func (o *Options) OnParseError(hdf HandlerFunc) *Handler

+ func (o *Options) OnStart(hdf HandlerFunc) *Handler

+ func (o *Options) OnStop(hdf HandlerFunc) *Handler

+ type Pipe struct

+ Caps caps.Caps

+ KV *xsync.MapOf[string, any]

+ L *Line

+ R *Line

+ func NewPipe(ctx context.Context) *Pipe

+ func (p *Pipe) Event(et string, args ...any) *Event

+ func (p *Pipe) Get() (m *msg.Msg)

+ func (p *Pipe) LineFrom(src msg.Dir) *Line

+ func (p *Pipe) LineTo(dst msg.Dir) *Line

+ func (p *Pipe) Put(m *msg.Msg)

+ func (p *Pipe) Start()

+ func (p *Pipe) Started() bool

+ func (p *Pipe) Stop()

+ func (p *Pipe) Stopped() bool

+ func (p *Pipe) Wait()

+ type PipeContext struct

+ Action Action

+ Callback *Callback

+ Input *Input

+ Pipe *Pipe

+ func Context(m *msg.Msg) *PipeContext

+ func (pc *PipeContext) Clear()

+ func (pc *PipeContext) FromJSON(src []byte) error

+ func (pc *PipeContext) HasKV() bool

+ func (pc *PipeContext) KV() map[string]any

+ func (pc *PipeContext) Reset()

+ func (pc *PipeContext) ToJSON(dst []byte) []byte

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL