Documentation
¶
Index ¶
- Constants
- Variables
- func ConsumerRefID(ctx api.StreamContext) string
- func CreateMockConnection(ctx api.StreamContext) modules.Connection
- func CreateMockErrConnection(ctx api.StreamContext) modules.Connection
- func DetachConnection(ctx api.StreamContext, conId string) error
- func DetachConnectionByRef(ctx api.StreamContext, conId, refId string) error
- func DropNameConnection(ctx api.StreamContext, selId string) error
- func InitConnectionManager(ctx context.Context)
- func InitConnectionManager4Test() error
- func LookupRefID(ctx api.StreamContext) (string, bool)
- func NewExponentialBackOffWithMaxElapsedTime(maxElapsedTime time.Duration) *backoff.ExponentialBackOff
- func PatrolConnectionStatusJob(ctx context.Context)
- func ReloadNamedConnection() error
- func WithLookupRefID(ctx *topoContext.DefaultContext, refID string) *topoContext.DefaultContext
- 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 FetchConnectionWithOptions(ctx api.StreamContext, opts FetchOptions) (*ConnWrapper, error)
- func UpdateConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnWrapper, error)
- type FetchOptions
- type Manager
- type Meta
Constants ¶
const ( DefaultInitialInterval = 100 * time.Millisecond DefaultMaxInterval = 10 * time.Second )
const (
LblName = "name"
)
Variables ¶
var ConnStatusGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "kuiper", Subsystem: "conn_status", Name: "gauge", Help: "gauge of connection status", }, []string{LblName})
Functions ¶
func ConsumerRefID ¶
func ConsumerRefID(ctx api.StreamContext) string
ConsumerRefID derives the stable Source/Sink consumer identity (ruleID + opID + instanceID) for the current context. Connect must save the returned value and pass it back verbatim to DetachConnectionByRef.
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 DetachConnectionByRef ¶ added in v2.2.8
func DetachConnectionByRef(ctx api.StreamContext, conId, refId string) error
DetachConnectionByRef detaches a connection using the reference ID supplied to FetchConnection.
func DropNameConnection ¶
func DropNameConnection(ctx api.StreamContext, selId string) error
func InitConnectionManager ¶
func InitConnectionManager4Test ¶
func InitConnectionManager4Test() error
func LookupRefID ¶
func LookupRefID(ctx api.StreamContext) (string, bool)
LookupRefID reads the framework-injected lookup resource identity. The second return value reports whether a non-empty identity was present.
func NewExponentialBackOffWithMaxElapsedTime ¶ added in v2.2.8
func NewExponentialBackOffWithMaxElapsedTime(maxElapsedTime time.Duration) *backoff.ExponentialBackOff
func PatrolConnectionStatusJob ¶ added in v2.2.0
func ReloadNamedConnection ¶
func ReloadNamedConnection() error
ReloadNamedConnection is called when server starts. It initializes all stored named connections
func WithLookupRefID ¶
func WithLookupRefID(ctx *topoContext.DefaultContext, refID string) *topoContext.DefaultContext
WithLookupRefID injects the framework-owned lookup resource identity into ctx. Called by lookup.CreateInstance before LookupSource.Connect; the source only reads and stores the value, never derives it.
The parameter is deliberately *topoContext.DefaultContext, the only context the lookup framework ever creates: there is no production need to inject into an arbitrary api.StreamContext, and wrapping one would silently drop its rule/op/instance identity.
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 the legacy compatibility shim. It keeps the historical connectionKey derivation (connectionSelector-or-refId) so unmigrated connectors keep resolving the same logical connection. The consumer ref, however, is normalized to ConsumerRefID(ctx): legacy DetachConnection derives exactly that value, and the historical habit of passing connection-identity material (endpoint/topic/URL) as refId never identified the consumer. Without this normalization every legacy Close would miss its ref and leak the reference. New code must use FetchConnectionWithOptions instead.
func FetchConnectionWithOptions ¶
func FetchConnectionWithOptions(ctx api.StreamContext, opts FetchOptions) (*ConnWrapper, error)
FetchConnectionWithOptions is the explicit-identity fetch path. Callers must canonicalize props/key and compute RefID before calling; Pool treats ConnectionKey as an opaque identity.
func UpdateConnection ¶
func UpdateConnection(ctx api.StreamContext, id, typ string, props map[string]any) (*ConnWrapper, error)
func (*ConnWrapper) IsInitialized ¶
func (cw *ConnWrapper) IsInitialized() bool
func (*ConnWrapper) Wait ¶
func (cw *ConnWrapper) Wait(connectorCtx api.StreamContext) (modules.Connection, error)
Wait will wait for connection connected or the caller interrupts (like rule exit)
type FetchOptions ¶
type FetchOptions struct {
// ConnectionKey is the opaque logical-connection identity. For named
// connections it is the connectionSelector; for anonymous ones it is
// the connector-provided canonical key (e.g. resolved SQL DB URL).
ConnectionKey string
// RefID identifies the holding consumer. Source/Sink use the
// rule+op+instance identity; lookup tables use the framework-injected
// lookup resource identity. It must be saved by the consumer at
// Connect time and passed back verbatim at detach time.
RefID string
// RequireExisting marks a fetch that may only attach to an already
// existing logical connection: a missing ConnectionKey is an error
// and anonymous creation is never performed. Used for
// connectionSelector/named references. False allows creating an
// anonymous Meta when the key is absent. This describes the fetch
// mode, not the Meta lifecycle type (Meta.Named).
RequireExisting bool
Type string
Props map[string]any
// StatusHandler receives connection status changes for this ref.
StatusHandler api.StatusChangeHandler
}
FetchOptions carries the explicit connection identity for the new fetch path (DESIGN §8, A1a). ConnectionKey selects which logical connection to share; RefID identifies which consumer holds it. The two must never be mixed: Pool never derives one from the other.
type Meta ¶
type Meta struct {
ID string `json:"id"`
Typ string `json:"typ"`
Props map[string]any `json:"props"`
// named means connection is created manually
Named bool `json:"named"`
// contains filtered or unexported fields
}
func GetAllConnectionsMeta ¶
func GetConnectionDetail ¶
func GetConnectionDetail(_ api.StreamContext, id string) (*Meta, error)
Source Files