2016-12-14 15:41:48 +00:00
|
|
|
// Package broker is an interface used for asynchronous messaging
|
2021-10-02 19:55:07 +03:00
|
|
|
package broker // import "go.unistack.org/micro/v3/broker"
|
2015-04-26 19:33:35 +01:00
|
|
|
|
2020-11-18 16:50:41 +03:00
|
|
|
import (
|
|
|
|
"context"
|
2021-07-01 15:10:43 +03:00
|
|
|
"errors"
|
2020-11-18 16:50:41 +03:00
|
|
|
|
2021-10-02 19:55:07 +03:00
|
|
|
"go.unistack.org/micro/v3/metadata"
|
2020-11-18 16:50:41 +03:00
|
|
|
)
|
2020-10-16 09:38:57 +03:00
|
|
|
|
2021-07-22 22:53:44 +03:00
|
|
|
// DefaultBroker default memory broker
|
2021-04-27 08:32:47 +03:00
|
|
|
var DefaultBroker Broker = NewBroker()
|
2020-08-28 11:52:51 +03:00
|
|
|
|
2021-07-22 22:53:44 +03:00
|
|
|
var (
|
|
|
|
// ErrNotConnected returns when broker used but not connected yet
|
|
|
|
ErrNotConnected = errors.New("broker not connected")
|
|
|
|
// ErrDisconnected returns when broker disconnected
|
|
|
|
ErrDisconnected = errors.New("broker disconnected")
|
|
|
|
)
|
|
|
|
|
2016-01-30 21:18:57 +00:00
|
|
|
// Broker is an interface used for asynchronous messaging.
|
2015-04-26 19:33:35 +01:00
|
|
|
type Broker interface {
|
2021-07-22 22:53:44 +03:00
|
|
|
// Name returns broker instance name
|
2021-01-29 13:17:32 +03:00
|
|
|
Name() string
|
2021-07-22 22:53:44 +03:00
|
|
|
// Init initilize broker
|
|
|
|
Init(opts ...Option) error
|
|
|
|
// Options returns broker options
|
2015-12-31 18:11:46 +00:00
|
|
|
Options() Options
|
2021-07-22 22:53:44 +03:00
|
|
|
// Address return configured address
|
2019-07-10 19:58:30 +01:00
|
|
|
Address() string
|
2021-07-22 22:53:44 +03:00
|
|
|
// Connect connects to broker
|
|
|
|
Connect(ctx context.Context) error
|
|
|
|
// Disconnect disconnect from broker
|
|
|
|
Disconnect(ctx context.Context) error
|
|
|
|
// Publish message to broker topic
|
|
|
|
Publish(ctx context.Context, topic string, msg *Message, opts ...PublishOption) error
|
|
|
|
// Subscribe subscribes to topic message via handler
|
|
|
|
Subscribe(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Subscriber, error)
|
2021-07-24 16:16:18 +03:00
|
|
|
// BatchPublish messages to broker with multiple topics
|
|
|
|
BatchPublish(ctx context.Context, msgs []*Message, opts ...PublishOption) error
|
|
|
|
// BatchSubscribe subscribes to topic messages via handler
|
|
|
|
BatchSubscribe(ctx context.Context, topic string, h BatchHandler, opts ...SubscribeOption) (Subscriber, error)
|
2021-07-22 22:53:44 +03:00
|
|
|
// String type of broker
|
2015-12-19 21:56:14 +00:00
|
|
|
String() string
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|
|
|
|
|
2015-12-23 19:07:26 +00:00
|
|
|
// Handler is used to process messages via a subscription of a topic.
|
2020-08-27 11:18:02 +03:00
|
|
|
type Handler func(Event) error
|
|
|
|
|
2021-07-22 22:53:44 +03:00
|
|
|
// Events contains multiple events
|
|
|
|
type Events []Event
|
|
|
|
|
2022-03-22 14:54:43 +03:00
|
|
|
// Ack try to ack all events and return
|
2021-07-22 22:53:44 +03:00
|
|
|
func (evs Events) Ack() error {
|
|
|
|
var err error
|
|
|
|
for _, ev := range evs {
|
|
|
|
if err = ev.Ack(); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2022-03-22 14:54:43 +03:00
|
|
|
// SetError sets error on event
|
2021-07-22 22:53:44 +03:00
|
|
|
func (evs Events) SetError(err error) {
|
|
|
|
for _, ev := range evs {
|
|
|
|
ev.SetError(err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// BatchHandler is used to process messages in batches via a subscription of a topic.
|
|
|
|
type BatchHandler func(Events) error
|
|
|
|
|
2020-08-27 11:18:02 +03:00
|
|
|
// Event is given to a subscription handler for processing
|
|
|
|
type Event interface {
|
2021-07-22 22:53:44 +03:00
|
|
|
// Topic returns event topic
|
2020-08-27 11:18:02 +03:00
|
|
|
Topic() string
|
2021-07-22 22:53:44 +03:00
|
|
|
// Message returns broker message
|
2020-08-27 11:18:02 +03:00
|
|
|
Message() *Message
|
2021-07-22 22:53:44 +03:00
|
|
|
// Ack acknowledge message
|
2020-08-27 11:18:02 +03:00
|
|
|
Ack() error
|
2021-07-22 22:53:44 +03:00
|
|
|
// Error returns message error (like decoding errors or some other)
|
2020-08-27 11:18:02 +03:00
|
|
|
Error() error
|
2021-07-22 22:53:44 +03:00
|
|
|
// SetError set event processing error
|
|
|
|
SetError(err error)
|
2020-08-27 11:18:02 +03:00
|
|
|
}
|
2020-08-18 14:00:51 +01:00
|
|
|
|
2021-07-01 15:10:43 +03:00
|
|
|
// RawMessage is a raw encoded JSON value.
|
|
|
|
// It implements Marshaler and Unmarshaler and can be used to delay decoding or precompute a encoding.
|
|
|
|
type RawMessage []byte
|
|
|
|
|
|
|
|
// MarshalJSON returns m as the JSON encoding of m.
|
|
|
|
func (m *RawMessage) MarshalJSON() ([]byte, error) {
|
|
|
|
if m == nil {
|
2021-07-01 23:23:01 +03:00
|
|
|
return []byte("null"), nil
|
2021-07-01 15:10:43 +03:00
|
|
|
}
|
|
|
|
return *m, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// UnmarshalJSON sets *m to a copy of data.
|
|
|
|
func (m *RawMessage) UnmarshalJSON(data []byte) error {
|
|
|
|
if m == nil {
|
|
|
|
return errors.New("RawMessage UnmarshalJSON on nil pointer")
|
|
|
|
}
|
|
|
|
*m = append((*m)[0:0], data...)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-08-25 14:33:36 +03:00
|
|
|
// Message is used to transfer data
|
2015-04-26 19:33:35 +01:00
|
|
|
type Message struct {
|
2021-07-22 22:53:44 +03:00
|
|
|
// Header contains message metadata
|
|
|
|
Header metadata.Metadata
|
|
|
|
// Body contains message body
|
|
|
|
Body RawMessage
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|
|
|
|
|
2021-07-23 12:03:18 +03:00
|
|
|
// NewMessage create broker message with topic filled
|
|
|
|
func NewMessage(topic string) *Message {
|
|
|
|
m := &Message{Header: metadata.New(2)}
|
|
|
|
m.Header.Set(metadata.HeaderTopic, topic)
|
|
|
|
return m
|
|
|
|
}
|
|
|
|
|
2016-12-06 18:37:35 +00:00
|
|
|
// Subscriber is a convenience return type for the Subscribe method
|
2015-04-26 19:33:35 +01:00
|
|
|
type Subscriber interface {
|
2021-07-22 22:53:44 +03:00
|
|
|
// Options returns subscriber options
|
2015-12-31 18:11:46 +00:00
|
|
|
Options() SubscribeOptions
|
2021-07-22 22:53:44 +03:00
|
|
|
// Topic returns topic for subscription
|
2015-04-26 19:33:35 +01:00
|
|
|
Topic() string
|
2021-07-22 22:53:44 +03:00
|
|
|
// Unsubscribe from topic
|
|
|
|
Unsubscribe(ctx context.Context) error
|
2015-04-26 19:33:35 +01:00
|
|
|
}
|