pubsub

Documentation

Index

• Constants
• Variables
• func ApplyBulkPublishResiliency(ctx context.Context, req *contribPubsub.BulkPublishRequest, ...) (contribPubsub.BulkPublishResponse, error)
• func NewBulkSubscribeEnvelope(req *BulkSubscribeEnvelope) map[string]interface{}
• func NewCloudEvent(req *CloudEvent, metadata map[string]string) (map[string]interface{}, error)
• func NewDefaultBulkPublisher(p contribPubsub.PubSub) contribPubsub.BulkPublisher
• func NewDefaultBulkSubscriber(p contribPubsub.PubSub) *defaultBulkSubscriber
• func NewOutbox(publishFn func(context.Context, *contribPubsub.PublishRequest) error, ...) outbox.Outbox
• type Adapter
• type BulkSubscribe
• type BulkSubscribeEnvelope
• type BulkSubscribeJSON
• type BulkSubscribeMessageItem
• type CloudEvent
• type Expr
• type NotAllowedError
  • func (e NotAllowedError) Error() string
• type NotFoundError
  • func (e NotFoundError) Error() string
• type RoutesJSON
• type Rule
• type RuleJSON
• type Subscription
  • func DeclarativeKubernetes(ctx context.Context, client operatorv1pb.OperatorClient, podName string, ...) []Subscription
  • func DeclarativeLocal(resourcesPaths []string, namespace string, log logger.Logger) (subs []Subscription)
  • func GetSubscriptionsGRPC(ctx context.Context, channel runtimev1pb.AppCallbackClient, log logger.Logger, ...) ([]Subscription, error)
  • func GetSubscriptionsHTTP(ctx context.Context, channel channel.AppChannel, log logger.Logger, ...) ([]Subscription, error)
• type SubscriptionJSON

Constants

const (
	Metadata = "metadata"
	Entries  = "entries"
)

const (
	APIVersionV1alpha1 = "dapr.io/v1alpha1"
	APIVersionV2alpha1 = "dapr.io/v2alpha1"
)

Variables

var ErrBulkPublishFailure = errors.New("bulk publish failed")

Functions

func ApplyBulkPublishResiliency

func ApplyBulkPublishResiliency(ctx context.Context, req *contribPubsub.BulkPublishRequest,
	policyDef *resiliency.PolicyDefinition,
	bulkPublisher contribPubsub.BulkPublisher,
) (contribPubsub.BulkPublishResponse, error)

func NewBulkSubscribeEnvelope

func NewBulkSubscribeEnvelope(req *BulkSubscribeEnvelope) map[string]interface{}

func NewCloudEvent

func NewCloudEvent(req *CloudEvent, metadata map[string]string) (map[string]interface{}, error)

NewCloudEvent encapsulates the creation of a Dapr cloudevent from an existing cloudevent or a raw payload.

func NewDefaultBulkPublisher

func NewDefaultBulkPublisher(p contribPubsub.PubSub) contribPubsub.BulkPublisher

NewDefaultBulkPublisher returns a new defaultBulkPublisher from a PubSub.

func NewDefaultBulkSubscriber

func NewDefaultBulkSubscriber(p contribPubsub.PubSub) *defaultBulkSubscriber

NewDefaultBulkSubscriber returns a new defaultBulkSubscriber from a PubSub.

func NewOutbox

func NewOutbox(publishFn func(context.Context, *contribPubsub.PublishRequest) error, getPubsubFn func(string) (contribPubsub.PubSub, bool), getStateFn func(string) (state.Store, bool), cloudEventExtractorFn func(map[string]any, string) string, namespace string) outbox.Outbox

NewOutbox returns an instance of an Outbox.

Types

type Adapter

type Adapter interface {
	Publish(context.Context, *contribPubsub.PublishRequest) error
	BulkPublish(context.Context, *contribPubsub.BulkPublishRequest) (contribPubsub.BulkPublishResponse, error)
	Outbox() outbox.Outbox
}

Adapter is the interface for message buses.

type BulkSubscribe

type BulkSubscribe struct {
	Enabled            bool  `json:"enabled"`
	MaxMessagesCount   int32 `json:"maxMessagesCount,omitempty"`
	MaxAwaitDurationMs int32 `json:"maxAwaitDurationMs,omitempty"`
}

type BulkSubscribeEnvelope

type BulkSubscribeEnvelope struct {
	ID        string
	Entries   []BulkSubscribeMessageItem
	Metadata  map[string]string
	Topic     string
	Pubsub    string
	EventType string
}

type BulkSubscribeJSON

type BulkSubscribeJSON struct {
	Enabled            bool  `json:"enabled"`
	MaxMessagesCount   int32 `json:"maxMessagesCount,omitempty"`
	MaxAwaitDurationMs int32 `json:"maxAwaitDurationMs,omitempty"`
}

type BulkSubscribeMessageItem

type BulkSubscribeMessageItem struct {
	EntryId     string            `json:"entryId"` //nolint:stylecheck
	Event       interface{}       `json:"event"`
	Metadata    map[string]string `json:"metadata"`
	ContentType string            `json:"contentType,omitempty"`
}

type CloudEvent

type CloudEvent struct {
	ID              string `mapstructure:"cloudevent.id"`
	Data            []byte `mapstructure:"-"` // cannot be overridden
	Topic           string `mapstructure:"-"` // cannot be overridden
	Pubsub          string `mapstructure:"-"` // cannot be overridden
	DataContentType string `mapstructure:"-"` // cannot be overridden
	TraceID         string `mapstructure:"cloudevent.traceid"`
	TraceState      string `mapstructure:"cloudevent.tracestate"`
	Source          string `mapstructure:"cloudevent.source"`
	Type            string `mapstructure:"cloudevent.type"`
	TraceParent     string `mapstructure:"cloudevent.traceparent"`
}

CloudEvent is a request object to create a Dapr compliant cloudevent. The cloud event properties can manually be overwritten by using metadata beginning with "cloudevent." as prefix.

type Expr

type Expr interface {
	Eval(variables map[string]interface{}) (interface{}, error)
}

type NotAllowedError

type NotAllowedError struct {
	Topic string
	ID    string
}

pubsub.NotAllowedError is returned by the runtime when publishing is forbidden.

func (NotAllowedError) Error

func (e NotAllowedError) Error() string

type NotFoundError

type NotFoundError struct {
	PubsubName string
}

pubsub.NotFoundError is returned by the runtime when the pubsub does not exist.

func (NotFoundError) Error

func (e NotFoundError) Error() string

type RoutesJSON

type RoutesJSON struct {
	Rules   []*RuleJSON `json:"rules,omitempty"`
	Default string      `json:"default,omitempty"`
}

type Rule

type Rule struct {
	Match Expr   `json:"match"`
	Path  string `json:"path"`
}

type RuleJSON

type RuleJSON struct {
	Match string `json:"match"`
	Path  string `json:"path"`
}

type Subscription

type Subscription struct {
	PubsubName      string            `json:"pubsubname"`
	Topic           string            `json:"topic"`
	DeadLetterTopic string            `json:"deadLetterTopic"`
	Metadata        map[string]string `json:"metadata"`
	Rules           []*Rule           `json:"rules,omitempty"`
	Scopes          []string          `json:"scopes"`
	BulkSubscribe   *BulkSubscribe    `json:"bulkSubscribe"`
}

func DeclarativeKubernetes

func DeclarativeKubernetes(ctx context.Context, client operatorv1pb.OperatorClient, podName string, namespace string, log logger.Logger) []Subscription

DeclarativeKubernetes loads subscriptions from the operator when running in Kubernetes.

func DeclarativeLocal

func DeclarativeLocal(resourcesPaths []string, namespace string, log logger.Logger) (subs []Subscription)

DeclarativeLocal loads subscriptions from the given local resources path.

func GetSubscriptionsGRPC

func GetSubscriptionsGRPC(ctx context.Context, channel runtimev1pb.AppCallbackClient, log logger.Logger, r resiliency.Provider) ([]Subscription, error)

func GetSubscriptionsHTTP

func GetSubscriptionsHTTP(ctx context.Context, channel channel.AppChannel, log logger.Logger, r resiliency.Provider) ([]Subscription, error)

type SubscriptionJSON

type SubscriptionJSON struct {
	PubsubName      string            `json:"pubsubname"`
	Topic           string            `json:"topic"`
	DeadLetterTopic string            `json:"deadLetterTopic"`
	Metadata        map[string]string `json:"metadata,omitempty"`
	Route           string            `json:"route"`  // Single route from v1alpha1
	Routes          RoutesJSON        `json:"routes"` // Multiple routes from v2alpha1
	BulkSubscribe   BulkSubscribeJSON `json:"bulkSubscribe,omitempty"`
}