io

type LosslessMarshaler ¶

type LosslessMarshaler struct{}

LosslessMarshaler marshals/unmarshals messages using gob. As opposed to other (un)marshalers in this package, all the attributes of the message (UUID, metadata, ...) are preserved. However, the result is not easily readable by humans or other marshalers.

func (m LosslessMarshaler) Marshal(topic string, msg *message.Message) ([]byte, error)

func (m LosslessMarshaler) Unmarshal(topic string, b []byte) (*message.Message, error)

type MarshalMessageFunc ¶

type MarshalMessageFunc func(topic string, msg *message.Message) ([]byte, error)

MarshalMessageFunc packages the message into a byte slice. The topic argument is there because some writers (i.e. loggers) might want to present the topic as part of their output.

type Publisher ¶

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

Publisher writes the messages to the underlying io.Writer. Its behaviour is highly customizable through the choice of the marshal function in config.

func NewPublisher(wc io.WriteCloser, 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 PublisherConfig struct {
	// MarshalFunc transforms the Watermill messages into raw bytes for transport.
	// Its behavior may be dependent on the topic.
	MarshalFunc MarshalMessageFunc
}

type Subscriber ¶

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

Subscriber reads bytes from its underlying io.Reader and interprets them as Watermill messages. It posts the messages on the output stream from Subscribe(). There are several ways in which Subscriber may interpret messages from the Reader, configurable by the unmarshal function in the config.

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

func (s *Subscriber) Close() error

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

type SubscriberConfig ¶

type SubscriberConfig struct {
	// BufferSize configures how many bytes will be read at a time from the Subscriber's Reader.
	// Each message will be treated as having at most BufferSize bytes.
	// If 0, Subscriber works in delimiter mode - it scans for messages delimited by the MessageDelimiter byte.
	BufferSize int
	// MessageDelimiter is the byte that is expected to separate messages if BufferSize is equal to 0.
	MessageDelimiter byte

	// PollInterval is the time between polling for new messages if the last read was empty. Defaults to time.Second.
	PollInterval time.Duration

	// UnmarshalFunc transforms the raw bytes into a Watermill message. Its behavior may be dependent on the topic.
	UnmarshalFunc UnmarshalMessageFunc
}

type UnmarshalMessageFunc ¶

type UnmarshalMessageFunc func(topic string, b []byte) (*message.Message, error)

UnmarshalMessageFunc restores the message from a byte slice. The topic argument is there to keep symmetry with MarshalMessageFunc, as some unmarshalers might restore the topic as well.