package msg

import "github.com/zerofox-oss/go-msg"

Index ¶

• Variables
• func CloneBody(m *Message) (io.Reader, error)
• func DumpBody(m *Message) ([]byte, error)
• type Attributes
• func (a Attributes) Get(key string) string
• func (a Attributes) Set(key, value string)
• type Message
• func WithBody(parent *Message, r io.Reader) *Message
• type MessageWriter
• type Receiver
• type ReceiverFunc
• func (f ReceiverFunc) Receive(ctx context.Context, m *Message) error
• type Server
• type Topic
• type TopicFunc
• func (t TopicFunc) NewWriter(ctx context.Context) MessageWriter

Package Files ¶

msg.go

Variables ¶

var ErrClosedMessageWriter = errors.New("msg: MessageWriter closed")

ErrClosedMessageWriter is the error used for write or close operations on a closed MessageWriter.

var ErrServerClosed = errors.New("msg: server closed")

ErrServerClosed represents a completed Shutdown

func CloneBody ¶ Uses

func CloneBody(m *Message) (io.Reader, error)

CloneBody returns a reader with the same contents and m.Body. m.Body is reset allowing it to be read from later.

func DumpBody ¶ Uses

func DumpBody(m *Message) ([]byte, error)

DumpBody returns the contents of m.Body while resetting m.Body allowing it to be read from later.

type Attributes ¶ Uses

type Attributes map[string][]string

Attributes represent the key-value metadata for a Message.

func (Attributes) Get ¶ Uses

func (a Attributes) Get(key string) string

Get returns the first value associated with the given key. It is case insensitive; CanonicalMIME is used to cannonicalize the provided key. If there are no values associated with the key, Get returns "". To access multiple values of a key, or to use non-canonical keys, access the map directly.

func (Attributes) Set ¶ Uses

func (a Attributes) Set(key, value string)

Set sets the header entries associated with key the single element element value. It replaces any existing values associated with key.

Note: MIMEHeader automatically capitalizes the first letter of the key.

type Message ¶ Uses

type Message struct {
    Attributes Attributes
    Body       io.Reader
}

A Message represents a discrete message in a messaging system.

func WithBody ¶ Uses

func WithBody(parent *Message, r io.Reader) *Message

WithBody creates a new Message with the given io.Reader as a Body containing the parent's Attributes.

p := &Message{
	Attributes: Attributes{},
	Body: strings.NewReader("hello world"),
}
p.Attributes.Set("hello", "world")
m := WithBody(p, strings.NewReader("world hello")

type MessageWriter ¶ Uses

type MessageWriter interface {
    io.Writer
    // Close should be called to signify the completion of a Write. Attributes
    // that represent a transform applied to a message should also be written
    // at this time.
    //
    // Close should forward a message to another MessageWriter or persist
    // to the messaging system.
    //
    // Once Close has been called, all subsequent Write and Close calls will result
    // in an ErrClosedMessageWriter error.
    io.Closer
    Attributes() *Attributes
}

A MessageWriter interface is used to write a message to an underlying data stream.

type Receiver ¶ Uses

type Receiver interface {
    Receive(context.Context, *Message) error
}

A Receiver processes a Message.

Receive should process the message and then return. Returning signals that the message has been processed. It is not valid to read from the Message.Body after or concurrently with the completion of the Receive call.

If Receive returns an error, the server (the caller of Receive) assumes the message has not been processed and, depending on the underlying pub/sub system, the message should be put back on the message queue.

type ReceiverFunc ¶ Uses

type ReceiverFunc func(context.Context, *Message) error

The ReceiverFunc is an adapter to allow the use of ordinary functions as a Receiver. ReceiverFunc(f) is a Receiver that calls f.

func (ReceiverFunc) Receive ¶ Uses

func (f ReceiverFunc) Receive(ctx context.Context, m *Message) error

Receive calls f(ctx,m)

type Server ¶ Uses

type Server interface {
    // Serve is a blocking function that gets data from an input stream,
    // creates a message, and calls Receive() on the provided receiver
    // with the Message and a Context derived from context.Background().
    // For example:
    //
    // 		parentctx = context.WithCancel(context.Background())
    // 		err := r.Receive(parentctx, m)
    //
    // Serve will return ErrServerClosed after Shutdown completes. Additional
    // error types should be considered to represent error conditions unique
    // to the implementation of a specific technology.
    //
    // Serve() should continue to listen until Shutdown is called on
    // the Server.
    Serve(Receiver) error

    // Shutdown gracefully shuts down the Server by letting any messages in
    // flight finish processing.  If the provided context cancels before
    // shutdown is complete, the Context's error is returned.
    Shutdown(context.Context) error
}

A Server serves messages to a receiver.

type Topic ¶ Uses

type Topic interface {
    // NewWriter returns a new MessageWriter
    NewWriter(context.Context) MessageWriter
}

Topic is a generic interface where messages are sent in a messaging system.

Multiple goroutines may invoke method on a Topic simultaneously.

type TopicFunc ¶ Uses

type TopicFunc func(context.Context) MessageWriter

The TopicFunc is an adapter to allow the use of ordinary functions as a Topic. TopicFunc(f) is a Topic that calls f.

func (TopicFunc) NewWriter ¶ Uses

func (t TopicFunc) NewWriter(ctx context.Context) MessageWriter

NewWriter calls f(ctx,m)

Directories ¶