pipe

package

Go to main page

Versions in this module

v0

v0.18.0

Feb 16, 2026

v0.17.2

Feb 12, 2026

v0.17.1

Feb 10, 2026

Changes in this version

type Options

+ StopTimeout time.Duration

v0.17.0

Feb 7, 2026

Changes in this version

type Callback

+ func (cb *Callback) Blackhole()

v0.16.2

Jan 30, 2026

v0.16.1

Jan 29, 2026

Changes in this version

+ func DropTags(m *msg.Msg) bool

v0.16.0

Jan 27, 2026

v0.15.2

Jan 17, 2026

v0.10.1

Jan 15, 2026

v0.10.0

Jan 15, 2026

Changes in this version

+ func ActionHasAccept(m *msg.Msg) bool

+ func ActionHasBorrow(m *msg.Msg) bool

+ func ActionHasDrop(m *msg.Msg) bool

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

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

type Action

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

+ func (ac Action) HasAccept() bool

+ func (ac Action) HasBorrow() bool

+ func (ac Action) HasDrop() bool

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

type Callback

+ LimitRate *rate.Limiter

+ LimitSkip bool

type Input

+ LimitRate *rate.Limiter

+ LimitSkip bool

type Pipe

+ Cancel context.CancelCauseFunc

+ Ctx context.Context

v0.9.0

Dec 19, 2025

v0.8.0

Dec 12, 2025

v0.7.1

Nov 7, 2025

v0.7.0

Nov 7, 2025

v0.6.1

Oct 10, 2025

v0.6.0

Oct 10, 2025

v0.5.2

Oct 10, 2025

v0.5.1

Jun 13, 2025

Changes in this version

+ var ErrFilter = errors.New("filter error")

+ var ErrStarted = errors.New("pipe started")

type Callback

+ Filter *filter.Filter

+ type CbFilterMode = int

+ const CBFILTER_ALL

+ const CBFILTER_EQ

+ const CBFILTER_GE

+ const CBFILTER_GT

+ const CBFILTER_LE

+ const CBFILTER_LT

+ const CBFILTER_NE

+ const CBFILTER_NONE

type Context

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

+ func (mx *Context) DropTag(tag string) bool

+ func (mx *Context) GetTags() map[string]string

+ func (mx *Context) UseTags() map[string]string

type Input

+ CbFilter CbFilterMode

+ CbFilterValue any

+ Filter *filter.Filter

v0.5.0

Oct 4, 2024

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