ncdtransport

Documentation

Overview

Author: Bo Maryniuk Node Controller daemon

Index

• type CdtCallback
• type CdtTransport
  • func NewCdtTransport(topic string) *CdtTransport
  • func (cdt *CdtTransport) AddCallback(callback CdtCallback)
  • func (cdt *CdtTransport) OnReceive(body *nats.Msg)
  • func (cdt *CdtTransport) Topic() string
• type InternalEventMessage
  • func NewInternalEventMessage(data map[string]interface{}) *InternalEventMessage
  • func (iem *InternalEventMessage) FromBytes(data []byte) *InternalEventMessage
  • func (iem *InternalEventMessage) FromData(data map[string]interface{}) *InternalEventMessage
• type MqMessage
  • func NewMqMessage() *MqMessage
  • func (bm *MqMessage) FromBytes(data []byte) *MqMessage
  • func (bm *MqMessage) ToBytes() []byte
  • func (bm *MqMessage) ToJSON() string
• type MsgIdBuff
  • func NewMsgIdBuff() *MsgIdBuff
  • func (mb *MsgIdBuff) Channel(channel string) *MsgIdStor
• type MsgIdStor
  • func NewMsgIdStor() *MsgIdStor
  • func (mis *MsgIdStor) Discard(msgid string) bool
  • func (mis *MsgIdStor) Pop(msgid string)
  • func (mis *MsgIdStor) Present(msgid string) bool
  • func (mis *MsgIdStor) Push(msgid string)
• type NatsURL
• type NcdPubSub
  • func NewNcdPubSub() *NcdPubSub
  • func (ncd *NcdPubSub) AddNatsServerURL(host string, port int) *NcdPubSub
  • func (ncd *NcdPubSub) Disconnect()
  • func (ncd *NcdPubSub) GetPublisher() *nats.Conn
  • func (ncd *NcdPubSub) GetSubscriber() *nats.Conn
  • func (ncd *NcdPubSub) IsConnected() bool
  • func (ncd *NcdPubSub) Start()
• type PgEventCallback
• type PgEventListener
  • func NewPgEventListener() *PgEventListener
  • func (pel *PgEventListener) AddCallback(callback PgEventCallback) *PgEventListener
  • func (pel *PgEventListener) SetChannel(channel string) *PgEventListener
  • func (pel *PgEventListener) SetDBName(name string) *PgEventListener
  • func (pel *PgEventListener) SetHost(host string) *PgEventListener
  • func (pel *PgEventListener) SetPassword(password string) *PgEventListener
  • func (pel *PgEventListener) SetPort(port int) *PgEventListener
  • func (pel *PgEventListener) SetSSLMode(mode bool) *PgEventListener
  • func (pel *PgEventListener) SetUser(user string) *PgEventListener
  • func (pel *PgEventListener) Start()
  • func (pel *PgEventListener) StartProcess()
• type PgSQLTransport
  • func NewPgSQLTransport() *PgSQLTransport
• type Publisher
• type Subscriber

Types

type CdtCallback

type CdtCallback func(data map[string]interface{})

type CdtTransport

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

func NewCdtTransport

func NewCdtTransport(topic string) *CdtTransport

Constructor

func (*CdtTransport) AddCallback

func (cdt *CdtTransport) AddCallback(callback CdtCallback)

AddCallback adds an arbitrary callback, implementing transport.CdtCallback type.

func (*CdtTransport) OnReceive

func (cdt *CdtTransport) OnReceive(body *nats.Msg)

OnReceive is triggered by MQ when the message arrives

func (*CdtTransport) Topic

func (cdt *CdtTransport) Topic() string

Channel returns subscribed topic

type InternalEventMessage

type InternalEventMessage struct {
	Payload map[string]interface{}
	Topic   string
	Action  string
}

func NewInternalEventMessage

func NewInternalEventMessage(data map[string]interface{}) *InternalEventMessage

func (*InternalEventMessage) FromBytes

func (iem *InternalEventMessage) FromBytes(data []byte) *InternalEventMessage

Load self content from given bytes

func (*InternalEventMessage) FromData

func (iem *InternalEventMessage) FromData(data map[string]interface{}) *InternalEventMessage

type MqMessage

type MqMessage struct {
	Id      string
	Action  string
	Topic   string
	Payload interface{}
}

MqMessage can drive various topics.

"Action" is an arbitrary convention-based label. For databases it is "insert", "delete" or "update", which reflects the trigger. Other mesages could have other topics.

"Topic" is a mapping path of message interpreter and the topic itself. For databases it is "/db/<topic>". For example, to update or add or remove a channel, the topic is "/db/channel". Other topic might be management of a node, so it can be e.g. "/cfg" which would mean that the "Payload" is a configuration management nanostate and should be passed down to the nanostate interpreter for further processing.

func NewMqMessage

func NewMqMessage() *MqMessage

func (*MqMessage) FromBytes

func (bm *MqMessage) FromBytes(data []byte) *MqMessage

Load self content from given bytes

func (*MqMessage) ToBytes

func (bm *MqMessage) ToBytes() []byte

Serialise this object to bytes

func (*MqMessage) ToJSON

func (bm *MqMessage) ToJSON() string

Serialise this object to JSON string

type MsgIdBuff

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

func NewMsgIdBuff

func NewMsgIdBuff() *MsgIdBuff

func (*MsgIdBuff) Channel

func (mb *MsgIdBuff) Channel(channel string) *MsgIdStor

type MsgIdStor

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

func NewMsgIdStor

func NewMsgIdStor() *MsgIdStor

func (*MsgIdStor) Discard

func (mis *MsgIdStor) Discard(msgid string) bool

Discard combines Present and Pop together

func (*MsgIdStor) Pop

func (mis *MsgIdStor) Pop(msgid string)

func (*MsgIdStor) Present

func (mis *MsgIdStor) Present(msgid string) bool

Check if message ID is present in the stor

func (*MsgIdStor) Push

func (mis *MsgIdStor) Push(msgid string)

type NatsURL

type NatsURL struct {
	Scheme string
	Fqdn   string
	Port   int
}

type NcdPubSub

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

func NewNcdPubSub

func NewNcdPubSub() *NcdPubSub

func (*NcdPubSub) AddNatsServerURL

func (ncd *NcdPubSub) AddNatsServerURL(host string, port int) *NcdPubSub

AddNatsServerURL adds NATS server URL to the cluster of servers to connect

func (*NcdPubSub) Disconnect

func (ncd *NcdPubSub) Disconnect()

Disconnect from the cluster

func (*NcdPubSub) GetPublisher

func (ncd *NcdPubSub) GetPublisher() *nats.Conn

func (*NcdPubSub) GetSubscriber

func (ncd *NcdPubSub) GetSubscriber() *nats.Conn

func (*NcdPubSub) IsConnected

func (ncd *NcdPubSub) IsConnected() bool

IsConnected currently only indicates if the connection is initialised

func (*NcdPubSub) Start

func (ncd *NcdPubSub) Start()

Start starts the Node Controller

type PgEventCallback

type PgEventCallback func(payload interface{})

type PgEventListener

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

func NewPgEventListener

func NewPgEventListener() *PgEventListener

func (*PgEventListener) AddCallback

func (pel *PgEventListener) AddCallback(callback PgEventCallback) *PgEventListener

Add a callback on event

func (*PgEventListener) SetChannel

func (pel *PgEventListener) SetChannel(channel string) *PgEventListener

SetChannel sets a listening channel name

func (*PgEventListener) SetDBName

func (pel *PgEventListener) SetDBName(name string) *PgEventListener

SetDBName sets the name of the database. Default is "postgres"

func (*PgEventListener) SetHost

func (pel *PgEventListener) SetHost(host string) *PgEventListener

SetHost changes hostname from "localhost" to whatever else.

func (*PgEventListener) SetPassword

func (pel *PgEventListener) SetPassword(password string) *PgEventListener

SetPassword

func (*PgEventListener) SetPort

func (pel *PgEventListener) SetPort(port int) *PgEventListener

SetPort another than 5432 port

func (*PgEventListener) SetSSLMode

func (pel *PgEventListener) SetSSLMode(mode bool) *PgEventListener

SetSSLMode turns ON or OFF the SSL connection. Some setups doesn't support TLS connections.

func (*PgEventListener) SetUser

func (pel *PgEventListener) SetUser(user string) *PgEventListener

SetUser sets a user name. Default is current user

func (*PgEventListener) Start

func (pel *PgEventListener) Start()

Start in foreground

func (*PgEventListener) StartProcess

func (pel *PgEventListener) StartProcess()

Start in background

type PgSQLTransport

type PgSQLTransport struct{}

func NewPgSQLTransport

func NewPgSQLTransport() *PgSQLTransport

type Publisher

type Publisher interface {
	SetPublisher(nc *nats.Conn)
	Topic() string
}

Publisher interface

type Subscriber

type Subscriber interface {
	OnReceive(body *nats.Msg)
	Topic() string
}

Subscriber interface