pubsub

README

Pub Sub

Pub Subs provide a common way to interact with different message bus implementations to achieve reliable, high-scale scenarios based on event-driven async communications, while allowing users to opt-in to advanced capabilities using defined metadata.

Currently supported pub-subs are:

• Hazelcast
• Redis Streams
• NATS
• Kafka
• Azure Service Bus
• RabbitMQ
• Azure Event Hubs
• GCP Pub/Sub
• MQTT

Implementing a new Pub Sub

A compliant pub sub needs to implement the following interface:

type PubSub interface {
	Init(metadata Metadata) error
	Publish(req *PublishRequest) error
	Subscribe(req SubscribeRequest, handler func(msg *NewMessage) error) error
}

Documentation

Index

• Constants
• type AppResponse
• type AppResponseStatus
• type CloudEventsEnvelope
  • func NewCloudEventsEnvelope(id, source, eventType, subject string, topic string, pubsubName string, ...) *CloudEventsEnvelope
• type Metadata
• type NewMessage
• type PubSub
• type PublishRequest
• type SubscribeRequest

Constants

const (
	// DefaultCloudEventType is the default event type for an Dapr published event
	DefaultCloudEventType = "com.dapr.event.sent"
	// CloudEventsSpecVersion is the specversion used by Dapr for the cloud events implementation
	CloudEventsSpecVersion = "1.0"
	// ContentType is the Cloud Events HTTP content type
	ContentType = "application/cloudevents+json"
	// DefaultCloudEventSource is the default event source
	DefaultCloudEventSource = "Dapr"
	// DefaultCloudEventDataContentType is the default content-type for the data attribute
	DefaultCloudEventDataContentType = "text/plain"
)

Types

type AppResponse

type AppResponse struct {
	Status AppResponseStatus `json:"status"`
}

AppResponse is the object describing the response from user code after a pubsub event

type AppResponseStatus

type AppResponseStatus string

AppResponseStatus represents a status of a PubSub response.

const (
	// Success means the message is received and processed correctly.
	Success AppResponseStatus = "SUCCESS"
	// Retry means the message is received but could not be processed and must be retried.
	Retry AppResponseStatus = "RETRY"
	// Drop means the message is received but should not be processed.
	Drop AppResponseStatus = "DROP"
)

type CloudEventsEnvelope

type CloudEventsEnvelope struct {
	ID              string      `json:"id"`
	Source          string      `json:"source"`
	Type            string      `json:"type"`
	SpecVersion     string      `json:"specversion"`
	DataContentType string      `json:"datacontenttype"`
	Data            interface{} `json:"data"`
	Subject         string      `json:"subject"`
	Topic           string      `json:"topic"`
	PubsubName      string      `json:"pubsubname"`
}

CloudEventsEnvelope describes the Dapr implementation of the Cloud Events spec Spec details: https://github.com/cloudevents/spec/blob/master/spec.md

func NewCloudEventsEnvelope

func NewCloudEventsEnvelope(id, source, eventType, subject string, topic string, pubsubName string, dataContentType string, data []byte) *CloudEventsEnvelope

NewCloudEventsEnvelope returns CloudEventsEnvelope from data or a new one when data content was not

type Metadata

type Metadata struct {
	Properties map[string]string `json:"properties"`
}

Metadata represents a set of message-bus specific properties

type NewMessage

type NewMessage struct {
	Data     []byte            `json:"data"`
	Topic    string            `json:"topic"`
	Metadata map[string]string `json:"metadata"`
}

NewMessage is an event arriving from a message bus instance

type PubSub

type PubSub interface {
	Init(metadata Metadata) error
	Publish(req *PublishRequest) error
	Subscribe(req SubscribeRequest, handler func(msg *NewMessage) error) error
	Close() error
}

PubSub is the interface for message buses

type PublishRequest

type PublishRequest struct {
	Data       []byte            `json:"data"`
	PubsubName string            `json:"pubsubname"`
	Topic      string            `json:"topic"`
	Metadata   map[string]string `json:"metadata"`
}

PublishRequest is the request to publish a message

type SubscribeRequest

type SubscribeRequest struct {
	Topic    string            `json:"topic"`
	Metadata map[string]string `json:"metadata"`
}

SubscribeRequest is the request to subscribe to a topic