primitive

Documentation

Index

• Constants
• func FillSinkCredential(dst, src SinkCredential)
• func GetDeadLetterEventbusName(eventbusID vanus.ID) string
• func LoadConfig(filename string, config interface{}) error
• type Action
• type AkSkSinkCredential
  • func (c *AkSkSinkCredential) GetType() CredentialType
• type Closer
• type CredentialType
• type GCloudSinkCredential
  • func (c *GCloudSinkCredential) GetType() CredentialType
• type Initializer
• type KvStorageConfig
• type OffsetType
• type PlainSinkCredential
  • func (c *PlainSinkCredential) GetType() CredentialType
• type Protocol
• type ProtocolSetting
• type ServerState
• type SinkCredential
  • func NewAkSkSinkCredential(accessKeyID, secretAccessKey string) SinkCredential
  • func NewGCloudSinkCredential(credentialJSON string) SinkCredential
  • func NewPlainSinkCredential(identifier, secret string) SinkCredential
• type Subscription
  • func (sub *Subscription) String() string
• type SubscriptionConfig
  • func (c *SubscriptionConfig) GetMaxRetryAttempts() int32
  • func (c *SubscriptionConfig) String() string
• type SubscriptionFilter
• type SubscriptionFilterList
  • func (l SubscriptionFilterList) String() string
• type TemplateConfig
  • func (tc *TemplateConfig) RecognizeTemplateType() (TemplateType, bool)
• type TemplateType
• type Transformer
  • func (t *Transformer) Exist() bool
  • func (t *Transformer) String() string
• type URI

Constants

const (
	SystemEventbusNamePrefix = "__"
	TimerEventbusName        = SystemEventbusNamePrefix + "Timer_RS"
	RetryEventbusName        = SystemEventbusNamePrefix + "retry_eb"

	XVanus               = "xvanus"
	XVanusEventbus       = XVanus + "eventbus"
	XVanusDeliveryTime   = XVanus + "deliverytime"
	XVanusRetryAttempts  = XVanus + "retryattempts"
	XVanusSubscriptionID = XVanus + "subid"

	LastDeliveryTime  = XVanus + "lastdltime"
	LastDeliveryError = XVanus + "lastdlerror"
	DeadLetterReason  = XVanus + "dlreason"

	MaxRetryAttempts = 32

	DefaultNamespace = "default"
	SystemNamespace  = "vanus-system"

	DefaultUser = "admin"
)

Functions

func FillSinkCredential

func FillSinkCredential(dst, src SinkCredential)

func GetDeadLetterEventbusName

func GetDeadLetterEventbusName(eventbusID vanus.ID) string

func LoadConfig

func LoadConfig(filename string, config interface{}) error

Types

type Action

type Action struct {
	Command []interface{} `json:"command"`
}

type AkSkSinkCredential

type AkSkSinkCredential struct {
	AccessKeyID     string `json:"access_key_id"`
	SecretAccessKey string `json:"secret_access_key"`
}

func (*AkSkSinkCredential) GetType

func (c *AkSkSinkCredential) GetType() CredentialType

type Closer

type Closer interface {
	Close(ctx context.Context) error
}

type CredentialType

type CredentialType string

const (
	Plain  CredentialType = "plain"
	AWS    CredentialType = "aws"
	GCloud CredentialType = "gcloud"

	SecretsMask = "******"
)

type GCloudSinkCredential

type GCloudSinkCredential struct {
	CredentialJSON string `json:"credential_json"`
}

func (*GCloudSinkCredential) GetType

func (c *GCloudSinkCredential) GetType() CredentialType

type Initializer

type Initializer interface {
	Initialize(context.Context) error
}

type KvStorageConfig

type KvStorageConfig struct {
	KeyPrefix  string   `yaml:"key_prefix" json:"keyPrefix"`
	ServerList []string `yaml:"server_list" json:"serverList"`
}

type OffsetType

type OffsetType int32

const (
	LatestOffset   OffsetType = 0
	EarliestOffset OffsetType = 1
	Timestamp      OffsetType = 2
)

type PlainSinkCredential

type PlainSinkCredential struct {
	Identifier string `json:"identifier"`
	Secret     string `json:"secret"`
}

func (*PlainSinkCredential) GetType

func (c *PlainSinkCredential) GetType() CredentialType

type Protocol

type Protocol string

const (
	HTTPProtocol      Protocol = "http"
	AwsLambdaProtocol Protocol = "aws-lambda"
	GCloudFunctions   Protocol = "gcloud-functions"
	GRPC              Protocol = "grpc"
)

type ProtocolSetting

type ProtocolSetting struct {
	Headers map[string]string `json:"headers,omitempty"`
}

type ServerState

type ServerState string

const (
	ServerStateCreated  ServerState = "created"
	ServerStateStarted  ServerState = "started"
	ServerStateRunning  ServerState = "running"
	ServerStateStopping ServerState = "stopping"
	ServerStateStopped  ServerState = "stopped"
)

type SinkCredential

type SinkCredential interface {
	GetType() CredentialType
}

func NewAkSkSinkCredential

func NewAkSkSinkCredential(accessKeyID, secretAccessKey string) SinkCredential

func NewGCloudSinkCredential

func NewGCloudSinkCredential(credentialJSON string) SinkCredential

func NewPlainSinkCredential

func NewPlainSinkCredential(identifier, secret string) SinkCredential

type Subscription

type Subscription struct {
	ID                   vanus.ID               `json:"id"`
	Filters              SubscriptionFilterList `json:"filters,omitempty"`
	Sink                 URI                    `json:"sink,omitempty"`
	EventbusID           vanus.ID               `json:"eventbus"`
	DeadLetterEventbusID vanus.ID               `json:"dead_letter_eventbus_id"`
	RetryEventbusID      vanus.ID               `json:"retry_eventbus_id"`
	TimerEventbusID      vanus.ID               `json:"timer_eventbus_id"`
	Offsets              info.ListOffsetInfo    `json:"offsets"`
	Transformer          *Transformer           `json:"transformer,omitempty"`
	Config               SubscriptionConfig     `json:"config,omitempty"`
	Protocol             Protocol               `json:"protocol,omitempty"`
	ProtocolSetting      *ProtocolSetting       `json:"protocol_setting,omitempty"`
	SinkCredential       SinkCredential         `json:"sink_credential,omitempty"`
}

func (*Subscription) String

func (sub *Subscription) String() string

type SubscriptionConfig

type SubscriptionConfig struct {
	RateLimit uint32 `json:"rate_limit,omitempty"`
	// consumer from
	OffsetType        OffsetType `json:"offset_type,omitempty"`
	OffsetTimestamp   *uint64    `json:"offset_timestamp,omitempty"`
	DeliveryTimeout   uint32     `json:"delivery_timeout,omitempty"`
	MaxRetryAttempts  *uint32    `json:"max_retry_attempts,omitempty"`
	DisableDeadLetter bool       `json:"disable_dead_letter,omitempty"`
	// send event with ordered
	OrderedEvent bool `json:"ordered_event"`
}

func (*SubscriptionConfig) GetMaxRetryAttempts

func (c *SubscriptionConfig) GetMaxRetryAttempts() int32

GetMaxRetryAttempts return MaxRetryAttempts if nil return -1.

func (*SubscriptionConfig) String

func (c *SubscriptionConfig) String() string

type SubscriptionFilter

type SubscriptionFilter struct {
	Exact  map[string]string      `json:"exact,omitempty"`
	Prefix map[string]string      `json:"prefix,omitempty"`
	Suffix map[string]string      `json:"suffix,omitempty"`
	CeSQL  string                 `json:"ce_sql,omitempty"`
	Not    *SubscriptionFilter    `json:"not,omitempty"`
	All    SubscriptionFilterList `json:"all,omitempty"`
	Any    SubscriptionFilterList `json:"any,omitempty"`
	CEL    string                 `json:"cel,omitempty"`
}

type SubscriptionFilterList

type SubscriptionFilterList []*SubscriptionFilter

func (SubscriptionFilterList) String

func (l SubscriptionFilterList) String() string

type TemplateConfig

type TemplateConfig struct {
	Type     TemplateType `json:"template_type,omitempty"`
	Template string       `json:"template,omitempty"`
}

func (*TemplateConfig) RecognizeTemplateType

func (tc *TemplateConfig) RecognizeTemplateType() (TemplateType, bool)

type TemplateType

type TemplateType string

const (
	TemplateTypeUnspecified TemplateType = ""
	TemplateTypeNone        TemplateType = "none"
	TemplateTypeText        TemplateType = "text"
	TemplateTypeJSON        TemplateType = "json"
)

type Transformer

type Transformer struct {
	Define   map[string]string `json:"define,omitempty"`
	Pipeline []*Action         `json:"pipeline,omitempty"`
	Template TemplateConfig    `json:",inline"`
}

func (*Transformer) Exist

func (t *Transformer) Exist() bool

func (*Transformer) String

func (t *Transformer) String() string

type URI

type URI string