microservices

README

Microservices Package

This package gives you all the functionality to interact with ioFog both via Local API and WebSockets:

• send new message to ioFog with REST (PostMessage)
• fetch next unread messages from ioFog (GetNextMessages)
• fetch messages for time period and list of publishers (GetMessagesFromPublishersWithinTimeFrame)
• get config options (GetConfig)
• create IoMessage, encode(decode) to(from) raw bytes, encode(decode) data to(from) base64 string (IoMessage methods)
• connect to ioFog Control Channel via WebSocket (EstablishControlWsConnection)
• connect to ioFog Message Channel via WebSocket (EstablishMessageWsConnection) and publish new message via this channel (SendMessageViaSocket)

Code snippets:

Get sdk:

go get github.com/eclipse-iofog/iofog-sdk-go

Import package:

import (
	msvcs "github.com/eclipse-iofog/iofog-sdk-go/pkg/microservices"
)

Create IoFog client with default settings:

client, err := msvcs.NewDefaultIoFogClient()

Or specify host, port, ssl and container id explicitly:

client, err := msvcs.NewIoFogClient("IoFog", false, "containerId", 54321)

REST calls

Get list of next unread IoMessages:

messages, err := client.GetNextMessages()

Post new IoMessage to ioFog via REST call:

response, err := client.PostMessage(&msvcs.IoMessage{
	SequenceNumber:1,
	SequenceTotal:1,
	InfoType:"text",
	InfoFormat:"utf-8",
	ContentData: []byte("foo"),
	ContextData: []byte("bar"),
})

Get an array of IoMessages from specified publishers within given timeframe:

messages, err := client.GetMessagesFromPublishersWithinTimeFrame(&msvcs.MessagesQueryParameters{
	TimeFrameStart: 1234567890123,
	TimeFrameEnd: 1234567892123,
	Publishers: []string{"sefhuiw4984twefsdoiuhsdf", "d895y459rwdsifuhSDFKukuewf", "SESD984wtsdidsiusidsufgsdfkh"},
})

Get container's config:

config, err := client.GetConfig()

WebSocket calls

Establish connection with message ws. This call returns two channels, so you can listen to incoming messages and receipts:

dataChannel, receiptChannel := client.EstablishMessageWsConnection()
for {
	select {
	case msg := <-dataChannel:
		// msg is IoMessage received
	case r := <-receiptChannel:
		// r is response with ID and Timestamp
}

After establishing this connection you can send your own message to IoFog:

client.SendMessageViaSocket(&msvcs.IoMessage{
	Tag: "aaa",
	SequenceNumber: 127,
	ContentData: []byte("Here goes some test data"),
	ContextData: []byte("This one is test too"),
})

Establish connection with control ws and pass channel to listen to incoming config update signals:

confChannel := client.EstablishControlWsConnection()
for {
	select {
	case <-confChannel:
		// signal received
		// we can fetch new config now
		config, err := client.GetConfig()
}

Documentation

Index

• Constants
• func PrepareMessageForSendingViaSocket(msg *IoMessage) ([]byte, error)
• type IoFogClient
  • func NewDefaultIoFogClient() (*IoFogClient, error)
  • func NewIoFogClient(id string, ssl bool, host string, port int) (*IoFogClient, error)
  • func (client *IoFogClient) EstablishControlWsConnection(signalBufSize int) <-chan byte
  • func (client *IoFogClient) EstablishMessageWsConnection(msgBufSize, receiptBufSize int) (<-chan interface{}, <-chan interface{})
  • func (client *IoFogClient) GetConfig() (map[string]interface{}, error)
  • func (client *IoFogClient) GetConfigIntoStruct(config interface{}) error
  • func (client *IoFogClient) GetMessagesFromPublishersWithinTimeFrame(query *MessagesQueryParameters) (*TimeFrameMessages, error)
  • func (client *IoFogClient) GetNextMessages() ([]IoMessage, error)
  • func (client *IoFogClient) PostMessage(msg *IoMessage) (*PostMessageResponse, error)
  • func (client *IoFogClient) SendMessageViaSocket(msg *IoMessage) error
• type IoMessage
  • func GetMessageReceivedViaSocket(msgBytes []byte) (*IoMessage, error)
  • func (msg *IoMessage) DecodeBinary(data []byte) error
  • func (msg *IoMessage) EncodeBinary() ([]byte, error)
• type MessagesQueryParameters
• type PostMessageResponse
• type TimeFrameMessages

Constants

const (
	IOFOG = "iofog"
	ID    = "id"

	PORT_IOFOG   = 54321
	SELFNAME     = "SELFNAME"
	SSL          = "SSL"
	SSL_DEFAULT  = false
	HOST_DEFAULT = "127.0.0.1"

	URL_GET_CONFIG              = "/v2/config/get"
	URL_GET_NEXT_MESSAGES       = "/v2/messages/next"
	URL_GET_PUBLISHERS_MESSAGES = "/v2/messages/query"
	URL_POST_MESSAGE            = "/v2/messages/new"
	URL_GET_CONTROL_WS          = "/v2/control/socket/id/"
	URL_GET_MESSAGE_WS          = "/v2/message/socket/id/"

	APPLICATION_JSON = "application/json"
	HTTP             = "http"
	HTTPS            = "https"
	WS               = "ws"
	WSS              = "wss"

	CODE_ACK            = 0xB
	CODE_CONTROL_SIGNAL = 0xC
	CODE_MSG            = 0xD
	CODE_RECEIPT        = 0xE

	WS_ATTEMPT_LIMIT   = 10
	WS_CONNECT_TIMEOUT = time.Second

	DEFAULT_SIGNAL_BUFFER_SIZE  = 5
	DEFAULT_MESSAGE_BUFFER_SIZE = 200
	DEFAULT_RECEIPT_BUFFER_SIZE = 200
)

const IOMESSAGE_VERSION = 4

Functions

func PrepareMessageForSendingViaSocket

func PrepareMessageForSendingViaSocket(msg *IoMessage) ([]byte, error)

Types

type IoFogClient

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

func NewDefaultIoFogClient

func NewDefaultIoFogClient() (*IoFogClient, error)

func NewIoFogClient

func NewIoFogClient(id string, ssl bool, host string, port int) (*IoFogClient, error)

func (*IoFogClient) EstablishControlWsConnection

func (client *IoFogClient) EstablishControlWsConnection(signalBufSize int) <-chan byte

func (*IoFogClient) EstablishMessageWsConnection

func (client *IoFogClient) EstablishMessageWsConnection(msgBufSize, receiptBufSize int) (<-chan interface{}, <-chan interface{})

func (*IoFogClient) GetConfig

func (client *IoFogClient) GetConfig() (map[string]interface{}, error)

func (*IoFogClient) GetConfigIntoStruct

func (client *IoFogClient) GetConfigIntoStruct(config interface{}) error

func (*IoFogClient) GetMessagesFromPublishersWithinTimeFrame

func (client *IoFogClient) GetMessagesFromPublishersWithinTimeFrame(query *MessagesQueryParameters) (*TimeFrameMessages, error)

func (*IoFogClient) GetNextMessages

func (client *IoFogClient) GetNextMessages() ([]IoMessage, error)

func (*IoFogClient) PostMessage

func (client *IoFogClient) PostMessage(msg *IoMessage) (*PostMessageResponse, error)

func (*IoFogClient) SendMessageViaSocket

func (client *IoFogClient) SendMessageViaSocket(msg *IoMessage) error

type IoMessage

type IoMessage struct {
	ID               string `json:"id"`
	Tag              string `json:"tag"`
	GroupId          string `json:"groupid"`
	SequenceNumber   int    `json:"sequencenumber"`
	SequenceTotal    int    `json:"sequencetotal"`
	Priority         int    `json:"priority"`
	Timestamp        int64  `json:"timestamp"`
	Publisher        string `json:"publisher"`
	AuthID           string `json:"authid"`
	AuthGroup        string `json:"authgroup"`
	Version          int    `json:"version"`
	ChainPosition    int64  `json:"chainposition"`
	Hash             string `json:"hash"`
	PreviousHash     string `json:"previoushash"`
	Nonce            string `json:"nonce"`
	DifficultyTarget int    `json:"difficultytarget"`
	InfoType         string `json:"infotype"`
	InfoFormat       string `json:"infoformat"`
	ContextData      []byte `json:"contextdata"`
	ContentData      []byte `json:"contentdata"`
}

func GetMessageReceivedViaSocket

func GetMessageReceivedViaSocket(msgBytes []byte) (*IoMessage, error)

func (*IoMessage) DecodeBinary

func (msg *IoMessage) DecodeBinary(data []byte) error

func (*IoMessage) EncodeBinary

func (msg *IoMessage) EncodeBinary() ([]byte, error)

type MessagesQueryParameters

type MessagesQueryParameters struct {
	ID             string   `json:"id"`
	TimeFrameStart int64    `json:"timeframestart"`
	TimeFrameEnd   int64    `json:"timeframeend"`
	Publishers     []string `json:"publishers"`
}

type PostMessageResponse

type PostMessageResponse struct {
	ID        string `json:"id"`
	Timestamp int64  `json:"timestamp"`
}

type TimeFrameMessages

type TimeFrameMessages getNextMessagesResponse