william/th7
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compare commits

base: william:eda649c65f44700b4f78b03e789ca91d039db967
Branches Tags
william:master
..
compare: william:187238c7e4e22d625ec0bd7888b62d085226ffee
Branches Tags
william:master

No commits in common. "eda649c65f44700b4f78b03e789ca91d039db967" and "187238c7e4e22d625ec0bd7888b62d085226ffee" have entirely different histories.
eda649c65f ... 187238c7e4

8 changed files with 153 additions and 68 deletions
Show Stats Expand all files Collapse all files

35
api/api.go
View File

@ -28,7 +28,7 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
adapter.port = cfg.API.Port
adapter.cfg = cfg
gin.SetMode(gin.ReleaseMode)
//gin.SetMode(gin.ReleaseMode)
adapter.router = gin.New()
@ -37,20 +37,15 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
go manager.start()
// API
v1 := adapter.router.Group("/v1")
v1.GET("/data/channel/:id", adapter.GetChannelByIDHandler)
v1.GET("/data/channels", adapter.GetChannelsHandler)
v1.GET("/data/ratio", adapter.GetRatioHandler)
v1.GET("/data", adapter.GetDataHandler)
adapter.router.GET("/v1/data/channel/:id", adapter.GetChannelByIDHandler)
adapter.router.GET("/v1/data/channels", adapter.GetChannelsHandler)
adapter.router.GET("/v1/data/ratio", adapter.GetRatioHandler)
adapter.router.GET("/v1/data", adapter.GetDataHandler)
v1.GET("/config/channel/:id", adapter.GetConfigChannelByIdHandler)
v1.GET("/config/channels", adapter.GetConfigChannelsHandler)
v1.GET("/config/device", adapter.GetConfigDeviceHandler)
v1.GET("/config/db", adapter.GetConfigDBHandler)
v1.POST("/config/device", adapter.SetConfigDeviceHandler)
v1.POST("/config/channel/:id", adapter.SetConfigChannelByIdHandler)
v1.POST("/config/db", adapter.SetConfigDBHandler)
adapter.router.GET("/v1/config/channel/:id", adapter.GetConfigChannelByIdHandler)
adapter.router.GET("/v1/config/channels", adapter.GetConfigChannelsHandler)
adapter.router.GET("/v1/config/device", adapter.GetConfigDeviceHandler)
adapter.router.GET("/v1/config/db", adapter.GetConfigDBHandler)
adapter.router.GET("/ws", func(c *gin.Context) {
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
@ -62,14 +57,22 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
socket: conn,
send: make(chan []byte),
manager: manager,
all: false,
channels: false,
config: false,
}
manager.register <- client
go client.writer()
go client.read()
go client.write()
})
// start web socket service that dumps new data to websocket chan
adapter.router.POST("/v1/config/device", adapter.SetConfigDeviceHandler)
adapter.router.POST("/v1/config/channel/:id", adapter.SetConfigChannelByIdHandler)
adapter.router.POST("/v1/config/db", adapter.SetConfigDBHandler)
// start service that dumps new data to websocket chan
go startService(manager, corePort)
// TH7 demo code. html file and javascript

24
api/service.go
View File

@ -10,35 +10,31 @@ import (
)
const (
CheckChannelDur = time.Millisecond * 1200
CheckChannelDur = time.Millisecond * 700
StartupSleepDur = time.Second * 5
)
type Wrapper struct {
type ChannelsWrapper struct {
Channels []core.Channel `json:"channels"`
Ratio core.Ratio `json:"ratio"`
Timestamp string `json:"time"`
}
// Regularly get the newest data and bung it out on ws
func startService(man *Manager, core ports.CorePort) {
var wrap Wrapper
var chwrap ChannelsWrapper
time.Sleep(StartupSleepDur)
for {
wrap.Channels = core.GetChannels()
wrap.Ratio = core.GetRatio()
wrap.Timestamp = time.Now().Format(time.RFC1123)
marshaled, err := json.Marshal(wrap)
chwrap.Channels = core.GetChannels()
marshaled, err := json.Marshal(chwrap)
if err != nil {
log.Println("Error marshaling struct:", wrap)
log.Println("Error marshaling struct:", chwrap.Channels)
}
man.broadcast <- []byte(marshaled)
man.broadcast <- Message{
messageType: CHANNELS,
data: []byte(marshaled),
}
time.Sleep(CheckChannelDur)
}

111
api/websocket.go
View File

@ -1,7 +1,10 @@
package api
import (
"fmt"
"log"
"net/http"
"strings"
"github.com/gorilla/websocket"
)
@ -14,15 +17,31 @@ var upgrader = websocket.Upgrader{
},
}
type MessageType = int
const (
ALL MessageType = iota
CHANNELS
CONFIG
)
type Message struct {
messageType MessageType
data []byte
}
type Client struct {
socket *websocket.Conn
send chan []byte
manager *Manager
channels bool
config bool
all bool
}
type Manager struct {
clients map[*Client]bool
broadcast chan []byte
broadcast chan Message
register chan *Client
unregister chan *Client
}
@ -30,12 +49,50 @@ type Manager struct {
func NewManager() *Manager {
return &Manager{
clients: make(map[*Client]bool),
broadcast: make(chan []byte),
broadcast: make(chan Message),
register: make(chan *Client),
unregister: make(chan *Client),
}
}
func (man *Manager) BroadcastAll(msg []byte) {
for client := range man.clients {
select {
case client.send <- msg:
default:
man.unregister <- client
}
}
}
func (man *Manager) BroadcastChannels(msg []byte) {
for client := range man.clients {
if !client.all && !client.channels {
continue
}
select {
case client.send <- msg:
default:
man.unregister <- client
}
}
}
func (man *Manager) BroadcastConfig(msg []byte) {
for client := range man.clients {
if !client.all && !client.config {
continue
}
select {
case client.send <- msg:
default:
man.unregister <- client
}
}
}
func (man *Manager) start() {
for {
select {
@ -46,25 +103,56 @@ func (man *Manager) start() {
// connection closed
case conn := <-man.unregister:
if _, ok := man.clients[conn]; ok {
fmt.Println("Unregistering:", conn)
close(conn.send)
delete(man.clients, conn)
}
// broadcast a message to all connected clients
case message := <-man.broadcast:
for client := range man.clients {
select {
case client.send <- message:
switch message.messageType {
case ALL:
man.BroadcastAll(message.data)
case CHANNELS:
man.BroadcastChannels(message.data)
case CONFIG:
man.BroadcastConfig(message.data)
}
}
}
}
func (c *Client) read() {
defer func() {
c.manager.unregister <- c
_ = c.socket.Close()
}()
for {
_, message, err := c.socket.ReadMessage()
if err != nil {
c.manager.unregister <- c
_ = c.socket.Close()
break
}
switch strings.ToLower(string(message)) {
case "all":
c.all = true
case "channels":
c.channels = true
case "config":
c.config = true
default:
man.unregister <- client
log.Println("received: ", message)
}
}
}
}
}
func (c *Client) writer() {
func (c *Client) write() {
defer func() {
_ = c.socket.Close()
@ -75,13 +163,12 @@ func (c *Client) writer() {
case message, ok := <-c.send:
if !ok {
_ = c.socket.WriteMessage(websocket.CloseMessage, []byte{})
c.manager.unregister <- c
return
}
err := c.socket.WriteMessage(websocket.TextMessage, message)
if err != nil {
c.manager.unregister <- c
fmt.Println(err)
return
}
}

3
config/config.go
View File

@ -3,7 +3,6 @@ package config
import (
"errors"
"fmt"
"log"
"strings"
"th7/data/config"
"th7/data/thermocouple"
@ -81,7 +80,7 @@ func Load() (config.Config, error) {
tc, err := GetThermocoupleType(v.GetString(head + ".type"))
if err != nil {
log.Printf("%s.type=%s\n", head, v.GetString(head+".type"))
fmt.Printf("%s.type=%s\n", head, v.GetString(head+".type"))
return cfg, err
}

7
db/influxdb.go
View File

@ -4,7 +4,6 @@ import (
"context"
"errors"
"fmt"
"log"
"os"
"strings"
"sync"
@ -85,7 +84,7 @@ func NewInfluxDBAdapter(config config.Config) (*InfluxDBAdapter, error) {
`
_, err = adapter.client.Query(initContext, query)
if err != nil {
log.Println("Error initialising InfluxDB. Check details/network connection.")
fmt.Println("Error initialising InfluxDB. Check details/network connection.")
return &adapter, err
}
@ -125,7 +124,7 @@ func (ad *InfluxDBAdapter) Save(channels []core.Channel) error {
// need at least 1 iteration to bother sending data to influx server
if logged == 0 {
log.Printf("[%s] InfluxDB: nothing to do.\n", time.Now().Format(time.DateTime))
fmt.Printf("[%s] InfluxDB: nothing to do.\n", time.Now().Format(time.DateTime))
return nil
}
@ -137,7 +136,7 @@ func (ad *InfluxDBAdapter) Save(channels []core.Channel) error {
return err
}
log.Printf("[%s] InfluxDB: logged %d channels OK\n", time.Now().Format(time.DateTime), logged)
fmt.Printf("[%s] InfluxDB: logged %d channels OK\n", time.Now().Format(time.DateTime), logged)
return nil
}

5
main.go
View File

@ -1,6 +1,7 @@
package main
import (
"fmt"
"log"
"os"
"os/signal"
@ -54,7 +55,7 @@ func main() {
defer dbPort.Close()
color.Set(color.FgHiRed)
log.Printf("Started on: %s\n", time.Now().Format(time.DateTime))
fmt.Printf("Started on: %s\n", time.Now().Format(time.DateTime))
color.Unset()
// if noweb is false and HTTP/REST API is enabled then start web service
@ -63,7 +64,7 @@ func main() {
go apiPort.Run()
color.Set(color.FgHiGreen)
log.Printf("TH7 API is live on http://localhost:%d/\n", cfg.API.Port)
fmt.Printf("TH7 API is live on http://localhost:%d/\n", cfg.API.Port)
color.Unset()
}

4
pcb/dummy.go
View File

@ -1,7 +1,7 @@
package pcb
import (
"log"
"fmt"
"math/rand"
"th7/data/pcb"
"time"
@ -17,7 +17,7 @@ func NewDummyAdapter() (*DummyAdapter, error) {
}
func (d *DummyAdapter) Deinit() error {
log.Println("dummy pcb adapter deinit")
fmt.Println("dummy pcb adapter deinit")
return nil
}

20
static/data-updater.js
View File

@ -91,16 +91,16 @@ setInterval(updateRead, 1000);
setInterval(updateConfig, 10000);
// websocket demo
const ws = new WebSocket(`ws://${window.location.host}/ws`);
ws.addEventListener("open", (event) => {
const ws = new WebSocket("ws://" + window.location.host + "/ws");
ws.onopen = (e) => {
console.log("Connection open");
});
ws.send("ALL");
};
ws.addEventListener("message", (event) => {
console.log(`Message received: ${event.data}`);
});
ws.onmessage = (e) => {
console.log("msg recv: " + e.data);
};
ws.addEventListener("close", (event) => {
console.log("Connection closed");
});
ws.onclose = (e) => {
console.log("Connection close");
};