zeebe

Documentation

Index

• Constants
• func MessageName(msg *message.Message) string
• func SetMessageName(name string, msg *message.Message)
• func SetTimeToLiveString(ttl int64, msg *message.Message)
• func StringToInt64(val string) int64
• func TimeToLive(msg *message.Message) int64
• type ConfigurationMap
  • func NewConfigurationMap(job entities.Job) (*ConfigurationMap, error)
  • func (c *ConfigurationMap) Get(key string) interface{}
  • func (c *ConfigurationMap) GetInt(key string) int64
  • func (c *ConfigurationMap) GetMap(key string) map[string]interface{}
  • func (c *ConfigurationMap) GetString(key string) string
  • func (c *ConfigurationMap) HasKey(key string) bool
• type DefaultMarshaler
  • func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*Message, error)
  • func (DefaultMarshaler) Unmarshal(job entities.Job) (*message.Message, error)
• type MarshalerUnmarshaler
• type Marshaller
• type Message
• type Publisher
  • func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, messages ...*message.Message) error
• type PublisherConfig
• type Subscriber
  • func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) Handler(ctx context.Context, messages chan *message.Message, ...) worker.JobHandler
  • func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)
• type SubscriberConfig
• type Unmarshaler

Constants

const (
	MessageNameMetadataKey = "message_name"
	TimeToLiveMetadataKey  = "message_ttl"
	DefaultTTl             = int64(5 * time.Minute / time.Millisecond)
)

Functions

func MessageName

func MessageName(msg *message.Message) string

func SetMessageName

func SetMessageName(name string, msg *message.Message)

func SetTimeToLiveString

func SetTimeToLiveString(ttl int64, msg *message.Message)

func StringToInt64

func StringToInt64(val string) int64

func TimeToLive

func TimeToLive(msg *message.Message) int64

Types

type ConfigurationMap

type ConfigurationMap struct {
	// contains filtered or unexported fields
}

func NewConfigurationMap

func NewConfigurationMap(job entities.Job) (*ConfigurationMap, error)

func (*ConfigurationMap) Get

func (c *ConfigurationMap) Get(key string) interface{}

func (*ConfigurationMap) GetInt

func (c *ConfigurationMap) GetInt(key string) int64

func (*ConfigurationMap) GetMap

func (c *ConfigurationMap) GetMap(key string) map[string]interface{}

func (*ConfigurationMap) GetString

func (c *ConfigurationMap) GetString(key string) string

func (*ConfigurationMap) HasKey

func (c *ConfigurationMap) HasKey(key string) bool

type DefaultMarshaler

type DefaultMarshaler struct{}

func (DefaultMarshaler) Marshal

func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*Message, error)

func (DefaultMarshaler) Unmarshal

func (DefaultMarshaler) Unmarshal(job entities.Job) (*message.Message, error)

type MarshalerUnmarshaler

type MarshalerUnmarshaler interface {
	Marshaller
	Unmarshaler
}

type Marshaller

type Marshaller interface {
	Marshal(topic string, msg *message.Message) (*Message, error)
}

Marshaller marshals Watermill's message to Zeebe's message.

type Message

type Message struct {
	// the name of the message
	Name string
	// the correlation key of the message
	CorrelationKey string
	// how long the message should be buffered on the broker, in milliseconds
	TimeToLive int64
	// the unique ID of the message; can be omitted. only useful to ensure only one message
	// with the given ID will ever be published (during its lifetime)
	MessageId string
	// the message variables as a JSON document; to be valid, the root of the document must be an
	// object, e.g. { "a": "foo" }. [ "foo" ] would not be valid.
	Variables string
}

type Publisher

type Publisher struct {
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, messages ...*message.Message) error

type PublisherConfig

type PublisherConfig struct {
	Marshaller Marshaller
	Client     zbc.Client
}

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

func NewSubscriber

func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) Handler

func (s *Subscriber) Handler(ctx context.Context, messages chan *message.Message, logFields watermill.LogFields) worker.JobHandler

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

type SubscriberConfig

type SubscriberConfig struct {
	Client      zbc.Client
	Worker      string
	Timeout     time.Duration
	Unmarshaler Unmarshaler
}

type Unmarshaler

type Unmarshaler interface {
	Unmarshal(entities.Job) (*message.Message, error)
}

Unmarshaler marshals Zeebe's message to Watermill's message.