brokervec

Documentation

Index

• type FileSub
  • func (fs *FileSub) AddFilterKey(key int)
  • func (fs *FileSub) Close()
  • func (fs *FileSub) CreateLog()
  • func (fs *FileSub) DisableNetworkFilter()
  • func (fs *FileSub) EnableNetworkFilter()
  • func (fs *FileSub) GetFilters() []int
  • func (fs *FileSub) GetName() (name string)
  • func (fs *FileSub) HasNetworkFilter() bool
  • func (fs *FileSub) Send(message Message)
• type FilterMessage
  • func (fm *FilterMessage) GetFilter() string
  • func (fm *FilterMessage) GetNonce() string
• type LocalMessage
  • func (lm *LocalMessage) GetMessage() string
  • func (lm *LocalMessage) GetNonce() string
  • func (lm *LocalMessage) GetTime() time.Time
• type LogMessage
  • func (logm *LogMessage) GetMessage() string
  • func (logm *LogMessage) GetNonce() string
  • func (logm *LogMessage) GetTime() time.Time
• type Message
• type NetworkMessage
  • func (nm *NetworkMessage) GetMessage() string
  • func (nm *NetworkMessage) GetNonce() string
  • func (nm *NetworkMessage) GetTime() time.Time
• type PubManager
  • func NewPubManager(vb *VectorBroker, listenPort string) *PubManager
  • func (pm *PubManager) AddLocalMsg(msg *LocalMessage, reply *string) error
  • func (pm *PubManager) AddMessage(msg Message, reply *string) (err error)
  • func (pm *PubManager) AddNetworkMsg(msg *NetworkMessage, reply *string) error
• type Publisher
• type SubManager
  • func NewSubManager(vb *VectorBroker, logfilename string, subport string) *SubManager
  • func (sm *SubManager) AddNetworkFilter(nonce string, reply *string) error
  • func (sm *SubManager) RemoveNetworkFilter(nonce string, reply *string) error
  • func (sm *SubManager) SendOldMessages(nonce string, reply *string) error
• type Subscriber
• type TCPPub
  • func (tp *TCPPub) GetName() (name string)
• type VectorBroker
  • func (vb *VectorBroker) AddMessage(message Message)
  • func (vb *VectorBroker) GetMessagesBefore(registerTime time.Time) ([]Message, int)
  • func (vb *VectorBroker) GetReadQueue() <-chan Message
  • func (vb *VectorBroker) Init(logfilename string, pubport string, subport string)
• type WSSub
  • func (ws *WSSub) AddFilterKey(key int)
  • func (ws *WSSub) Close()
  • func (ws *WSSub) DisableNetworkFilter()
  • func (ws *WSSub) EnableNetworkFilter()
  • func (ws *WSSub) GetFilters() []int
  • func (ws *WSSub) GetName() (name string)
  • func (ws *WSSub) HasNetworkFilter() bool
  • func (ws *WSSub) Send(message Message)

Types

type FileSub

type FileSub struct {
	Name string

	NetworkFilter bool
	FilterKeys    []int
	// contains filtered or unexported fields
}

Log File Subscriber

func (*FileSub) AddFilterKey

func (fs *FileSub) AddFilterKey(key int)

func (*FileSub) Close

func (fs *FileSub) Close()

func (*FileSub) CreateLog

func (fs *FileSub) CreateLog()

func (*FileSub) DisableNetworkFilter

func (fs *FileSub) DisableNetworkFilter()

func (*FileSub) EnableNetworkFilter

func (fs *FileSub) EnableNetworkFilter()

func (*FileSub) GetFilters

func (fs *FileSub) GetFilters() []int

func (*FileSub) GetName

func (fs *FileSub) GetName() (name string)

func (*FileSub) HasNetworkFilter

func (fs *FileSub) HasNetworkFilter() bool

func (*FileSub) Send

func (fs *FileSub) Send(message Message)

type FilterMessage

type FilterMessage struct {
	Regex string
	Nonce string
}

A FilterMessage is a message from a subscriber containing the filter it wishes to apply to a Message's field in the form of a regular expression.

func (*FilterMessage) GetFilter

func (fm *FilterMessage) GetFilter() string

func (*FilterMessage) GetNonce

func (fm *FilterMessage) GetNonce() string

type LocalMessage

type LocalMessage struct {
	Pid         string
	Vclock      []byte
	Message     string
	Nonce       string
	ReceiptTime time.Time
}

A LocalMessage is a message that was not sent over the network in the distributive system using GoVec.

func (*LocalMessage) GetMessage

func (lm *LocalMessage) GetMessage() string

func (*LocalMessage) GetNonce

func (lm *LocalMessage) GetNonce() string

func (*LocalMessage) GetTime

func (lm *LocalMessage) GetTime() time.Time

type LogMessage

type LogMessage struct {
	Message     string
	ReceiptTime time.Time
}

A LogMessage is an internal message for the broker's log file. It should not be sent to subscribers or received from publishers.

func (*LogMessage) GetMessage

func (logm *LogMessage) GetMessage() string

func (*LogMessage) GetNonce

func (logm *LogMessage) GetNonce() string

func (*LogMessage) GetTime

func (logm *LogMessage) GetTime() time.Time

type Message

type Message interface {
	GetMessage() string
	GetNonce() string
	GetTime() time.Time
}

Message abstraction to hold different types of message for use in the broker

type NetworkMessage

type NetworkMessage struct {
	Pid         string
	Vclock      []byte
	Message     string
	Nonce       string
	ReceiptTime time.Time
}

A NetworkMessage is a message that was sent over the network in the system using GoVec.

func (*NetworkMessage) GetMessage

func (nm *NetworkMessage) GetMessage() string

func (*NetworkMessage) GetNonce

func (nm *NetworkMessage) GetNonce() string

func (*NetworkMessage) GetTime

func (nm *NetworkMessage) GetTime() time.Time

type PubManager

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

func NewPubManager

func NewPubManager(vb *VectorBroker, listenPort string) *PubManager

Create a new PubManager that listens on the port listenPort and sets up a tcp connection in a new GoRoutine

func (*PubManager) AddLocalMsg

func (pm *PubManager) AddLocalMsg(msg *LocalMessage, reply *string) error

Adding local message to queue

func (*PubManager) AddMessage

func (pm *PubManager) AddMessage(msg Message, reply *string) (err error)

Add a mesage to the message queue and return a reply or an error if failed.

func (*PubManager) AddNetworkMsg

func (pm *PubManager) AddNetworkMsg(msg *NetworkMessage, reply *string) error

Adding network message to queue

type Publisher

type Publisher interface {
	GetName() string
}

type SubManager

type SubManager struct {
	Subscribers map[string]Subscriber

	// to subscribers map
	Filters map[int]string
	// contains filtered or unexported fields
}

func NewSubManager

func NewSubManager(vb *VectorBroker, logfilename string, subport string) *SubManager

Construct a new SubManager, initialize the network handling and broadcast routines

func (*SubManager) AddNetworkFilter

func (sm *SubManager) AddNetworkFilter(nonce string, reply *string) error

RPC call to add a network filter. Prevents a subscriber from receiving local messages.

func (*SubManager) RemoveNetworkFilter

func (sm *SubManager) RemoveNetworkFilter(nonce string, reply *string) error

RPC call to remove a network filter. Allows a subscriber to receive local and network messages.

func (*SubManager) SendOldMessages

func (sm *SubManager) SendOldMessages(nonce string, reply *string) error

RPC call to send all messages that were received before the subscriber joined to that subscriber.

type Subscriber

type Subscriber interface {
	Send(message Message)
	GetName() string
	Close()
	HasNetworkFilter() bool
	EnableNetworkFilter()
	DisableNetworkFilter()
	AddFilterKey(key int)
	GetFilters() []int
}

Subscriber interface and structs.

type TCPPub

type TCPPub struct {
	Name string
	Conn net.Conn
}

func (*TCPPub) GetName

func (tp *TCPPub) GetName() (name string)

type VectorBroker

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

Vector Messaging Server

• All licences like other licenses ...

  How to Use This Library

  Step 1: Create a Global Variable of type brokervec.VectorBroker and Initialize it like this =

  broker.Init(logpath, pubport, subport)

  Where:

• the logpath is the path and name of the log file you want created, or "" if no log file is wanted. E.g. "C:/temp/test" will result in the file "C:/temp/test-log.txt" being created.

• the pubport is the port you want to be open for publishers to send messages to the broker.

• the subport is the port you want to be open for subscribers to receive messages from the broker.

  Step 2: Setup your GoVec so that the realtime boolean is set to true and the correct brokeraddr and brokerpubport values are set in the Initialize method you intend to use.

  Step 3 (optional): Setup a Subscriber to connect to the broker via a WebSocket over the correct subport. For example, setup a web browser running JavaScript to connect and display messages as they are received.

  A simple standalone program can be found in runbroker.go which will setup a broker with the desired parameters. Please read the documentation in runbroker.go for more details.

  Tests can be run via GoVector/test/broker_test.go

func (*VectorBroker) AddMessage

func (vb *VectorBroker) AddMessage(message Message)

Adds a message to the broadcast queue and to the message archive for this session.

func (*VectorBroker) GetMessagesBefore

func (vb *VectorBroker) GetMessagesBefore(registerTime time.Time) ([]Message, int)

Return all messages that were received before registerTime and the number of messages returned.

func (*VectorBroker) GetReadQueue

func (vb *VectorBroker) GetReadQueue() <-chan Message

Returns a queue that messages can be read from but not written to.

func (*VectorBroker) Init

func (vb *VectorBroker) Init(logfilename string, pubport string, subport string)

initializing the server

type WSSub

type WSSub struct {
	Name           string
	Conn           *websocket.Conn
	TimeRegistered time.Time
	NetworkFilter  bool
	FilterKeys     []int
}

WebSocket Subscriber

func (*WSSub) AddFilterKey

func (ws *WSSub) AddFilterKey(key int)

func (*WSSub) Close

func (ws *WSSub) Close()

func (*WSSub) DisableNetworkFilter

func (ws *WSSub) DisableNetworkFilter()

func (*WSSub) EnableNetworkFilter

func (ws *WSSub) EnableNetworkFilter()

func (*WSSub) GetFilters

func (ws *WSSub) GetFilters() []int

func (*WSSub) GetName

func (ws *WSSub) GetName() (name string)

func (*WSSub) HasNetworkFilter

func (ws *WSSub) HasNetworkFilter() bool

func (*WSSub) Send

func (ws *WSSub) Send(message Message)

Sending message block to the client