messenger

Documentation

Overview

* Copyright (c) 2016 The ConnectorDB Contributors Licensed under the MIT license. *

* Copyright (c) 2016 The ConnectorDB Contributors Licensed under the MIT license. *

Index

• Constants
• type Message
• type Messenger
  • func ConnectMessenger(opt *nats.Options, err error) (*Messenger, error)
  • func (m *Messenger) Close()
  • func (m *Messenger) Flush()
  • func (m *Messenger) Publish(routing string, msg Message) error
  • func (m *Messenger) Subscribe(routing string, chn chan Message) (*nats.Subscription, error)
• type MsgPackEncoder
  • func (mpe MsgPackEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error)
  • func (mpe MsgPackEncoder) Encode(subject string, v interface{}) ([]byte, error)

Constants

const MessageEncoding string = "msgpack"

MessageEncoding is the encoding used for messages

Types

type Message

type Message struct {
	Stream    string                    `json:"stream" msgpack:"s,omitempty"`
	Transform string                    `json:"transform,omitempty" msgpack:"t,omitempty"`
	Data      datastream.DatapointArray `json:"data" msgpack:"d,omitempty"`
}

Message is what is sent over NATS

type Messenger

type Messenger struct {
	SendConn  *nats.Conn        //The NATS connection
	SendEconn *nats.EncodedConn //The Encoded conn, ie, a data message
	RecvConn  *nats.Conn
	RecvEconn *nats.EncodedConn
}

Messenger holds an open connection to the gnatsd daemon

func ConnectMessenger

func ConnectMessenger(opt *nats.Options, err error) (*Messenger, error)

ConnectMessenger initializes a connection with the gnatsd messenger. Allows daisy-chaining errors

func (*Messenger) Close

func (m *Messenger) Close()

Close shuts down a Messenger

func (*Messenger) Flush

func (m *Messenger) Flush()

Flush makes sure all commands are acknowledged by the server

func (*Messenger) Publish

func (m *Messenger) Publish(routing string, msg Message) error

Publish sends the given message over the connection

func (*Messenger) Subscribe

func (m *Messenger) Subscribe(routing string, chn chan Message) (*nats.Subscription, error)

Subscribe creates a subscription for the given routing string. The routing string is of the format:

[user]/[device]/[stream]/[substream//]

In order to skip something, you can use wildcards, and to skip "the rest" you can use ">" (this is literally the gnatsd routing) An example of subscribing to all posts by sender user user1:

msgr.Subscribe("user1/>",chn)

An example of subscribing to everything is:

msgr.Subscribe(">",chn)

Subscribing to a stream is: msgr.Subscribe("user/device/stream")

type MsgPackEncoder

type MsgPackEncoder struct {
}

The MsgPackEncoder encodes the data using msgpack (more wire-efficient than json)

func (MsgPackEncoder) Decode

func (mpe MsgPackEncoder) Decode(subject string, data []byte, vPtr interface{}) (err error)

Decode is to fit the interface

func (MsgPackEncoder) Encode

func (mpe MsgPackEncoder) Encode(subject string, v interface{}) ([]byte, error)

Encode is to fit the interface