SiegfriedSiegert/MQTTListener
1
0
Fork 0
Code Issues Pull Requests Releases 1 Wiki Activity

Write data in database table

Browse Source
This commit is contained in:
Siegfried Siegert 2020-09-02 09:02:30 +02:00
parent 8f7a285289
commit ac3bb1ebe9
7 changed files with 106 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
Config/Config.go
View File

@ -1,32 +0,0 @@
package Config
import (
//"fmt"
"log"
"github.com/pelletier/go-toml"
)
var BrokerAddress = ""
var BrokerPort int64 = 8883
var BrokerClientId = "go_client"
var BrokerUsername = ""
var BrokerPassword = ""
var EnableTLS = true
func ReadConfig() {
config, err := toml.LoadFile("config.toml")
if err != nil {
//fmt.Println("Error ", err.Error())
log.Fatal(err.Error())
} else {
// retrieve data directly
BrokerAddress = config.Get("BROKER.Address").(string)
BrokerPort = config.Get("BROKER.Port").(int64)
BrokerClientId = config.Get("BROKER.ClientId").(string)
BrokerUsername = config.Get("BROKER.Username").(string)
BrokerPassword = config.Get("BROKER.Password").(string)
EnableTLS = config.Get("BROKER.EnableTLS").(bool)
}
}

39
config/config.go Normal file
View File

@ -0,0 +1,39 @@
package config
import (
//"fmt"
"log"
"github.com/pelletier/go-toml"
)
type Config struct {
BrokerAddress string
BrokerPort int64
BrokerClientID string
BrokerUsername string
BrokerPassword string
EnableTLS bool
}
func NewConfig(configFilename string) (*Config, error) {
tomlData, err := toml.LoadFile(configFilename)
conf := Config{"", 8883, "go_client", "", "", true}
if err != nil {
//fmt.Println("Error ", err.Error())
log.Printf("ERROR: NewConfig(): %s", err.Error())
return nil, err
} else {
// retrieve data directly
conf.BrokerAddress = tomlData.Get("BROKER.Address").(string)
conf.BrokerPort = tomlData.Get("BROKER.Port").(int64)
conf.BrokerClientID = tomlData.Get("BROKER.ClientId").(string)
conf.BrokerUsername = tomlData.Get("BROKER.Username").(string)
conf.BrokerPassword = tomlData.Get("BROKER.Password").(string)
conf.EnableTLS = tomlData.Get("BROKER.EnableTLS").(bool)
}
return &conf, nil
}

31
env/env.go vendored Normal file
View File

@ -0,0 +1,31 @@
package env
import (
"mqttListener/config"
"mqttListener/models"
"fmt"
//"log"
"net/http"
)
type Env struct {
DB models.Datastore
Config *config.Config
}
func (env *Env) DevicesIndex(w http.ResponseWriter, r *http.Request) {
if r.Method != "GET" {
http.Error(w, http.StatusText(405), 405)
return
}
devices, err := env.DB.AllDevices()
if err != nil {
http.Error(w, http.StatusText(500), 500)
return
}
for _, device := range devices {
fmt.Fprintf(w, "%s, %s, %s, £%.2f\n", device.ID, device.MAC, device.SN, device.LastMsg)
}
}

2
go.mod
View File

@ -8,4 +8,6 @@ require (
github.com/mmcdole/gofeed v1.0.0
github.com/pelletier/go-toml v1.2.0
github.com/spf13/cobra v1.0.0
golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
golang.org/x/tools v0.0.0-20200831203904-5a2aa26beb65 // indirect
)

35
main.go
View File

@ -1,7 +1,8 @@
package main
import (
"mqttListener/Config"
"mqttListener/config"
"mqttListener/env"
"mqttListener/models"
"mqttListener/mqtt"
@ -15,32 +16,31 @@ import (
"net/http"
)
type Env struct {
db models.Datastore
}
func main() {
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
Config.ReadConfig()
config, err := config.NewConfig("config.toml")
if err != nil {
log.Panic(err)
}
// DEBUG Config
// fmt.Println("BrokerAddress = ", Config.BrokerAddress)
fmt.Println("BrokerAddress = ", config.BrokerAddress)
db, err := models.NewDB("simple.sqlite")
if err != nil {
log.Panic(err)
}
env := &Env{db}
env := &env.Env{db, config}
mqtt.Setup()
mqtt.Setup(env)
mqtt.Connect()
go mqtt.Listen()
http.HandleFunc("/devices", env.devicesIndex)
http.HandleFunc("/devices", env.DevicesIndex)
go http.ListenAndServe(":3000", nil)
fmt.Println("awaiting signal")
@ -49,18 +49,3 @@ func main() {
mqtt.Disconnect()
}
func (env *Env) devicesIndex(w http.ResponseWriter, r *http.Request) {
if r.Method != "GET" {
http.Error(w, http.StatusText(405), 405)
return
}
devices, err := env.db.AllDevices()
if err != nil {
http.Error(w, http.StatusText(500), 500)
return
}
for _, device := range devices {
fmt.Fprintf(w, "%s, %s, %s, £%.2f\n", device.ID, device.MAC, device.SN, device.LastMsg)
}
}

9
models/devices.go
View File

@ -1,5 +1,9 @@
package models
import (
"fmt"
)
type Device struct {
ID string
MAC string
@ -30,12 +34,15 @@ func (db *DB) AllDevices() ([]*Device, error) {
}
func (db *DB) InsertDevice(device *Device) error {
sqlStmt := `INSERT OR REPLACE INTO table (id, mac, sn, lastMsg)
sqlStmt := `INSERT OR REPLACE INTO devices (id, mac, sn, lastMsg)
VALUES($1, $2, $3, $4);`
_, err := db.Exec(sqlStmt, device.ID, device.MAC, device.SN, device.LastMsg)
if err != nil {
return err
}
// DEBUG
fmt.Printf("Sucessfully inserted device with ID %s to database\n", device.ID)
return err
}

32
mqtt/mqtt.go
View File

@ -8,7 +8,7 @@ import (
"strconv"
"strings"
"mqttListener/Config"
"mqttListener/env"
"mqttListener/models"
"regexp"
"time"
@ -22,18 +22,11 @@ import (
URI = mqtt://<user>:<pass>@<server>:<port>
*/
/* TODO: This options are just a suggestion
Options:
[-help] Display help
[-customer] Customer- or project number
[-device] Device number
[-uri <uri>] Broker URI
*/
var client MQTT.Client
var opts *MQTT.ClientOptions
var topic string
var uri url.URL
var ev *env.Env
func Connect() {
client = MQTT.NewClient(opts)
@ -64,17 +57,19 @@ func Listen() {
*/
}
func Setup() {
func Setup(environment *env.Env) {
ev = environment
scheme := "tcp://"
if Config.EnableTLS {
if ev.Config.EnableTLS {
scheme = "ssl://"
}
urlString := scheme
urlString += Config.BrokerUsername + ":" + Config.BrokerPassword + "@"
urlString += Config.BrokerAddress + ":"
urlString += strconv.FormatInt(Config.BrokerPort, 10)
urlString += ev.Config.BrokerUsername + ":" + ev.Config.BrokerPassword + "@"
urlString += ev.Config.BrokerAddress + ":"
urlString += strconv.FormatInt(ev.Config.BrokerPort, 10)
fmt.Println("broker urlString: ", urlString)
@ -89,7 +84,7 @@ func Setup() {
topic = "/ATB/#"
opts = createClientOptions(Config.BrokerClientId, uri)
opts = createClientOptions(ev.Config.BrokerClientID, uri)
}
@ -101,7 +96,7 @@ func createClientOptions(ClientID string, uri *url.URL) *MQTT.ClientOptions {
opts.SetPassword(password)
opts.SetClientID(ClientID)
opts.SetDefaultPublishHandler(f)
if Config.EnableTLS {
if ev.Config.EnableTLS {
tlsConfig := &tls.Config{InsecureSkipVerify: true, ClientAuth: tls.NoClientCert}
opts.SetTLSConfig(tlsConfig)
}
@ -180,4 +175,9 @@ var subscriptionHandler = func(client MQTT.Client, msg MQTT.Message) {
fmt.Printf("Device SN = %s\n", device.SN)
// TODO: store this device in database
err = ev.DB.InsertDevice(&device)
if err != nil {
fmt.Println(err.Error())
return
}
}