louis/oktopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #190 from OktopUSP/dev

Browse Source 
Multi Protocol Architecture + Websockets MTP | Closes #100
This commit is contained in:
Leandro Antônio Farias Machado 2024-02-22 08:10:51 -03:00 committed by GitHub
commit 0a6b626164
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
35 changed files with 9563 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
README.md
View File

@ -194,8 +194,8 @@ Currently, telecommunications giants and startups, publishing new software daily
<h4>Developer:</h4>
Run app using Docker:
<pre>
leandro@leandro-laptop:~$ cd oktopus/devops
leandro@leandro-laptop:~/oktopus/devops$ docker compose up
user@user-laptop:~$ cd oktopus/devops
user@user-laptop:~/oktopus/devops$ docker compose up
</pre>
</li>
<li>
@ -206,39 +206,46 @@ Basic manual compilation and run:
</li>
<li>
Mqtt broker:
<pre>leandro@leandro-laptop:~$ cd oktopus/backend/services/mochi/ && go run cmd/main.go -redis "127.0.0.1:6379"</pre>
<pre>user@user-laptop:~$ cd oktopus/backend/services/mochi/ && go run cmd/main.go -redis "127.0.0.1:6379"</pre>
</li>
<li>
TR-369 controller:
<pre>
leandro@leandro-laptop:~$ cd oktopus/backend/services/controller/ && go run cmd/oktopus/main.go -u root -P root -mongo "mongodb://127.0.0.1:27017"</pre>
user@user-laptop:~$ cd oktopus/backend/services/controller/ && go run cmd/oktopus/main.go -u root -P root -mongo "mongodb://127.0.0.1:27017"</pre>
</li>
<li>
Socketio server:
<pre>
leandro@leandro-laptop:~$ cd oktopus/backend/services/socketio && npm i && npm start</pre>
user@user-laptop:~$ cd oktopus/backend/services/socketio && npm i && npm start</pre>
</li>
<li>
Websockets server:
<pre>
user@user-laptop:~$ cd oktopus/backend/services/ws && go run cmd/main.go</pre>
</li>
<li>
Frontend:
<pre>
leandro@leandro-laptop:~$ cd oktopus/frontend && npm i && npm run dev</pre>
user@user-laptop:~$ cd oktopus/frontend && npm i && npm run dev</pre>
</li>
</ul>
</li>
<li>
<h4>Device test agent (obuspa):</h4>
<p>You might follow the instructions to build <a href="https://github.com/BroadbandForum/obuspa">obuspa</a> at their repo QUICK_START_GUIDE.md and remember to define 'INCLUDE_PROGRAMMATIC_FACTORY_RESET' at <a href="https://github.com/BroadbandForum/obuspa/blob/master/src/vendor/vendor_defs.h">'vendor_defs.h'</a> inside src/vendor folder, also advice you to build it with make tool.</p>
<p><b>
P.S: This guide in here is too simple, and at the same time hard, because you have to compile obuspa C code yourself, for an easier, more complete and real-world simulation of devices you might use <a href="https://github.com/OktopUSP/agent-sim">Oktopus TR-369 Agent Simulator</a>.
</b></p>
<p>Follow the instructions to build <a href="https://github.com/BroadbandForum/obuspa">obuspa</a> at their repo QUICK_START_GUIDE.md and remember to define 'INCLUDE_PROGRAMMATIC_FACTORY_RESET' at <a href="https://github.com/BroadbandForum/obuspa/blob/master/src/vendor/vendor_defs.h">'vendor_defs.h'</a> inside src/vendor folder, also advice you to build it with make tool.</p>
<p>You can customize <a href="https://github.com/OktopUSP/oktopus/tree/main/agent/oktopus-mqtt-obuspa.txt">'oktopus-mqtt-obuspa.txt'</a> accordingly to your needs, there you can change broker address, client password, and etc. '-i' option defines the interface your usp packets will go through, in this case it's localhost, also keep in mind that after you run obuspa if you made a change to 'oktopus-mqtt-obuspa.txt' and want it to take effect you must delete '/usr/local/var/obuspa/usp.db' file before running the agent again (docs provide you other methods to do it too).'-r' option defines the config file you want the agent to use. </p>
<p>Run agent:</p>
<pre>leandro@leandro-laptop:~/oktopus$ obuspa -p -v 4 -r agent/oktopus-mqtt-obuspa.txt -i lo</pre>
<p>Run MQTT agent:</p>
<pre>user@user-laptop:~/oktopus$ obuspa -p -v 4 -r agent/oktopus-mqtt-obuspa.txt -i lo</pre>
<p>Run Websockets agent:</p>
<pre>user@user-laptop:~/oktopus$ obuspa -p -v 4 -r agent/oktopus-websockets-obuspa.txt -i lo</pre>
<p>Obuspa has a lot of info and docs at their repo, which shows you many options to set your environment and customize agent to your embedded project. This basic agent has the purpose to test oktopus connection and shows an idea of how it's like with true devices, although with those, you're able to explore much more parameters and execute lots of configurations. Obuspa has default parameters and mocked info:</p>
<img src="https://github.com/OktopUSP/oktopus/assets/83298718/4599d566-eada-4313-8ae1-31dae82391de"/>
<img src="https://github.com/OktopUSP/oktopus/assets/83298718/501b4ccd-6147-4957-9096-695134e34b5e"/>
</li>
<p>
In case you want a more complete and real-world simulation of devices you might use <a href="https://github.com/OktopUSP/agent-sim">Oktopus TR-369 Agent Simulator</a>.
</p>
</ul>
--------------------------------------------------------------------------------------------------------------------------------------------------------

59
agent/oktopus-websockets-obuspa.txt Normal file
View File

@ -0,0 +1,59 @@
##########################################################################################################
#
# This file contains a factory reset database in text format
#
# If no USP database exists when OB-USP-AGENT starts, then OB-USP-AGENT will create a database containing
# the parameters specified in a text file located by the '-r' option.
# Example:
# obuspa -p -v 4 -r factory_reset_example.txt
#
# Each line of this file contains either a comment (denoted by '#' at the start of the line)
# or a USP data model parameter and its factory reset value.
# The parameter and value are separated by whitespace.
# The value may optionally be enclosed in speech marks "" (this is the only way to specify an empty string)
#
##########################################################################################################
#
# The following parameters will definitely need modifying
#
Device.LocalAgent.EndpointID "oktopus-0-ws"
# Controller's websocket server (for agent initiated sessions)
Device.LocalAgent.Controller.1.EndpointID "oktopusController"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.Host "host.docker.internal"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.Port "8080"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.Path "ws/agent"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.EnableEncryption "false"
# Agent's websocket server (for controller initiated sessions)
Device.LocalAgent.MTP.1.WebSocket.Port "8080"
Device.LocalAgent.MTP.1.WebSocket.Path "/usp"
Device.LocalAgent.MTP.1.WebSocket.EnableEncryption "false"
#
# The following parameters may be modified
#
Device.LocalAgent.MTP.1.Alias "cpe-1"
Device.LocalAgent.MTP.1.Enable "true"
Device.LocalAgent.MTP.1.Protocol "WebSocket"
Device.LocalAgent.MTP.1.WebSocket.KeepAliveInterval "30"
Device.LocalAgent.Controller.1.Alias "cpe-1"
Device.LocalAgent.Controller.1.Enable "true"
Device.LocalAgent.Controller.1.AssignedRole "Device.LocalAgent.ControllerTrust.Role.1"
Device.LocalAgent.Controller.1.PeriodicNotifInterval "86400"
Device.LocalAgent.Controller.1.PeriodicNotifTime "0001-01-01T00:00:00Z"
Device.LocalAgent.Controller.1.USPNotifRetryMinimumWaitInterval "5"
Device.LocalAgent.Controller.1.USPNotifRetryIntervalMultiplier "2000"
Device.LocalAgent.Controller.1.ControllerCode ""
Device.LocalAgent.Controller.1.MTP.1.Alias "oktopus0"
Device.LocalAgent.Controller.1.MTP.1.Enable "true"
Device.LocalAgent.Controller.1.MTP.1.Protocol "WebSocket"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.KeepAliveInterval "30"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.SessionRetryMinimumWaitInterval "5"
Device.LocalAgent.Controller.1.MTP.1.WebSocket.SessionRetryIntervalMultiplier "2000"
Internal.Reboot.Cause "LocalFactoryReset"

30
backend/services/controller/.env
View File

@ -1,5 +1,8 @@
SECRET_API_KEY="secretkey" # !IMPORTANT: Change this to your own secret key, and don't share.
MONGO_URI=""
# --------------------------------- database --------------------------------- #
MONGO_URI="" # example: mongodb://localhost:27017
# ---------------------------------------------------------------------------- #
# ----------------------------------- mqtt ----------------------------------- #
DEVICES_STATUS_TOPIC=""
DEVICE_PUB_TOPIC=""
BROKER_ADDR=""
@ -9,10 +12,29 @@ BROKER_USERNAME=""
BROKER_PASSWORD=""
BROKER_CLIENTID=""
BROKER_QOS=""
MQTT_DISABLE=""
# ---------------------------------------------------------------------------- #
# --------------------------------- api rest --------------------------------- #
SECRET_API_KEY="secretkey" # !IMPORTANT: Change this to your own secret key, and don't share.
REST_API_PORT=""
REST_API_CORS="" # addresses must be separated by commas example: "http://localhost:3000,http://myapp.com"
# ---------------------------------------------------------------------------- #
# ----------------------------------- stomp ---------------------------------- #
STOMP_ADDR="" # example: localhost:61613
STOMP_USERNAME=""
STOMP_PASSWORD=""
MQTT_DISABLE=""
STOMP_DISABLE=""
STOMP_DISABLE=""
# ---------------------------------------------------------------------------- #
# -------------------------------- websockets -------------------------------- #
WS_ADDR=""
WS_PORT=""
WS_TOKEN=""
WS_TLS=""
WS_AUTH=""
WS_ROUTE=""
WS_DISABLE=""
WS_SKIP_VERIFY=""
# ---------------------------------------------------------------------------- #

67
backend/services/controller/cmd/oktopus/main.go
View File

@ -19,8 +19,10 @@ import (
"github.com/leandrofars/oktopus/internal/mtp"
"github.com/leandrofars/oktopus/internal/stomp"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/ws"
)
// TODO: refact where this version number comes from
const VERSION = "0.0.1"
func main() {
@ -41,8 +43,9 @@ func main() {
}
// Locks app running until it receives a stop command as Ctrl+C.
signal.Notify(done, syscall.SIGINT, syscall.SIGTERM)
signal.Notify(done, syscall.SIGINT)
//TODO: refact app confiurations and env loading to another package
log.SetFlags(log.LstdFlags | log.Lshortfile)
/*
@ -68,6 +71,14 @@ func main() {
flStompAddr := flag.String("stomp", lookupEnvOrString("STOMP_ADDR", "127.0.0.1:61613"), "Stomp broker address")
flStompUser := flag.String("stomp_user", lookupEnvOrString("STOMP_USERNAME", ""), "Stomp broker username")
flStompPasswd := flag.String("stomp_passwd", lookupEnvOrString("STOMP_PASSWORD", ""), "Stomp broker password")
flWsToken := flag.String("ws_token", lookupEnvOrString("WS_TOKEN", ""), "Websocket token")
flWsAuth := flag.Bool("ws_auth", lookupEnvOrBool("WS_AUTH", true), "Websocket auth enable or not")
flWsAddr := flag.String("ws_addr", lookupEnvOrString("WS_ADDR", "localhost"), "Websocket server address")
flWsPort := flag.String("ws_port", lookupEnvOrString("WS_PORT", "8080"), "Websocket server port")
flWsRoute := flag.String("ws_route", lookupEnvOrString("WS_ROUTE", "/ws/controller"), "Websocket server route")
flWsTls := flag.Bool("ws_tls", lookupEnvOrBool("WS_TLS", false), "Websocket server tls")
flWsSkipVerify := flag.Bool("ws_skip_verify", lookupEnvOrBool("WS_SKIP_VERIFY", false), "Websocket skip tls certificate verify")
flDisableWs := flag.Bool("ws_disable", lookupEnvOrBool("WS_DISABLE", false), "Disable WS MTP")
flDisableStomp := flag.Bool("stomp_disable", lookupEnvOrBool("STOMP_DISABLE", false), "Disable STOMP MTP")
flDisableMqtt := flag.Bool("mqtt_disable", lookupEnvOrBool("MQTT_DISABLE", false), "Disable MQTT MTP")
flHelp := flag.Bool("help", false, "Help")
@ -87,21 +98,31 @@ func main() {
apiMsgQueue := make(map[string](chan usp_msg.Msg))
var m sync.Mutex
//TODO: refact mtps initialization through main.go
/*
If you want to use another message protocol just make it implement Broker interface.
*/
log.Println("Start MTP protocols: MQTT | Websockets | STOMP")
if *flDisableMqtt && *flDisableStomp {
if *flDisableMqtt && *flDisableStomp && *flDisableWs {
log.Println("ERROR: you have to enable at least one MTP")
os.Exit(0)
}
wg := new(sync.WaitGroup)
wg.Add(2)
wg.Add(3) // Three wait groups (mqtt, stomp, ws)
/* ------------------------------ MTPs clients ------------------------------ */
var stompClient stomp.Stomp
var mqttClient mqtt.Mqtt
var wsClient ws.Ws
/* -------------------------------------------------------------------------- */
/* ------------------------ MTPs disconnect channels ------------------------ */
var mqttDone chan os.Signal
var wsDone chan os.Signal
var stompDone chan os.Signal
/* -------------------------------------------------------------------------- */
go func() {
mqttClient = mqtt.Mqtt{
@ -120,9 +141,11 @@ func main() {
QMutex: &m,
}
mqttDone = make(chan os.Signal, 1)
if !*flDisableMqtt {
// MQTT will try connect to broker forever
go mtp.MtpService(&mqttClient, done, wg)
go mtp.MtpService(&mqttClient, mqttDone, wg)
} else {
wg.Done()
}
@ -135,9 +158,35 @@ func main() {
Password: *flStompPasswd,
}
stompDone = make(chan os.Signal, 1)
if !*flDisableStomp {
// STOMP will try to connect for a bunch of times and then exit
go mtp.MtpService(&stompClient, done, wg)
go mtp.MtpService(&stompClient, stompDone, wg)
} else {
wg.Done()
}
}()
go func() {
wsClient = ws.Ws{
Addr: *flWsAddr,
Port: *flWsPort,
Token: *flWsToken,
Route: *flWsRoute,
Auth: *flWsAuth,
TLS: *flWsTls,
InsecureSkipVerify: *flWsSkipVerify,
DB: database,
Ctx: ctx,
MsgQueue: apiMsgQueue,
QMutex: &m,
}
wsDone = make(chan os.Signal, 1)
if !*flDisableWs {
go mtp.MtpService(&wsClient, wsDone, wg)
} else {
wg.Done()
}
@ -145,16 +194,22 @@ func main() {
wg.Wait()
a := api.NewApi(*flApiPort, database, &mqttClient, apiMsgQueue, &m)
a := api.NewApi(*flApiPort, database, &mqttClient, apiMsgQueue, &m, wsClient) //TODO: websockets instance
api.StartApi(a)
<-done
cancel()
// send done signal to all MTPs
wsDone <- os.Interrupt
mqttDone <- os.Interrupt
stompDone <- os.Interrupt
log.Println("(⌐■_■) Oktopus is out!")
}
//TODO: refact functions below to another package
func lookupEnvOrString(key string, defaultVal string) string {
if val, _ := os.LookupEnv(key); val != "" {
return val

3
backend/services/controller/go.mod
View File

@ -7,7 +7,6 @@ require (
github.com/go-stomp/stomp v2.1.4+incompatible
github.com/golang-jwt/jwt/v5 v5.2.0
github.com/google/uuid v1.3.0
github.com/googollee/go-socket.io v1.7.0
github.com/gorilla/mux v1.8.0
github.com/joho/godotenv v1.5.1
github.com/rs/cors v1.9.0
@ -19,9 +18,7 @@ require (
)
require (
github.com/gofrs/uuid v4.0.0+incompatible // indirect
github.com/golang/snappy v0.0.1 // indirect
github.com/gomodule/redigo v1.8.4 // indirect
github.com/gorilla/websocket v1.4.2 // indirect
github.com/klauspost/compress v1.13.6 // indirect
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect

9
backend/services/controller/go.sum
View File

@ -5,22 +5,16 @@ github.com/eclipse/paho.golang v0.10.0 h1:oUGPjRwWcZQRgDD9wVDV7y7i7yBSxts3vcvcNJ
github.com/eclipse/paho.golang v0.10.0/go.mod h1:rhrV37IEwauUyx8FHrvmXOKo+QRKng5ncoN1vJiJMcs=
github.com/go-stomp/stomp v2.1.4+incompatible h1:D3SheUVDOz9RsjVWkoh/1iCOwD0qWjyeTZMUZ0EXg2Y=
github.com/go-stomp/stomp v2.1.4+incompatible/go.mod h1:VqCtqNZv1226A1/79yh+rMiFUcfY3R109np+7ke4n0c=
github.com/gofrs/uuid v4.0.0+incompatible h1:1SD/1F5pU8p29ybwgQSwpQk+mwdRrXCYuPhW6m+TnJw=
github.com/gofrs/uuid v4.0.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
github.com/golang-jwt/jwt/v5 v5.2.0 h1:d/ix8ftRUorsN+5eMIlF4T6J8CAt9rch3My2winC1Jw=
github.com/golang-jwt/jwt/v5 v5.2.0/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/gomodule/redigo v1.8.4 h1:Z5JUg94HMTR1XpwBaSH4vq3+PNSIykBLxMdglbw10gg=
github.com/gomodule/redigo v1.8.4/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/googollee/go-socket.io v1.7.0 h1:ODcQSAvVIPvKozXtUGuJDV3pLwdpBLDs1Uoq/QHIlY8=
github.com/googollee/go-socket.io v1.7.0/go.mod h1:0vGP8/dXR9SZUMMD4+xxaGo/lohOw3YWMh2WRiWeKxg=
github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
@ -43,7 +37,6 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
github.com/rs/cors v1.9.0 h1:l9HGsTsHJcvW14Nk7J9KFz8bzeAWXn3CG6bgt7LsrAE=
github.com/rs/cors v1.9.0/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@ -87,8 +80,6 @@ google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqw
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

38
backend/services/controller/internal/api/api.go
View File

@ -12,19 +12,19 @@ import (
"github.com/leandrofars/oktopus/internal/api/middleware"
"github.com/leandrofars/oktopus/internal/db"
"github.com/leandrofars/oktopus/internal/mqtt"
"github.com/leandrofars/oktopus/internal/mtp"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/utils"
"github.com/leandrofars/oktopus/internal/ws"
"google.golang.org/protobuf/proto"
)
type Api struct {
Port string
Db db.Database
Broker mtp.Broker
MsgQueue map[string](chan usp_msg.Msg)
QMutex *sync.Mutex
Mqtt mqtt.Mqtt
Port string
Db db.Database
MsgQueue map[string](chan usp_msg.Msg)
QMutex *sync.Mutex
Mqtt *mqtt.Mqtt
Websockets *ws.Ws
}
const REQUEST_TIMEOUT = time.Second * 30
@ -34,14 +34,14 @@ const (
AdminUser
)
func NewApi(port string, db db.Database, mqtt *mqtt.Mqtt, msgQueue map[string](chan usp_msg.Msg), m *sync.Mutex) Api {
func NewApi(port string, db db.Database, mqtt *mqtt.Mqtt, msgQueue map[string](chan usp_msg.Msg), m *sync.Mutex, w ws.Ws) Api {
return Api{
Port: port,
Db: db,
Broker: mqtt,
MsgQueue: msgQueue,
QMutex: m,
Mqtt: *mqtt,
Port: port,
Db: db,
MsgQueue: msgQueue,
QMutex: m,
Mqtt: mqtt,
Websockets: &w,
}
}
@ -133,8 +133,14 @@ func (a *Api) uspCall(msg usp_msg.Msg, sn string, w http.ResponseWriter, device
a.MsgQueue[msg.Header.MsgId] = make(chan usp_msg.Msg)
a.QMutex.Unlock()
log.Println("Sending Msg:", msg.Header.MsgId)
//TODO: Check what MTP the device is connected to
a.Broker.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
if device.Mqtt == db.Online {
a.Mqtt.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
} else if device.Websockets == db.Online {
a.Websockets.Publish(tr369Message, "", "", false)
} else if device.Stomp == db.Online {
//TODO: send stomp message
}
select {
case msg := <-a.MsgQueue[msg.Header.MsgId]:

1
backend/services/controller/internal/api/device.go
View File

@ -232,6 +232,7 @@ func (a *Api) deviceUpdateMsg(w http.ResponseWriter, r *http.Request) {
a.uspCall(msg, sn, w, device)
}
// TODO: react this function, return err and deal with it in the caller, remove header superfluos
func (a *Api) deviceExists(sn string, w http.ResponseWriter) db.Device {
device, err := a.Db.RetrieveDevice(sn)
if err != nil {

10
backend/services/controller/internal/api/fw_update.go
View File

@ -7,6 +7,7 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/leandrofars/oktopus/internal/db"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/utils"
"google.golang.org/protobuf/proto"
@ -51,7 +52,14 @@ func (a *Api) deviceFwUpdate(w http.ResponseWriter, r *http.Request) {
a.MsgQueue[msg.Header.MsgId] = make(chan usp_msg.Msg)
a.QMutex.Unlock()
log.Println("Sending Msg:", msg.Header.MsgId)
a.Broker.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
if device.Mqtt == db.Online {
a.Mqtt.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
} else if device.Websockets == db.Online {
a.Websockets.Publish(tr369Message, "", "", false)
} else if device.Stomp == db.Online {
//TODO: send stomp message
}
var getMsgAnswer *usp_msg.GetResp

14
backend/services/controller/internal/api/info.go
View File

@ -8,7 +8,6 @@ import (
"time"
"github.com/leandrofars/oktopus/internal/db"
"github.com/leandrofars/oktopus/internal/utils"
)
type StatusCount struct {
@ -23,6 +22,7 @@ type GeneralInfo struct {
VendorsCount []db.VendorsCount
}
// TODO: fix when mqtt broker is not set don't break api
func (a *Api) generalInfo(w http.ResponseWriter, r *http.Request) {
var result GeneralInfo
@ -49,10 +49,10 @@ func (a *Api) generalInfo(w http.ResponseWriter, r *http.Request) {
}
for _, v := range statuscount {
switch v.Status {
case utils.Online:
switch db.Status(v.Status) {
case db.Online:
result.StatusCount.Online = v.Count
case utils.Offline:
case db.Offline:
result.StatusCount.Offline = v.Count
}
}
@ -120,10 +120,10 @@ func (a *Api) statusInfo(w http.ResponseWriter, r *http.Request) {
var status StatusCount
for _, v := range vendors {
switch v.Status {
case utils.Online:
switch db.Status(v.Status) {
case db.Online:
status.Online = v.Count
case utils.Offline:
case db.Offline:
status.Offline = v.Count
}
}

12
backend/services/controller/internal/api/wifi.go
View File

@ -10,6 +10,7 @@ import (
"github.com/gorilla/mux"
"github.com/leandrofars/oktopus/internal/db"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/utils"
"google.golang.org/protobuf/proto"
@ -31,7 +32,7 @@ type WiFi struct {
func (a *Api) deviceWifi(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
sn := vars["sn"]
a.deviceExists(sn, w)
device := a.deviceExists(sn, w)
if r.Method == http.MethodGet {
msg := utils.NewGetMsg(usp_msg.Get{
@ -69,7 +70,14 @@ func (a *Api) deviceWifi(w http.ResponseWriter, r *http.Request) {
a.MsgQueue[msg.Header.MsgId] = make(chan usp_msg.Msg)
a.QMutex.Unlock()
log.Println("Sending Msg:", msg.Header.MsgId)
a.Broker.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
if device.Mqtt == db.Online {
a.Mqtt.Publish(tr369Message, "oktopus/v1/agent/"+sn, "oktopus/v1/api/"+sn, false)
} else if device.Websockets == db.Online {
a.Websockets.Publish(tr369Message, "", "", false)
} else if device.Stomp == db.Online {
//TODO: send stomp message
}
//TODO: verify in protocol and in other models, the Device.Wifi parameters. Maybe in the future, to use SSIDReference from AccessPoint
select {

28
backend/services/controller/internal/db/db.go
View File

@ -2,24 +2,34 @@ package db
import (
"context"
"log"
"sync"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"log"
)
//TODO: create another package fo structs and interfaces
type Database struct {
client *mongo.Client
devices *mongo.Collection
users *mongo.Collection
ctx context.Context
m *sync.Mutex
}
func NewDatabase(ctx context.Context, mongoUri string) Database {
var db Database
clientOptions := options.Client().ApplyURI(mongoUri)
client, err := mongo.Connect(ctx, clientOptions)
if err != nil {
log.Fatal(err)
}
db.client = client
log.Println("Trying to ping Mongo database...")
err = client.Ping(ctx, nil)
if err != nil {
@ -27,10 +37,26 @@ func NewDatabase(ctx context.Context, mongoUri string) Database {
}
log.Println("Connected to MongoDB-->", mongoUri)
devices := client.Database("oktopus").Collection("devices")
createIndexes(ctx, devices)
users := client.Database("oktopus").Collection("users")
db.devices = devices
db.users = users
db.ctx = ctx
db.m = &sync.Mutex{}
return db
}
func createIndexes(ctx context.Context, devices *mongo.Collection) {
indexField := bson.M{"sn": 1}
_, err := devices.Indexes().CreateOne(ctx, mongo.IndexModel{
Keys: indexField,
Options: options.Index().SetUnique(true),
})
if err != nil {
log.Println("ERROR to create index in database:", err)
}
}

73
backend/services/controller/internal/db/device.go
View File

@ -17,6 +17,14 @@ const (
WEBSOCKETS
)
type Status uint8
const (
Offline Status = iota
Associating
Online
)
type Device struct {
SN string
Model string
@ -24,22 +32,67 @@ type Device struct {
Vendor string
Version string
ProductClass string
Status uint8
MTP []map[string]string
Status Status
Mqtt Status
Stomp Status
Websockets Status
}
// TODO: don't change device status of other MTP
func (d *Database) CreateDevice(device Device) error {
var result bson.M
opts := options.FindOneAndReplace().SetUpsert(true)
err := d.devices.FindOneAndReplace(d.ctx, bson.D{{"sn", device.SN}}, device, opts).Decode(&result)
if err != nil {
if err == mongo.ErrNoDocuments {
log.Printf("New device %s added to database", device.SN)
return nil
var deviceExistent Device
d.m.Lock()
defer d.m.Unlock()
/* ------------------ Do not overwrite status of other mtp ------------------ */
err := d.devices.FindOne(d.ctx, bson.D{{"sn", device.SN}}, nil).Decode(&deviceExistent)
if err == nil {
if deviceExistent.Mqtt == Online {
device.Mqtt = Online
}
if deviceExistent.Stomp == Online {
device.Stomp = Online
}
if deviceExistent.Websockets == Online {
device.Websockets = Online
}
} else {
if err != nil && err != mongo.ErrNoDocuments {
log.Println(err)
return err
}
log.Fatal(err)
}
log.Printf("Device %s already existed, and got replaced for new info", device.SN)
/* -------------------------------------------------------------------------- */
callback := func(sessCtx mongo.SessionContext) (interface{}, error) {
// Important: You must pass sessCtx as the Context parameter to the operations for them to be executed in the
// transaction.
opts := options.FindOneAndReplace().SetUpsert(true)
err = d.devices.FindOneAndReplace(d.ctx, bson.D{{"sn", device.SN}}, device, opts).Decode(&result)
if err != nil {
if err == mongo.ErrNoDocuments {
log.Printf("New device %s added to database", device.SN)
return nil, nil
}
return nil, err
}
log.Printf("Device %s already existed, and got replaced for new info", device.SN)
return nil, nil
}
session, err := d.client.StartSession()
if err != nil {
return err
}
defer session.EndSession(d.ctx)
_, err = session.WithTransaction(d.ctx, callback)
if err != nil {
return err
}
return err
}
func (d *Database) RetrieveDevices(filter bson.A) ([]Device, error) {

54
backend/services/controller/internal/db/status.go
View File

@ -4,13 +4,59 @@ import (
"log"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
)
// TODO: fix this function to also change device status at different mtp
func (d *Database) UpdateStatus(sn string, status uint8) error {
var result bson.M
err := d.devices.FindOneAndUpdate(d.ctx, bson.D{{"sn", sn}}, bson.D{{"$set", bson.D{{"status", status}}}}).Decode(&result)
func (d *Database) UpdateStatus(sn string, status Status, mtp MTP) error {
var result Device
d.m.Lock()
defer d.m.Unlock()
err := d.devices.FindOne(d.ctx, bson.D{{"sn", sn}}, nil).Decode(&result)
if err != nil {
log.Println(err)
}
//TODO: abolish this logic, find another approach, microservices design maybe?
/*
In case the device status is online, we must check if the mtp
changing is going to affect the global status. In case it does,
we must update the global status accordingly.
*/
/*
mix the existent device status to the updated one
*/
switch mtp {
case MQTT:
result.Mqtt = status
case STOMP:
result.Stomp = status
case WEBSOCKETS:
result.Websockets = status
}
/*
check if the global status needs update
*/
var globalStatus primitive.E
if result.Mqtt == Offline && result.Stomp == Offline && result.Websockets == Offline {
globalStatus = primitive.E{"status", Offline}
}
if result.Mqtt == Online || result.Stomp == Online || result.Websockets == Online {
globalStatus = primitive.E{"status", Online}
}
_, err = d.devices.UpdateOne(d.ctx, bson.D{{"sn", sn}}, bson.D{
{
"$set", bson.D{
{mtp.String(), status},
globalStatus,
},
},
})
if err != nil {
if err == mongo.ErrNoDocuments {
log.Printf("Device %s is not mapped into database", sn)

83
backend/services/controller/internal/mqtt/mqtt.go
View File

@ -12,6 +12,7 @@ import (
"github.com/eclipse/paho.golang/autopaho"
"github.com/eclipse/paho.golang/paho"
"github.com/leandrofars/oktopus/internal/db"
"github.com/leandrofars/oktopus/internal/mtp/handler"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/usp_record"
"github.com/leandrofars/oktopus/internal/utils"
@ -86,6 +87,7 @@ func (m *Mqtt) Connect() {
}
func (m *Mqtt) Disconnect() {
log.Println("Disconnecting from MQTT broker...")
err := c.Disconnect(m.Ctx)
if err != nil {
log.Fatalf("failed to send Disconnect: %s", err)
@ -183,7 +185,8 @@ func (m *Mqtt) messageHandler(status, controller, apiMsg chan *paho.Publish) {
}
if payload == ONLINE {
log.Println("Device connected:", device)
m.handleNewDevice(device)
tr369Message := handler.HandleNewDevice(device)
m.Publish(tr369Message, "oktopus/v1/agent/"+device, "oktopus/v1/controller/"+device, false)
//m.deleteRetainedMessage(d, device)
} else if payload == OFFLINE {
log.Println("Device disconnected:1", device)
@ -195,7 +198,11 @@ func (m *Mqtt) messageHandler(status, controller, apiMsg chan *paho.Publish) {
case c := <-controller:
topic := c.Topic
sn := strings.Split(topic, "/")
m.handleNewDevicesResponse(c.Payload, sn[3])
device := handler.HandleNewDevicesResponse(c.Payload, sn[3], db.MQTT)
err := m.DB.CreateDevice(device)
if err != nil {
log.Fatal(err)
}
case api := <-apiMsg:
log.Println("Handle api request")
m.handleApiRequest(api.Payload)
@ -231,79 +238,9 @@ func (m *Mqtt) handleApiRequest(api []byte) {
}
}
func (m *Mqtt) handleNewDevice(deviceMac string) {
payload := usp_msg.Msg{
Header: &usp_msg.Header{
MsgId: "uniqueIdentifierForThismessage",
MsgType: usp_msg.Header_GET,
},
Body: &usp_msg.Body{
MsgBody: &usp_msg.Body_Request{
Request: &usp_msg.Request{
ReqType: &usp_msg.Request_Get{
Get: &usp_msg.Get{
ParamPaths: []string{
"Device.DeviceInfo.Manufacturer",
"Device.DeviceInfo.ModelName",
"Device.DeviceInfo.SoftwareVersion",
"Device.DeviceInfo.SerialNumber",
"Device.DeviceInfo.ProductClass",
},
MaxDepth: 1,
},
},
},
},
},
}
teste, _ := proto.Marshal(&payload)
record := utils.NewUspRecord(teste, deviceMac)
tr369Message, err := proto.Marshal(&record)
if err != nil {
log.Fatalln("Failed to encode tr369 record:", err)
}
m.Publish(tr369Message, "oktopus/v1/agent/"+deviceMac, "oktopus/v1/controller/"+deviceMac, false)
}
func (m *Mqtt) handleNewDevicesResponse(p []byte, sn string) {
var record usp_record.Record
var message usp_msg.Msg
err := proto.Unmarshal(p, &record)
if err != nil {
log.Fatal(err)
}
err = proto.Unmarshal(record.GetNoSessionContext().Payload, &message)
if err != nil {
log.Fatal(err)
}
var device db.Device
msg := message.Body.MsgBody.(*usp_msg.Body_Response).Response.GetGetResp()
device.Vendor = msg.ReqPathResults[0].ResolvedPathResults[0].ResultParams["Manufacturer"]
device.Model = msg.ReqPathResults[1].ResolvedPathResults[0].ResultParams["ModelName"]
device.Version = msg.ReqPathResults[2].ResolvedPathResults[0].ResultParams["SoftwareVersion"]
device.ProductClass = msg.ReqPathResults[4].ResolvedPathResults[0].ResultParams["ProductClass"]
device.SN = sn
mtp := map[string]string{
db.MQTT.String(): "online",
}
device.MTP = append(device.MTP, mtp)
device.Status = utils.Online
err = m.DB.CreateDevice(device)
if err != nil {
log.Fatal(err)
}
}
func (m *Mqtt) handleDevicesDisconnect(p string) {
// Update status of device at database
err := m.DB.UpdateStatus(p, utils.Offline)
err := m.DB.UpdateStatus(p, db.Offline, db.MQTT)
if err != nil {
log.Fatal(err)
}

70
backend/services/controller/internal/mtp/handler/handler.go Normal file
View File

@ -0,0 +1,70 @@
package handler
import (
"log"
"github.com/leandrofars/oktopus/internal/db"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/usp_record"
"github.com/leandrofars/oktopus/internal/utils"
"google.golang.org/protobuf/proto"
)
func HandleNewDevice(deviceMac string) []byte {
payload := utils.NewGetMsg(usp_msg.Get{
ParamPaths: []string{
"Device.DeviceInfo.Manufacturer",
"Device.DeviceInfo.ModelName",
"Device.DeviceInfo.SoftwareVersion",
"Device.DeviceInfo.SerialNumber",
"Device.DeviceInfo.ProductClass",
},
MaxDepth: 1,
})
teste, _ := proto.Marshal(&payload)
record := utils.NewUspRecord(teste, deviceMac)
tr369Message, err := proto.Marshal(&record)
if err != nil {
log.Fatalln("Failed to encode tr369 record:", err)
}
return tr369Message
}
func HandleNewDevicesResponse(p []byte, sn string, mtp db.MTP) db.Device {
var record usp_record.Record
var message usp_msg.Msg
err := proto.Unmarshal(p, &record)
if err != nil {
log.Fatal(err)
}
err = proto.Unmarshal(record.GetNoSessionContext().Payload, &message)
if err != nil {
log.Fatal(err)
}
var device db.Device
msg := message.Body.MsgBody.(*usp_msg.Body_Response).Response.GetGetResp()
device.Vendor = msg.ReqPathResults[0].ResolvedPathResults[0].ResultParams["Manufacturer"]
device.Model = msg.ReqPathResults[1].ResolvedPathResults[0].ResultParams["ModelName"]
device.Version = msg.ReqPathResults[2].ResolvedPathResults[0].ResultParams["SoftwareVersion"]
device.ProductClass = msg.ReqPathResults[4].ResolvedPathResults[0].ResultParams["ProductClass"]
device.SN = sn
switch db.MTP(mtp) {
case db.MQTT:
device.Mqtt = db.Online
case db.WEBSOCKETS:
device.Websockets = db.Online
case db.STOMP:
device.Stomp = db.Online
}
device.Status = db.Online
return device
}

15
backend/services/controller/internal/mtp/mtp.go
View File

@ -23,7 +23,7 @@ type Broker interface {
//Request(msg []byte, msgType usp_msg.Header_MsgType, pubTopic string, subTopic string)
}
// Not used, since we are using a broker solution.
// Not used, since we are using a broker approach.
type P2P interface {
}
@ -31,14 +31,7 @@ type P2P interface {
func MtpService(b Broker, done chan os.Signal, wg *sync.WaitGroup) {
b.Connect()
wg.Done()
go func() {
for range done {
b.Disconnect()
log.Println("Successfully disconnected to MTPs!")
// Receives signal and then replicates it to the rest of the app.
done <- os.Interrupt
}
}()
//b.Subscribe()
<-done
log.Println("Disconnect of MTP!")
b.Disconnect()
}

10
backend/services/controller/internal/utils/utils.go
View File

@ -1,18 +1,14 @@
package utils
import (
"net"
"github.com/google/uuid"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/usp_record"
"net"
)
// Status are saved at database as numbers
const (
Online = iota
Associating
Offline
)
//TODO: change usp utils related to another package
// Get interfaces MACs, and the first interface MAC is gonna be used as mqtt clientId
func GetMacAddr() ([]string, error) {

289
backend/services/controller/internal/ws/ws.go
View File

@ -1,88 +1,239 @@
package ws
import (
"fmt"
"context"
"crypto/tls"
"encoding/json"
"log"
"net/http"
"reflect"
"sync"
"time"
socketio "github.com/googollee/go-socket.io"
"github.com/googollee/go-socket.io/engineio"
"github.com/googollee/go-socket.io/engineio/transport"
"github.com/googollee/go-socket.io/engineio/transport/polling"
"github.com/googollee/go-socket.io/engineio/transport/websocket"
"github.com/gorilla/mux"
"github.com/leandrofars/oktopus/internal/api/cors"
"github.com/gorilla/websocket"
"github.com/leandrofars/oktopus/internal/db"
"github.com/leandrofars/oktopus/internal/mtp/handler"
usp_msg "github.com/leandrofars/oktopus/internal/usp_message"
"github.com/leandrofars/oktopus/internal/usp_record"
"google.golang.org/protobuf/proto"
)
/* ----------- [Deprecated code] migrated to Socketio with NodeJs ----------- */
func Ws() {
server := socketio.NewServer(&engineio.Options{
PingTimeout: 5 * time.Second,
PingInterval: 10 * time.Second,
Transports: []transport.Transport{
&polling.Transport{
Client: &http.Client{
Timeout: time.Minute,
},
CheckOrigin: func(r *http.Request) bool {
return true
},
},
&websocket.Transport{
CheckOrigin: func(r *http.Request) bool {
return true
},
},
},
})
type Ws struct {
Addr string
Port string
Token string
Route string
Auth bool
TLS bool
InsecureSkipVerify bool
Ctx context.Context
NewDeviceQueue map[string]string
NewDevQMutex *sync.Mutex
DB db.Database
MsgQueue map[string](chan usp_msg.Msg)
QMutex *sync.Mutex
}
server.OnConnect("/", func(s socketio.Conn) error {
s.SetContext("")
fmt.Println("connected:", s.ID())
return nil
})
const (
WS_CONNECTION_RETRY = 10 * time.Second
)
server.OnEvent("/", "offer", func(s socketio.Conn, msg string) string {
log.Printf("offer: %s", msg)
const (
OFFLINE = "0"
ONLINE = "1"
)
return "test"
})
type deviceStatus struct {
Eid string
Status string
}
server.OnError("/", func(s socketio.Conn, e error) {
fmt.Println("error:", e)
})
// Global Websocket connection used in this package
var wsConn *websocket.Conn
server.OnDisconnect("/", func(s socketio.Conn, reason string) {
fmt.Println("closed", reason)
})
func (w *Ws) Connect() {
log.Println("Connecting to WS endpoint...")
go func() {
if err := server.Serve(); err != nil {
log.Fatalf("socketio listen error: %s\n", err)
}
}()
const wsPort = "5000"
r := mux.NewRouter()
r.Handle("/socket.io/", server)
//r.Use(func(handler http.Handler) http.Handler {
// return middleware.Middleware(handler)
//})
corsOpts := cors.GetCorsConfig()
srv := &http.Server{
Addr: "0.0.0.0:" + wsPort,
Handler: corsOpts.Handler(r), // Pass our instance of gorilla/mux in.
prefix := "ws://"
if w.TLS {
prefix = "wss://"
}
go func() {
if err := srv.ListenAndServe(); err != nil {
wsUrl := prefix + w.Addr + ":" + w.Port + w.Route
if w.Auth {
log.Println("WS token:", w.Token)
// e.g. ws://localhost:8080/ws/controller?token=123456
wsUrl = wsUrl + "?token=" + w.Token
}
dialer := websocket.Dialer{
TLSClientConfig: &tls.Config{
InsecureSkipVerify: w.InsecureSkipVerify,
},
}
// Keeps trying to connect to the WS endpoint until it succeeds or receives a stop signal
go func(dialer websocket.Dialer) {
for {
c, _, err := dialer.Dial(wsUrl, nil)
if err != nil {
log.Printf("Error to connect to %s, err: %s", wsUrl, err)
time.Sleep(WS_CONNECTION_RETRY)
continue
}
// instantiate global ws connection
wsConn = c
log.Println("Connected to WS endpoint--> ", wsUrl)
go w.Subscribe()
break
}
}(dialer)
}
func (w *Ws) Disconnect() {
log.Println("Disconnecting from WS endpoint...")
if wsConn != nil {
err := wsConn.Close()
if err != nil {
log.Println("Error while disconnecting from WS endpoint:", err.Error())
}
log.Println("Succesfully disconnected from WS endpoint")
} else {
log.Println("No WS connection to close")
}
}
// Websockets doesn't follow pub/sub architecture, but we use these
// naming here to implement the Broker interface and abstract the MTP layer.
/* -------------------------------------------------------------------------- */
func (w *Ws) Subscribe() {
var m sync.Mutex
w.NewDevQMutex = &m
w.NewDeviceQueue = make(map[string]string)
for {
msgType, wsMsg, err := wsConn.ReadMessage()
if err != nil {
if websocket.IsCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
log.Printf("websocket error: %v", err)
w.Connect()
return
}
log.Println("websocket unexpected error:", err)
return
}
//TODO: deal with message in new go routine
if msgType == websocket.TextMessage {
var deviceStatus deviceStatus
err = json.Unmarshal(wsMsg, &deviceStatus)
if err != nil {
log.Println("Websockets Text Message is not about devices status")
}
log.Println("Received device status message")
var status db.Status
switch deviceStatus.Status {
case ONLINE:
status = db.Online
case OFFLINE:
status = db.Offline
default:
log.Println("Invalid device status")
return
}
w.DB.UpdateStatus(deviceStatus.Eid, status, db.WEBSOCKETS)
//TODO: return error 1003 to device
//TODO: get status messages
continue
}
//log.Printf("binary data: %s", string(wsMsg))
//TODO: if error at processing message return error 1003 to devicec
//TODO: deal with received messages in parallel
var record usp_record.Record
//var message usp_msg.Msg
err = proto.Unmarshal(wsMsg, &record)
if err != nil {
log.Println(err)
}
}()
log.Printf("Running websocket at port %s", wsPort)
connRecord := &usp_record.Record_WebsocketConnect{
WebsocketConnect: &usp_record.WebSocketConnectRecord{},
}
noSessionRecord := &usp_record.Record_NoSessionContext{
NoSessionContext: &usp_record.NoSessionContextRecord{},
}
//log.Printf("Record Type: %++v", record.RecordType)
deviceId := record.FromId
// New Device Handler
if reflect.TypeOf(record.RecordType) == reflect.TypeOf(connRecord) {
log.Println("Websocket new device:", deviceId)
tr369Message := handler.HandleNewDevice(deviceId)
w.NewDevQMutex.Lock()
w.NewDeviceQueue[deviceId] = ""
w.NewDevQMutex.Unlock()
w.Publish(tr369Message, "", "", false)
continue
}
//TODO: see what type of message was received
if reflect.TypeOf(record.RecordType) == reflect.TypeOf(noSessionRecord) {
//log.Printf("Websocket device %s message", record.FromId)
// New device answer
if _, ok := w.NewDeviceQueue[deviceId]; ok {
log.Printf("New device %s response", deviceId)
device := handler.HandleNewDevicesResponse(wsMsg, deviceId, db.WEBSOCKETS)
w.NewDevQMutex.Lock()
delete(w.NewDeviceQueue, deviceId)
w.NewDevQMutex.Unlock()
w.DB.CreateDevice(device)
if err != nil {
log.Fatal(err)
}
continue
}
log.Println("Handle api request")
var msg usp_msg.Msg
err = proto.Unmarshal(record.GetNoSessionContext().Payload, &msg)
if err != nil {
log.Println(err)
continue
}
if _, ok := w.MsgQueue[msg.Header.MsgId]; ok {
//m.QMutex.Lock()
w.MsgQueue[msg.Header.MsgId] <- msg
//m.QMutex.Unlock()
} else {
log.Printf("Message answer to request %s arrived too late", msg.Header.MsgId)
}
}
//log.Printf("recv: %++v", record)
}
}
func (w *Ws) Publish(msg []byte, topic, respTopic string, retain bool) {
err := wsConn.WriteMessage(websocket.BinaryMessage, msg)
if err != nil {
log.Println("write:", err)
return
}
}
/* -------------------------------------------------------------------------- */

3
backend/services/stomp/server/processor.go
View File

@ -1,6 +1,7 @@
package server
import (
"log"
"net"
"strings"
"time"
@ -50,6 +51,8 @@ func (proc *requestProcessor) Serve(l net.Listener) error {
} else {
topic := proc.tm.Find(r.Sub.Destination())
topic.Subscribe(r.Sub)
//TODO: if subscribed to oktopus/v1/agent send online message to ...status topic
log.Println(r.Sub.Destination())
}
case client.UnsubscribeOp:

5
backend/services/ws/.env Normal file
View File

@ -0,0 +1,5 @@
SERVER_PORT=""
SERVER_AUTH_TOKEN=""
CONTROLLER_EID=""
SERVER_AUTH_ENABLE=""
SERVER_TLS_ENABLE=""

3
backend/services/ws/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
.env.local
ws
*.pem

24
backend/services/ws/cmd/main.go
View File

@ -1,9 +1,31 @@
package main
import (
"log"
"os"
"os/signal"
"syscall"
"github.com/OktopUSP/oktopus/ws/internal/config"
"github.com/OktopUSP/oktopus/ws/internal/ws"
)
// TODO: refact from where this version is get
const VERSION = "0.0.1"
func main() {
ws.StartNewServer()
done := make(chan os.Signal, 1)
conf := config.NewConfig()
// Locks app running until it receives a stop command as Ctrl+C.
signal.Notify(done, syscall.SIGINT)
log.Println("Starting Oktopus Websockets Version:", VERSION)
ws.StartNewServer(conf)
<-done
log.Println("(⌐■_■) Websockets server is out!")
}

2
backend/services/ws/go.mod
View File

@ -5,6 +5,8 @@ go 1.21.6
require (
github.com/gorilla/mux v1.8.1
github.com/gorilla/websocket v1.5.1
github.com/joho/godotenv v1.5.1
google.golang.org/protobuf v1.32.0
)
require golang.org/x/net v0.17.0 // indirect

8
backend/services/ws/go.sum
View File

@ -1,6 +1,14 @@
github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=

112
backend/services/ws/internal/config/config.go Normal file
View File

@ -0,0 +1,112 @@
// Loads environemnt variables and returns a config struct
package config
import (
"flag"
"log"
"os"
"strconv"
"github.com/joho/godotenv"
)
type Config struct {
Port string // server port: e.g. ":8080"
Auth bool // server auth enable/disable
Token string // controller auth token
ControllerEID string // controller endpoint id
Tls bool // enable/diable websockets server tls
}
func NewConfig() Config {
//Defines log format
log.SetFlags(log.LstdFlags | log.Lshortfile)
loadEnv()
/*
App variables priority:
1º - Flag through command line.
2º - Env variables.
3º - Default flag value.
*/
/* ------------------------------ define flags ------------------------------ */
flPort := flag.String("port", lookupEnvOrString("SERVER_PORT", ":8080"), "Server port")
flToken := flag.String("token", lookupEnvOrString("SERVER_AUTH_TOKEN", ""), "Controller auth token")
flAuth := flag.Bool("auth", lookupEnvOrBool("SERVER_AUTH_ENABLE", false), "Server auth enable/disable")
flControllerEid := flag.String("controller-eid", lookupEnvOrString("CONTROLLER_EID", "oktopusController"), "Controller eid")
flTls := flag.Bool("tls", lookupEnvOrBool("SERVER_TLS_ENABLE", false), "Enable/diable websockets server tls")
flHelp := flag.Bool("help", false, "Help")
flag.Parse()
/* -------------------------------------------------------------------------- */
if *flHelp {
flag.Usage()
os.Exit(0)
}
return Config{
Port: *flPort,
Token: *flToken,
Auth: *flAuth,
ControllerEID: *flControllerEid,
Tls: *flTls,
}
}
// Load environment variables from .env or .env.local file
func loadEnv() {
err := godotenv.Load()
if err != nil {
log.Println("Error to load environment variables:", err)
}
localEnv := ".env.local"
if _, err := os.Stat(localEnv); err == nil {
_ = godotenv.Overload(localEnv)
log.Println("Loaded variables from '.env.local'")
} else {
log.Println("Loaded variables from '.env'")
}
}
/* ---------------------------- helper functions ---------------------------- */
/*
They are used to lookup if a environment variable is set with a value
different of "" and return it.
In case the var doesn't exist, it returns the default value.
Also, they're useful to convert the string value of vars to the desired type.
*/
func lookupEnvOrString(key string, defaultVal string) string {
if val, _ := os.LookupEnv(key); val != "" {
return val
}
return defaultVal
}
func lookupEnvOrInt(key string, defaultVal int) int {
if val, _ := os.LookupEnv(key); val != "" {
v, err := strconv.Atoi(val)
if err != nil {
log.Fatalf("LookupEnvOrInt[%s]: %v", key, err)
}
return v
}
return defaultVal
}
func lookupEnvOrBool(key string, defaultVal bool) bool {
if val, _ := os.LookupEnv(key); val != "" {
v, err := strconv.ParseBool(val)
if err != nil {
log.Fatalf("LookupEnvOrBool[%s]: %v", key, err)
}
return v
}
return defaultVal
}
/* -------------------------------------------------------------------------- */

6606
backend/services/ws/internal/usp_message/usp-msg-1-2.pb.go Executable file
View File

File diff suppressed because it is too large Load Diff

530
backend/services/ws/internal/usp_message/usp-msg-1-2.proto Executable file
View File

@ -0,0 +1,530 @@
syntax = "proto3";
//**************************************************************************
// TR-369 USP Message Protocol Buffer Schema
//
// Copyright (c) 2017-2018, Broadband Forum
//
// The undersigned members have elected to grant the copyright to
// their contributed material used in this software:
// Copyright (c) 2017-2018 ARRIS Enterprises, LLC.
//
// Redistribution and use in source and binary forms, with or
// without modification, are permitted provided that the following
// conditions are met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following
// disclaimer in the documentation and/or other materials
// provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its
// contributors may be used to endorse or promote products
// derived from this software without specific prior written
// permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
// CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
// NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
// ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// The above license is used as a license under copyright only.
// Please reference the Forum IPR Policy for patent licensing terms
// <https://www.broadband-forum.org/ipr-policy>.
//
// Any moral rights which are necessary to exercise under the above
// license grant are also deemed granted under this license.
//
// | Version | Name | Date |
// | TR-369 1.0.0 | User Services Platform | APR, 2018 |
// | TR-369 1.0.1 | User Services Platform | JUN, 2018 |
// | TR-369 1.0.2 | User Services Platform | OCT, 2018 |
// | TR-369 1.1 | User Services Platform | SEP, 2019 |
//
// BBF software release registry: http://www.broadband-forum.org/software
//**************************************************************************
package usp;
option go_package="./usp-msg";
message Msg {
Header header = 1; // Make required in the protocol
Body body = 2; // Make required in the protocol
}
message Header {
string msg_id = 1; // Make required in the protocol
MsgType msg_type = 2; // Make required in the protocol
enum MsgType {
ERROR = 0;
GET = 1;
GET_RESP = 2;
NOTIFY = 3;
SET = 4;
SET_RESP = 5;
OPERATE = 6;
OPERATE_RESP = 7;
ADD = 8;
ADD_RESP = 9;
DELETE = 10;
DELETE_RESP = 11;
GET_SUPPORTED_DM = 12;
GET_SUPPORTED_DM_RESP = 13;
GET_INSTANCES = 14;
GET_INSTANCES_RESP = 15;
NOTIFY_RESP = 16;
GET_SUPPORTED_PROTO = 17;
GET_SUPPORTED_PROTO_RESP = 18;
}
}
message Body {
oneof msg_body {
Request request = 1;
Response response = 2;
Error error = 3;
}
}
message Request {
oneof req_type {
Get get = 1;
GetSupportedDM get_supported_dm = 2;
GetInstances get_instances = 3;
Set set = 4;
Add add = 5;
Delete delete = 6;
Operate operate = 7;
Notify notify = 8;
GetSupportedProtocol get_supported_protocol = 9;
}
}
message Response {
oneof resp_type {
GetResp get_resp = 1;
GetSupportedDMResp get_supported_dm_resp = 2;
GetInstancesResp get_instances_resp = 3;
SetResp set_resp = 4;
AddResp add_resp = 5;
DeleteResp delete_resp = 6;
OperateResp operate_resp = 7;
NotifyResp notify_resp = 8;
GetSupportedProtocolResp get_supported_protocol_resp = 9;
}
}
message Error {
fixed32 err_code = 1;
string err_msg = 2;
repeated ParamError param_errs = 3;
message ParamError {
string param_path = 1;
fixed32 err_code = 2;
string err_msg = 3;
}
}
message Get {
repeated string param_paths = 1;
fixed32 max_depth = 2;
}
message GetResp {
repeated RequestedPathResult req_path_results = 1;
message RequestedPathResult {
string requested_path = 1;
fixed32 err_code = 2;
string err_msg = 3;
repeated ResolvedPathResult resolved_path_results = 4;
}
message ResolvedPathResult {
string resolved_path = 1;
map<string, string> result_params = 2;
}
}
message GetSupportedDM {
repeated string obj_paths = 1;
bool first_level_only = 2;
bool return_commands = 3;
bool return_events = 4;
bool return_params = 5;
}
message GetSupportedDMResp {
repeated RequestedObjectResult req_obj_results = 1;
message RequestedObjectResult {
string req_obj_path = 1;
fixed32 err_code = 2;
string err_msg = 3;
string data_model_inst_uri = 4;
repeated SupportedObjectResult supported_objs = 5;
}
message SupportedObjectResult {
string supported_obj_path = 1;
ObjAccessType access = 2;
bool is_multi_instance = 3;
repeated SupportedCommandResult supported_commands = 4;
repeated SupportedEventResult supported_events = 5;
repeated SupportedParamResult supported_params = 6;
repeated string divergent_paths = 7;
}
message SupportedParamResult {
string param_name = 1;
ParamAccessType access = 2;
ParamValueType value_type = 3;
ValueChangeType value_change = 4;
}
message SupportedCommandResult {
string command_name = 1;
repeated string input_arg_names = 2;
repeated string output_arg_names = 3;
CmdType command_type = 4;
}
message SupportedEventResult {
string event_name = 1;
repeated string arg_names = 2;
}
enum ParamAccessType {
PARAM_READ_ONLY = 0;
PARAM_READ_WRITE = 1;
PARAM_WRITE_ONLY = 2;
}
enum ObjAccessType {
OBJ_READ_ONLY = 0;
OBJ_ADD_DELETE = 1;
OBJ_ADD_ONLY = 2;
OBJ_DELETE_ONLY = 3;
}
enum ParamValueType {
PARAM_UNKNOWN = 0;
PARAM_BASE_64 = 1;
PARAM_BOOLEAN = 2;
PARAM_DATE_TIME = 3;
PARAM_DECIMAL = 4;
PARAM_HEX_BINARY = 5;
PARAM_INT = 6;
PARAM_LONG = 7;
PARAM_STRING = 8;
PARAM_UNSIGNED_INT = 9;
PARAM_UNSIGNED_LONG = 10;
}
enum ValueChangeType {
VALUE_CHANGE_UNKNOWN = 0;
VALUE_CHANGE_ALLOWED = 1;
VALUE_CHANGE_WILL_IGNORE = 2;
}
enum CmdType {
CMD_UNKNOWN = 0;
CMD_SYNC = 1;
CMD_ASYNC = 2;
}
}
message GetInstances {
repeated string obj_paths = 1;
bool first_level_only = 2;
}
message GetInstancesResp {
repeated RequestedPathResult req_path_results = 1;
message RequestedPathResult {
string requested_path = 1;
fixed32 err_code = 2;
string err_msg = 3;
repeated CurrInstance curr_insts = 4;
}
message CurrInstance {
string instantiated_obj_path = 1;
map<string, string> unique_keys = 2;
}
}
message GetSupportedProtocol {
string controller_supported_protocol_versions = 1;
}
message GetSupportedProtocolResp {
string agent_supported_protocol_versions = 1;
}
message Add {
bool allow_partial = 1;
repeated CreateObject create_objs = 2;
message CreateObject {
string obj_path = 1;
repeated CreateParamSetting param_settings = 2;
}
message CreateParamSetting {
string param = 1;
string value = 2;
bool required = 3;
}
}
message AddResp {
repeated CreatedObjectResult created_obj_results = 1;
message CreatedObjectResult {
string requested_path = 1;
OperationStatus oper_status = 2;
message OperationStatus {
oneof oper_status {
OperationFailure oper_failure = 1;
OperationSuccess oper_success = 2;
}
message OperationFailure {
fixed32 err_code = 1;
string err_msg = 2;
}
message OperationSuccess {
string instantiated_path = 1;
repeated ParameterError param_errs = 2;
map<string, string> unique_keys = 3;
}
}
}
message ParameterError {
string param = 1;
fixed32 err_code = 2;
string err_msg = 3;
}
}
message Delete {
bool allow_partial = 1;
repeated string obj_paths = 2;
}
message DeleteResp {
repeated DeletedObjectResult deleted_obj_results = 1;
message DeletedObjectResult {
string requested_path = 1;
OperationStatus oper_status = 2;
message OperationStatus {
oneof oper_status {
OperationFailure oper_failure = 1;
OperationSuccess oper_success = 2;
}
message OperationFailure {
fixed32 err_code = 1;
string err_msg = 2;
}
message OperationSuccess {
repeated string affected_paths = 1;
repeated UnaffectedPathError unaffected_path_errs = 2;
}
}
}
message UnaffectedPathError {
string unaffected_path = 1;
fixed32 err_code = 2;
string err_msg = 3;
}
}
message Set {
bool allow_partial = 1;
repeated UpdateObject update_objs = 2;
message UpdateObject {
string obj_path = 1;
repeated UpdateParamSetting param_settings = 2;
}
message UpdateParamSetting {
string param = 1;
string value = 2;
bool required = 3;
}
}
message SetResp {
repeated UpdatedObjectResult updated_obj_results = 1;
message UpdatedObjectResult {
string requested_path = 1;
OperationStatus oper_status = 2;
message OperationStatus {
oneof oper_status {
OperationFailure oper_failure = 1;
OperationSuccess oper_success = 2;
}
message OperationFailure {
fixed32 err_code = 1;
string err_msg = 2;
repeated UpdatedInstanceFailure updated_inst_failures = 3;
}
message OperationSuccess {
repeated UpdatedInstanceResult updated_inst_results = 1;
}
}
}
message UpdatedInstanceFailure {
string affected_path = 1;
repeated ParameterError param_errs = 2;
}
message UpdatedInstanceResult {
string affected_path = 1;
repeated ParameterError param_errs = 2;
map<string, string> updated_params = 3;
}
message ParameterError {
string param = 1;
fixed32 err_code = 2;
string err_msg = 3;
}
}
message Operate {
string command = 1;
string command_key = 2;
bool send_resp = 3;
map<string, string> input_args = 4;
}
message OperateResp {
repeated OperationResult operation_results = 1;
message OperationResult {
string executed_command = 1;
oneof operation_resp {
string req_obj_path = 2;
OutputArgs req_output_args = 3;
CommandFailure cmd_failure = 4;
}
message OutputArgs {
map<string, string> output_args = 1;
}
message CommandFailure {
fixed32 err_code = 1;
string err_msg = 2;
}
}
}
message Notify {
string subscription_id = 1;
bool send_resp = 2;
oneof notification {
Event event = 3;
ValueChange value_change = 4;
ObjectCreation obj_creation = 5;
ObjectDeletion obj_deletion = 6;
OperationComplete oper_complete = 7;
OnBoardRequest on_board_req = 8;
}
message Event {
string obj_path = 1;
string event_name = 2;
map<string, string> params = 3;
}
message ValueChange {
string param_path = 1;
string param_value = 2;
}
message ObjectCreation {
string obj_path = 1;
map<string, string> unique_keys = 2;
}
message ObjectDeletion {
string obj_path = 1;
}
message OperationComplete {
string obj_path = 1;
string command_name = 2;
string command_key = 3;
oneof operation_resp {
OutputArgs req_output_args = 4;
CommandFailure cmd_failure = 5;
}
message OutputArgs {
map<string, string> output_args = 1;
}
message CommandFailure {
fixed32 err_code = 1;
string err_msg = 2;
}
}
message OnBoardRequest {
string oui = 1;
string product_class = 2;
string serial_number = 3;
string agent_supported_protocol_versions = 4;
}
}
message NotifyResp {
string subscription_id = 1;
}

1075
backend/services/ws/internal/usp_record/usp-record-1-2.pb.go Executable file
View File

File diff suppressed because it is too large Load Diff

134
backend/services/ws/internal/usp_record/usp-record-1-2.proto Executable file
View File

@ -0,0 +1,134 @@
syntax = "proto3";
//**************************************************************************
// TR-369 USP Record Protocol Buffer Schema
//
// Copyright (c) 2017-2018, Broadband Forum
//
// The undersigned members have elected to grant the copyright to
// their contributed material used in this software:
// Copyright (c) 2017-2018 ARRIS Enterprises, LLC.
//
// Redistribution and use in source and binary forms, with or
// without modification, are permitted provided that the following
// conditions are met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following
// disclaimer in the documentation and/or other materials
// provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its
// contributors may be used to endorse or promote products
// derived from this software without specific prior written
// permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
// CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
// NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
// ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// The above license is used as a license under copyright only.
// Please reference the Forum IPR Policy for patent licensing terms
// <https://www.broadband-forum.org/ipr-policy>.
//
// Any moral rights which are necessary to exercise under the above
// license grant are also deemed granted under this license.
//
//
// | Version | Name | Date |
// | TR-369 1.0.0 | User Services Platform | APR, 2018 |
// | TR-369 1.0.1 | User Services Platform | JUN, 2018 |
// | TR-369 1.0.2 | User Services Platform | OCT, 2018 |
// | TR-369 1.1 | User Services Platform | SEP, 2019 |
//
// BBF software release registry: http://www.broadband-forum.org/software
//**************************************************************************
package usp_record;
option go_package="./usp-record";
message Record {
string version = 1;
string to_id = 2;
string from_id = 3;
PayloadSecurity payload_security = 4;
bytes mac_signature = 5; //MAC or Signature
bytes sender_cert = 6;
oneof record_type {
NoSessionContextRecord no_session_context = 7;
SessionContextRecord session_context = 8;
WebSocketConnectRecord websocket_connect = 9;
MQTTConnectRecord mqtt_connect = 10;
STOMPConnectRecord stomp_connect = 11;
DisconnectRecord disconnect = 12;
}
enum PayloadSecurity {
PLAINTEXT = 0;
TLS12 = 1;
}
}
message NoSessionContextRecord {
bytes payload = 2;
}
message SessionContextRecord {
uint64 session_id = 1;
uint64 sequence_id = 2;
uint64 expected_id = 3;
uint64 retransmit_id = 4;
PayloadSARState payload_sar_state = 5;
PayloadSARState payloadrec_sar_state = 6;
repeated bytes payload = 7;
enum PayloadSARState {
NONE = 0; //No segmentation
BEGIN = 1; //Begin segmentation
INPROCESS = 2; //Segmentation in process
COMPLETE = 3; //Segmentation is complete
}
}
message WebSocketConnectRecord {
// An empty message
}
message MQTTConnectRecord {
MQTTVersion version = 1;
string subscribed_topic = 2;
enum MQTTVersion {
V3_1_1 = 0; // Represents MQTT v3.1.1, a.k.a. v4 in the MQTT Spec
V5 = 1;
}
}
message STOMPConnectRecord {
STOMPVersion version = 1;
string subscribed_destination = 2;
enum STOMPVersion {
V1_2 = 0;
}
}
message DisconnectRecord {
string reason = 1;
fixed32 reason_code = 2;
}

222
backend/services/ws/internal/ws/handler/client.go Normal file
View File

@ -0,0 +1,222 @@
package handler
import (
"log"
"net/http"
"strings"
"time"
"github.com/OktopUSP/oktopus/ws/internal/usp_record"
"github.com/gorilla/websocket"
"google.golang.org/protobuf/proto"
)
const (
// Time allowed to write a message to the peer.
writeWait = 30 * time.Second
// Time allowed to read the next pong message from the peer.
pongWait = 10 * time.Second
// Send pings to peer with this period. Must be less than pongWait.
pingPeriod = (pongWait * 9) / 10
// Maximum message size allowed from peer.
//maxMessageSize = 512
// Websockets version of the protocol
wsVersion = "13"
// USP specification version
uspVersion = "v1.usp"
)
var (
newline = []byte{'\n'}
space = []byte{' '}
upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool { return true },
}
)
// Client is a middleman between the websocket connection and the hub.
type Client struct {
hub *Hub
//Websockets client endpoint id, eid follows usp specification
eid string
// The websocket connection.
conn *websocket.Conn
// Buffered channel of outbound messages.
send chan message
}
// readPump pumps messages from the websocket connection to the hub.
//
// The application runs readPump in a per-connection goroutine. The application
// ensures that there is at most one reader on a connection by executing all
// reads from this goroutine.
// cEID = controller endpoint id
func (c *Client) readPump(cEID string) {
defer func() {
c.hub.unregister <- c
c.conn.Close()
}()
//c.conn.SetReadLimit(maxMessageSize)
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
for {
_, data, err := c.conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
log.Printf("error: %v", err)
}
break
}
message := constructMsg(cEID, c.eid, data)
c.hub.broadcast <- message
}
}
func constructMsg(eid string, from string, data []byte) message {
if eid == "" {
var record usp_record.Record
err := proto.Unmarshal(data, &record)
if err != nil {
log.Println(err)
}
eid = record.ToId
}
return message{
eid: eid,
from: from,
data: data,
msgType: websocket.BinaryMessage,
}
}
// writePump pumps messages from the hub to the websocket connection.
//
// A goroutine running writePump is started for each connection. The
// application ensures that there is at most one writer to a connection by
// executing all writes from this goroutine.
func (c *Client) writePump() {
ticker := time.NewTicker(pingPeriod)
defer func() {
ticker.Stop()
c.conn.Close()
}()
for {
select {
case message, ok := <-c.send:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok {
// The hub closed the channel.
log.Println("The hub closed the channel of", c.eid)
c.conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.conn.NextWriter(message.msgType)
if err != nil {
return
}
w.Write(message.data)
// Add queued messages to the current websocket message.
n := len(c.send)
for i := 0; i < n; i++ {
w.Write(newline)
send := <-c.send
w.Write(send.data)
}
if err := w.Close(); err != nil {
return
}
case <-ticker.C:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
// Handle USP Controller events
func ServeController(w http.ResponseWriter, r *http.Request, token, cEID string, authEnable bool) {
if authEnable {
recv_token := r.URL.Query().Get("token")
if recv_token != token {
w.WriteHeader(http.StatusUnauthorized)
w.Write([]byte("Unauthorized"))
return
}
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Println(err)
return
}
client := &Client{hub: hub, eid: cEID, conn: conn, send: make(chan message)}
client.hub.register <- client
go client.writePump()
go client.readPump("")
}
// Handle USP Agent events, cEID = controller endpoint id
func ServeAgent(w http.ResponseWriter, r *http.Request, cEID string) {
//TODO: find out a way to authenticate agents
header := http.Header{
"Sec-Websocket-Protocol": {uspVersion},
"Sec-Websocket-Version": {wsVersion},
}
deviceid := extractDeviceId(r.Header)
if deviceid == "" {
w.WriteHeader(http.StatusBadRequest)
log.Println("Device id not found")
w.Write([]byte("Device id not found"))
return
}
conn, err := upgrader.Upgrade(w, r, header)
if err != nil {
log.Println(err)
return
}
client := &Client{hub: hub, eid: deviceid, conn: conn, send: make(chan message)}
client.hub.register <- client
// Allow collection of memory referenced by the caller by doing all work in
// new goroutines.
go client.writePump()
//TODO: get cEID from device message toId record field (must refact nice part of the code for this to be dynamic)
go client.readPump(cEID)
}
// gets device id from websockets header
func extractDeviceId(header http.Header) string {
// Header must be like that: bbf-usp-protocol; eid="<endpoint-id>" <endpoint-id> is the same ar the record.FromId/record.ToId
// log.Println("Header sec-websocket-extensions:", header.Get("sec-websocket-extensions"))
wsHeaderExtension := header.Get("sec-websocket-extensions")
// Split the input string by double quotes
deviceid := strings.Split(wsHeaderExtension, "\"")
if len(deviceid) < 2 {
return ""
}
return deviceid[1]
}

128
backend/services/ws/internal/ws/handler/hub.go Normal file
View File

@ -0,0 +1,128 @@
package handler
import (
"encoding/json"
"log"
"github.com/gorilla/websocket"
)
// Keeps the content and the destination of a websockets message
type message struct {
// Websockets client endpoint id, eid follows usp specification.
// This field is needed for us to know which agent or controller
// the message is intended to be delivered to.
eid string
data []byte
msgType int
from string
}
// Hub maintains the set of active clients and broadcasts messages to the
// clients.
type Hub struct {
// Registered clients.
clients map[string]*Client
// Inbound messages from the clients.
broadcast chan message
// Register requests from the clients.
register chan *Client
// Unregister requests from clients.
unregister chan *Client
}
const (
OFFLINE = "0"
ONLINE = "1"
)
type deviceStatus struct {
Eid string
Status string
}
// Global hub instance
var hub *Hub
// Controller Endpoint ID
var ceid string
func InitHandlers(eid string) {
ceid = eid
log.Println("New hub, Controller eid:", ceid)
hub = newHub()
hub.run()
}
func newHub() *Hub {
return &Hub{
broadcast: make(chan message),
register: make(chan *Client),
unregister: make(chan *Client),
clients: make(map[string]*Client),
}
}
func (h *Hub) run() {
for {
select {
case client := <-h.register:
// register new eid
h.clients[client.eid] = client
log.Printf("New client connected: %s", client.eid)
case client := <-h.unregister:
// verify if eid exists
if _, ok := h.clients[client.eid]; ok {
// delete eid from map of connections
delete(h.clients, client.eid)
// close client messages receiving channel
close(client.send)
}
log.Println("Disconnected client", client.eid)
data, _ := json.Marshal(deviceStatus{client.eid, OFFLINE})
msg := message{
from: "WS server",
eid: ceid,
data: data,
msgType: websocket.TextMessage,
}
log.Printf("%++v", msg)
//TODO: set this snippet of code as a function to avoid repetition
if c, ok := h.clients[msg.eid]; ok {
select {
// send message to receiver client
case c.send <- msg:
log.Printf("Sent a message %s --> %s", msg.from, msg.eid)
default:
// in case the msg sending fails, close the client connection
// because it means that the client is no longer active
log.Printf("Failed to send a msg to %s, disconnecting client...", msg.eid)
close(c.send)
delete(h.clients, c.eid)
}
}
case message := <-h.broadcast: //TODO: ver a conexão de quem está enviando
log.Println("send message to", message.eid)
// verify if eid exists
if c, ok := h.clients[message.eid]; ok {
select {
// send message to receiver client
case c.send <- message:
log.Printf("Sent a message %s --> %s", message.from, message.eid)
default:
// in case the message sending fails, close the client connection
// because it means that the client is no longer active
log.Printf("Failed to send a message to %s, disconnecting client...", message.eid)
close(c.send)
delete(h.clients, c.eid)
}
} else {
//TODO: create queue for receiver while the client is not online
log.Printf("Message receiver not found: %s", message.eid)
}
}
}
}

67
backend/services/ws/internal/ws/ws.go
View File

@ -1,53 +1,42 @@
package ws
// Websockets server implementation inspired by https://github.com/gorilla/websocket/tree/main/examples/chat
import (
"log"
"net/http"
"github.com/OktopUSP/oktopus/ws/internal/config"
"github.com/OktopUSP/oktopus/ws/internal/ws/handler"
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool { return true },
}
// Starts New Websockets Server
func StartNewServer(c config.Config) {
// Initialize handlers of websockets events
go handler.InitHandlers(c.ControllerEID)
func StartNewServer() {
r := mux.NewRouter()
r.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
header := http.Header{
"Sec-Websocket-Protocol": {"v1.usp"},
"Sec-Websocket-Version": {"13"},
}
conn, err := upgrader.Upgrade(w, r, header)
if err != nil {
log.Println(err)
}
for {
messageType, p, err := conn.ReadMessage()
if err != nil {
log.Println(err)
return
}
log.Println(string(p))
if err := conn.WriteMessage(messageType, p); err != nil {
log.Println(err)
return
}
}
r.HandleFunc("/ws/agent", func(w http.ResponseWriter, r *http.Request) {
handler.ServeAgent(w, r, c.ControllerEID)
})
r.HandleFunc("/ws/controller", func(w http.ResponseWriter, r *http.Request) {
handler.ServeController(w, r, c.Token, c.ControllerEID, c.Auth)
})
log.Println("Websockets server running")
err := http.ListenAndServe(":8080", r)
if err != nil {
log.Fatal("ListenAndServe: ", err)
}
go func() {
if c.Tls {
log.Println("Websockets server running with TLS")
err := http.ListenAndServeTLS(c.Port, "cert.pem", "key.pem", r)
if err != nil {
log.Fatal("ListenAndServeTLS: ", err)
}
} else {
log.Println("Websockets server running")
err := http.ListenAndServe(c.Port, r)
if err != nil {
log.Fatal("ListenAndServe: ", err)
}
}
}()
}

19
devops/docker-compose.yaml
View File

@ -7,12 +7,13 @@ services:
stdin_open: true
volumes:
- ../:/app/oktopus
command: bash -c "cd /app/oktopus/backend/services/controller && go run cmd/oktopus/main.go -mongo mongodb://172.16.235.2:27017 -a 172.16.235.4 -p 1883"
command: bash -c "cd /app/oktopus/backend/services/controller && go run cmd/oktopus/main.go -mongo mongodb://172.16.235.2:27017 -a 172.16.235.4 -p 1883 -ws_addr 172.16.235.8"
ports:
- 8000:8000
depends_on:
- mongodb
- mochi
- websockets
networks:
usp_network:
ipv4_address: 172.16.235.7
@ -38,7 +39,7 @@ services:
stdin_open: true
volumes:
- ../:/app/oktopus
command: bash -c "cd /app/oktopus/backend/services/mochi/cmd/ && go run main.go -redis 'redis:6379'"
command: bash -c "cd /app/oktopus/backend/services/mochi/cmd/ && go run main.go -redis 'redis:6379' -info :8081"
ports:
- 1883:1883
depends_on:
@ -47,6 +48,20 @@ services:
usp_network:
ipv4_address: 172.16.235.4
websockets:
image: 'golang:1.21.6'
container_name: websockets
tty: true
stdin_open: true
volumes:
- ../:/app/oktopus
command: bash -c "cd /app/oktopus/backend/services/ws/ && go run cmd/main.go"
ports:
- 8080:8080
networks:
usp_network:
ipv4_address: 172.16.235.8
socketio:
image: 'node:16.20.2'
container_name: socketio

12
frontend/src/sections/overview/overview-latest-orders.js
View File

@ -22,17 +22,17 @@ import { useRouter } from 'next/router';
const statusMap = {
1: 'warning',
0: 'success',
2: 'error'
2: 'success',
0: 'error'
};
const status = (s)=>{
if (s == 0){
return "Online"
return "Offline"
} else if (s == 1){
return "Associating"
}else if (s==2){
return "Offline"
return "Online"
}else {
return "Unknown"
}
@ -99,9 +99,9 @@ export const OverviewLatestOrders = (props) => {
<TableCell>
<SvgIcon
fontSize="small"
sx={{cursor: order.Status !== 2 && 'pointer'}}
sx={{cursor: order.Status == 2 && 'pointer'}}
onClick={()=>{
if (order.Status !== 2){
if (order.Status == 2){
router.push("devices/"+order.SN+"/discovery")
}
}