Merge pull request #18 from Bit-Nation/develop

master re base

Makefile

@@ -5,6 +5,7 @@ deps:
 	go get github.com/whyrusleeping/gx-go
 	go get github.com/mattn/goveralls
 	go get -u github.com/kardianos/govendor
+	go get github.com/stretchr/testify
 install:
 	gx install
 	go get ./...

Readme.md

@@ -1,8 +1,8 @@
 # Panthalassa
 > Bitnation's backend - contains the mesh and some utils
 
-[![Coverage Status](https://coveralls.io/repos/github/Bit-Nation/panthalassa/badge.svg?branch=feature%2Faes)](https://coveralls.io/github/Bit-Nation/panthalassa?branch=develop)
-[![Build Status](https://semaphoreci.com/api/v1/florianlenz/panthalassa/branches/feature-aes/badge.svg)](https://semaphoreci.com/florianlenz/panthalassa)
+[![Build Status](https://semaphoreci.com/api/v1/florianlenz/panthalassa/branches/develop/badge.svg)](https://semaphoreci.com/florianlenz/panthalassa) (Develop)
+[![Build Status](https://semaphoreci.com/api/v1/florianlenz/panthalassa/branches/master/badge.svg)](https://semaphoreci.com/florianlenz/panthalassa) (Master)
 
 
 ## Development

api/device/README.md

@@ -0,0 +1,4 @@
+# Device Api
+> The device api is responsible for communication between the client and panthalassa
+
+## Supported Call's

api/device/api.go

@@ -0,0 +1,95 @@
+package device_api
+
+import (
+	"encoding/json"
+	"github.com/Bit-Nation/panthalassa/api/device/rpc"
+)
+
+type UpStream interface {
+	Send(data string)
+}
+
+type apiCall struct {
+	Type string `json:"type"`
+	Id   uint32 `json:"id"`
+	Data string `json:"data"`
+}
+
+func (c *apiCall) Marshal() ([]byte, error) {
+	return json.Marshal(c)
+}
+
+type Response struct {
+	Content string
+	Closer  chan error
+}
+
+type Api struct {
+	device UpStream
+	state  *State
+}
+
+func New(deviceInterface UpStream) *Api {
+
+	api := Api{
+		state:  newState(),
+		device: deviceInterface,
+	}
+
+	return &api
+}
+
+func (a *Api) Send(call rpc.JsonRPCCall) (chan Response, error) {
+
+	//Validate call
+	if err := call.Valid(); err != nil {
+		return nil, err
+	}
+
+	//Get call data
+	callContent, err := call.Data()
+	if err != nil {
+		return nil, err
+	}
+
+	//Create internal representation
+	c := apiCall{
+		Type: call.Type(),
+		Data: callContent,
+	}
+	respChan := make(chan Response, 1)
+	c.Id = a.state.Add(respChan)
+
+	//Marshal the call data
+	callData, err := c.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	//Send json rpc call to device
+	a.device.Send(string(callData))
+
+	return respChan, nil
+
+}
+
+func (a *Api) Receive(id uint32, data string) error {
+
+	//Get the response channel
+	resp, err := a.state.Cut(id)
+	if err != nil {
+		return err
+	}
+
+	//Closer
+	closer := make(chan error)
+
+	//Send response to response channel
+	resp <- Response{
+		Content: data,
+		Closer:  closer,
+	}
+
+	return <-closer
+
+}

api/device/api_test.go

@@ -0,0 +1,76 @@
+package device_api
+
+import (
+	"github.com/stretchr/testify/require"
+	"testing"
+
+	"encoding/json"
+)
+
+type upStreamTest struct {
+	send func(string)
+}
+
+func (u *upStreamTest) Send(data string) {
+	u.send(data)
+}
+
+type testRPCCall struct {
+	callType  string
+	data      string
+	dataError error
+	valid     func(data string) error
+}
+
+func (c *testRPCCall) Type() string {
+	return c.callType
+}
+func (c *testRPCCall) Data() (string, error) {
+	return c.data, c.dataError
+}
+func (c *testRPCCall) Valid() error {
+	return c.valid(c.data)
+}
+
+func Test(t *testing.T) {
+
+	//The api call we got from the send function
+	var receivedApiCall apiCall
+
+	//Create up stream test implementation
+	upStream := upStreamTest{
+		send: func(data string) {
+			var call apiCall
+			require.Nil(t, json.Unmarshal([]byte(data), &call))
+			receivedApiCall = call
+		},
+	}
+
+	//Create api
+	api := New(&upStream)
+
+	//Send api call with test data
+	respChan, err := api.Send(&testRPCCall{
+		callType:  "Test",
+		data:      `{"key": "value"}`,
+		dataError: nil,
+		valid: func(data string) error {
+			return nil
+		},
+	})
+	require.Nil(t, err)
+
+	//Waiting for the response form the api AND we then close it
+	go func() {
+		for {
+			res := <-respChan
+			require.Equal(t, "response", res.Content)
+			res.Closer <- nil
+		}
+	}()
+
+	//This assertion will be evaluated then "res.Closer <- nil" will be "done"
+	//"Receive" will only return after a value was send into the "res.Closer" channel
+	require.Nil(t, api.Receive(receivedApiCall.Id, "response"))
+
+}

api/device/logger.go

@@ -0,0 +1,9 @@
+package device_api
+
+import (
+	log "github.com/ipfs/go-log"
+)
+
+const LoggerSystem = "panthalassa_device_api"
+
+var Logger = log.Logger(LoggerSystem)

api/device/rpc/main.go

@@ -0,0 +1,12 @@
+package rpc
+
+type JsonRPCCall interface {
+	Type() string
+	Data() (string, error)
+	Valid() error
+}
+
+type JsonRPCResponse interface {
+	Valid()
+	Close()
+}

api/device/state.go

@@ -0,0 +1,55 @@
+package device_api
+
+import (
+	"errors"
+	"fmt"
+	"math/rand"
+	"sync"
+)
+
+type State struct {
+	requests map[uint32]chan Response
+	m        *sync.Mutex
+}
+
+func newState() *State {
+
+	return &State{
+		requests: make(map[uint32]chan Response),
+		m:        &sync.Mutex{},
+	}
+
+}
+
+func (s *State) Add(respChan chan Response) uint32 {
+
+	s.m.Lock()
+	var key uint32
+	//@todo we should have a backup break for the for loop
+	for {
+		key = rand.Uint32()
+		if _, exist := s.requests[key]; !exist {
+			break
+		}
+	}
+	s.requests[key] = respChan
+	s.m.Unlock()
+
+	return key
+
+}
+
+//Return's the channel an removes it from the state map
+func (s *State) Cut(index uint32) (chan Response, error) {
+
+	s.m.Lock()
+	respChan, exist := s.requests[index]
+	if !exist {
+		return nil, errors.New(fmt.Sprintf("a request channel for id (%d) does not exist", index))
+	}
+	delete(s.requests, index)
+	s.m.Unlock()
+
+	return respChan, nil
+
+}

api/device/state_test.go

@@ -0,0 +1,31 @@
+package device_api
+
+import (
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestState(t *testing.T) {
+
+	s := newState()
+
+	testChan := make(chan Response)
+
+	//Register test channel
+	id := s.Add(testChan)
+
+	//Check if successfully registered
+	s.m.Lock()
+	require.Equal(t, testChan, s.requests[id])
+	s.m.Unlock()
+
+	//Cut should remove the channel from the state and return it
+	registeredChan, err := s.Cut(id)
+	require.Nil(t, err)
+	require.Equal(t, testChan, registeredChan)
+
+	//Cutting a already received channel should as well result in an error
+	registeredChan, err = s.Cut(id)
+	require.EqualError(t, err, "a request channel for id (4039455774) does not exist")
+
+}