message

Documentation

Index

• func NewMessage(payload []byte, agent string, collective string, msgType string, ...) (inter.Message, error)
• func NewMessageFromRequest(req protocol.Request, replyto string, choria inter.Framework) (inter.Message, error)
• type Message
  • func (m *Message) Agent() string
  • func (m *Message) Base64Payload() string
  • func (m *Message) CacheTransport()
  • func (m *Message) CallerID() string
  • func (m *Message) Collective() string
  • func (m *Message) CustomTarget() string
  • func (m *Message) DiscoveredHosts() []string
  • func (m *Message) ExpectedMessageID() string
  • func (m *Message) Filter() *protocol.Filter
  • func (m *Message) IsCachedTransport() bool
  • func (m *Message) NotifyPublish()
  • func (m *Message) OnPublish(f func())
  • func (m *Message) Payload() []byte
  • func (m *Message) ProtocolVersion() protocol.ProtocolVersion
  • func (m *Message) ReplyTarget() string
  • func (m *Message) ReplyTo() string
  • func (m *Message) Request() inter.Message
  • func (m *Message) RequestID() string
  • func (m *Message) SenderID() string
  • func (m *Message) SetBase64Payload(payload []byte) error
  • func (m *Message) SetCollective(collective string) error
  • func (m *Message) SetCustomTarget(t string)
  • func (m *Message) SetDiscoveredHosts(hosts []string)
  • func (m *Message) SetExpectedMsgID(id string) error
  • func (m *Message) SetFilter(f *protocol.Filter)
  • func (m *Message) SetPayload(p []byte)
  • func (m *Message) SetProtocolVersion(version protocol.ProtocolVersion)
  • func (m *Message) SetReplyTo(replyTo string) error
  • func (m *Message) SetRequestID(id string)
  • func (m *Message) SetTTL(ttl int)
  • func (m *Message) SetType(msgType string) (err error)
  • func (m *Message) String() string
  • func (m *Message) TTL() int
  • func (m *Message) TimeStamp() time.Time
  • func (m *Message) Transport(ctx context.Context) (protocol.TransportMessage, error)
  • func (m *Message) Type() string
  • func (m *Message) UncachedReplyTransport() (protocol.TransportMessage, error)
  • func (m *Message) UncachedRequestTransport(ctx context.Context) (protocol.TransportMessage, error)
  • func (m *Message) UniqueTransport()
  • func (m *Message) Validate() (bool, error)
  • func (m *Message) ValidateTTL() bool

Functions

func NewMessage

func NewMessage(payload []byte, agent string, collective string, msgType string, request inter.Message, choria inter.Framework) (inter.Message, error)

NewMessage constructs a basic Message instance

func NewMessageFromRequest

func NewMessageFromRequest(req protocol.Request, replyto string, choria inter.Framework) (inter.Message, error)

NewMessageFromRequest constructs a Message based on a Request

Types

type Message

type Message struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Message represents a Choria message

func (*Message) Agent

func (m *Message) Agent() string

func (*Message) Base64Payload

func (m *Message) Base64Payload() string

Base64Payload retrieves the payload Base64 encoded

func (*Message) CacheTransport

func (m *Message) CacheTransport()

CacheTransport ensures that multiples calls to Transport() returns the same transport message

func (*Message) CallerID

func (m *Message) CallerID() string

func (*Message) Collective

func (m *Message) Collective() string

Collective retrieves the sub collective this message is targeting

func (*Message) CustomTarget

func (m *Message) CustomTarget() string

func (*Message) DiscoveredHosts

func (m *Message) DiscoveredHosts() []string

func (*Message) ExpectedMessageID

func (m *Message) ExpectedMessageID() string

ExpectedMessageID retrieves the expected message ID

func (*Message) Filter

func (m *Message) Filter() *protocol.Filter

func (*Message) IsCachedTransport

func (m *Message) IsCachedTransport() bool

IsCachedTransport determines if transport messages will be cached

func (*Message) NotifyPublish

func (m *Message) NotifyPublish()

NotifyPublish triggers the callback set using OnPublish() in a blocking fashion

func (*Message) OnPublish

func (m *Message) OnPublish(f func())

OnPublish sets a callback that should be called just before this message is published, it might be called several times for batched or direct messages. This will be called after the message have been signed - potentially by a remote signer etc

func (*Message) Payload

func (m *Message) Payload() []byte

func (*Message) ProtocolVersion

func (m *Message) ProtocolVersion() protocol.ProtocolVersion

func (*Message) ReplyTarget

func (m *Message) ReplyTarget() string

func (*Message) ReplyTo

func (m *Message) ReplyTo() string

ReplyTo retrieve the NATS reply target

func (*Message) Request

func (m *Message) Request() inter.Message

func (*Message) RequestID

func (m *Message) RequestID() string

func (*Message) SenderID

func (m *Message) SenderID() string

func (*Message) SetBase64Payload

func (m *Message) SetBase64Payload(payload []byte) error

SetBase64Payload sets the payload for the message, use it if the payload is Base64 encoded

func (*Message) SetCollective

func (m *Message) SetCollective(collective string) error

SetCollective sets the sub collective this message is targeting

func (*Message) SetCustomTarget

func (m *Message) SetCustomTarget(t string)

func (*Message) SetDiscoveredHosts

func (m *Message) SetDiscoveredHosts(hosts []string)

func (*Message) SetExpectedMsgID

func (m *Message) SetExpectedMsgID(id string) error

SetExpectedMsgID sets the Request ID that is expected from the reply data

func (*Message) SetFilter

func (m *Message) SetFilter(f *protocol.Filter)

func (*Message) SetPayload

func (m *Message) SetPayload(p []byte)

func (*Message) SetProtocolVersion

func (m *Message) SetProtocolVersion(version protocol.ProtocolVersion)

SetProtocolVersion sets the version of the protocol that will be used by Transport()

func (*Message) SetReplyTo

func (m *Message) SetReplyTo(replyTo string) error

SetReplyTo sets the NATS target where replies to this message should go

func (*Message) SetRequestID

func (m *Message) SetRequestID(id string)

func (*Message) SetTTL

func (m *Message) SetTTL(ttl int)

func (*Message) SetType

func (m *Message) SetType(msgType string) (err error)

SetType sets the message type. One message, request, direct_request, service_request or reply

func (*Message) String

func (m *Message) String() string

String creates a string representation of the message for logs etc

func (*Message) TTL

func (m *Message) TTL() int

func (*Message) TimeStamp

func (m *Message) TimeStamp() time.Time

func (*Message) Transport

func (m *Message) Transport(ctx context.Context) (protocol.TransportMessage, error)

Transport creates a TransportMessage for this Message

In the case of a reply Message made using NewMessage the Transport will have the same version as the request that made it. If you made the Message using some other way then look at choria.NewReplyTransportForMessage.

For requests you need to set the protocol version using SetProtocolVersion() before calling Transport

func (*Message) Type

func (m *Message) Type() string

Type retrieves the message type

func (*Message) UncachedReplyTransport

func (m *Message) UncachedReplyTransport() (protocol.TransportMessage, error)

func (*Message) UncachedRequestTransport

func (m *Message) UncachedRequestTransport(ctx context.Context) (protocol.TransportMessage, error)

func (*Message) UniqueTransport

func (m *Message) UniqueTransport()

UniqueTransport ensures that every call to Transport() produce a unique transport message

func (*Message) Validate

func (m *Message) Validate() (bool, error)

Validate tests the Message and makes sure its settings are sane

func (*Message) ValidateTTL

func (m *Message) ValidateTTL() bool

ValidateTTL validates the message age, true if the message should be allowed