connection

Documentation

Index

• Constants
• Variables
• func CreateMockConnection(ctx api.StreamContext) modules.Connection
• func CreateMockErrConnection(ctx api.StreamContext) modules.Connection
• func DetachConnection(ctx api.StreamContext, conId string) error
• func DropNameConnection(ctx api.StreamContext, selId string) error
• func GetAllConnectionStatus(ctx api.StreamContext) map[string]modules.ConnectionStatus
• func InitConnectionManager()
• func InitConnectionManager4Test() error
• func InitMockTest()
• func NewExponentialBackOff() *backoff.ExponentialBackOff
• func PingConnection(ctx api.StreamContext, id string) error
• func ReloadNamedConnection() error
• type ConnWrapper
  • func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnWrapper, error)
  • func FetchConnection(ctx api.StreamContext, refId, typ string, props map[string]interface{}, ...) (*ConnWrapper, error)
  • func (cw *ConnWrapper) IsInitialized() bool
  • func (cw *ConnWrapper) SetConn(conn modules.Connection, err error)
  • func (cw *ConnWrapper) Wait() (modules.Connection, error)
• type Manager
• type Meta
  • func GetAllConnectionsMeta() []*Meta
  • func GetConnectionDetail(_ api.StreamContext, id string) (*Meta, error)
  • func (meta *Meta) GetRefCount() int
  • func (meta *Meta) NotifyStatus(status string, s string)

Constants

const (
	DefaultInitialInterval = 100 * time.Millisecond
	DefaultMaxInterval     = 1 * time.Second
)

Variables

var DefaultBackoffMaxElapsedDuration = 100 * time.Minute

Functions

func CreateMockConnection

func CreateMockConnection(ctx api.StreamContext) modules.Connection

func CreateMockErrConnection

func CreateMockErrConnection(ctx api.StreamContext) modules.Connection

func DetachConnection

func DetachConnection(ctx api.StreamContext, conId string) error

func DropNameConnection

func DropNameConnection(ctx api.StreamContext, selId string) error

func GetAllConnectionStatus

func GetAllConnectionStatus(ctx api.StreamContext) map[string]modules.ConnectionStatus

func InitConnectionManager

func InitConnectionManager()

func InitConnectionManager4Test

func InitConnectionManager4Test() error

func InitMockTest

func InitMockTest()

func NewExponentialBackOff

func NewExponentialBackOff() *backoff.ExponentialBackOff

func PingConnection

func PingConnection(ctx api.StreamContext, id string) error

func ReloadNamedConnection

func ReloadNamedConnection() error

ReloadNamedConnection is called when server starts. It initializes all stored named connections

Types

type ConnWrapper

type ConnWrapper struct {
	ID string
	// contains filtered or unexported fields
}

func CreateNamedConnection

func CreateNamedConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnWrapper, error)

func FetchConnection

func FetchConnection(ctx api.StreamContext, refId, typ string, props map[string]interface{}, sc api.StatusChangeHandler) (*ConnWrapper, error)

FetchConnection is called by source/sink to get or create an anonymous connection instance in the pool

func (*ConnWrapper) IsInitialized

func (cw *ConnWrapper) IsInitialized() bool

func (*ConnWrapper) SetConn

func (cw *ConnWrapper) SetConn(conn modules.Connection, err error)

func (*ConnWrapper) Wait

func (cw *ConnWrapper) Wait() (modules.Connection, error)

type Manager

type Manager struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

type Meta

type Meta struct {
	ID    string         `json:"id"`
	Typ   string         `json:"typ"`
	Props map[string]any `json:"props"`

	Named bool `json:"-"`
	// contains filtered or unexported fields
}

func GetAllConnectionsMeta

func GetAllConnectionsMeta() []*Meta

func GetConnectionDetail

func GetConnectionDetail(_ api.StreamContext, id string) (*Meta, error)

func (*Meta) GetRefCount

func (meta *Meta) GetRefCount() int

func (*Meta) NotifyStatus

func (meta *Meta) NotifyStatus(status string, s string)