Compare commits

..

3 Commits

Author SHA1 Message Date
eda649c65f use router group 2023-12-13 08:21:30 +00:00
dcd1a03c69 use log 2023-12-13 08:17:16 +00:00
ea3c096ea9 simplify websockets 2023-12-13 08:15:23 +00:00
8 changed files with 68 additions and 153 deletions

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,15 +37,20 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
go manager.start()
// API
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 := 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/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)
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("/ws", func(c *gin.Context) {
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
@ -54,25 +59,17 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
}
client := &Client{
socket: conn,
send: make(chan []byte),
manager: manager,
all: false,
channels: false,
config: false,
socket: conn,
send: make(chan []byte),
manager: manager,
}
manager.register <- client
go client.read()
go client.write()
go client.writer()
})
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
// start web socket service that dumps new data to websocket chan
go startService(manager, corePort)
// TH7 demo code. html file and javascript

View File

@ -10,31 +10,35 @@ import (
)
const (
CheckChannelDur = time.Millisecond * 700
CheckChannelDur = time.Millisecond * 1200
StartupSleepDur = time.Second * 5
)
type ChannelsWrapper struct {
Channels []core.Channel `json:"channels"`
type Wrapper 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 chwrap ChannelsWrapper
var wrap Wrapper
time.Sleep(StartupSleepDur)
for {
chwrap.Channels = core.GetChannels()
marshaled, err := json.Marshal(chwrap)
wrap.Channels = core.GetChannels()
wrap.Ratio = core.GetRatio()
wrap.Timestamp = time.Now().Format(time.RFC1123)
marshaled, err := json.Marshal(wrap)
if err != nil {
log.Println("Error marshaling struct:", chwrap.Channels)
log.Println("Error marshaling struct:", wrap)
}
man.broadcast <- Message{
messageType: CHANNELS,
data: []byte(marshaled),
}
man.broadcast <- []byte(marshaled)
time.Sleep(CheckChannelDur)
}

View File

@ -1,10 +1,7 @@
package api
import (
"fmt"
"log"
"net/http"
"strings"
"github.com/gorilla/websocket"
)
@ -17,31 +14,15 @@ 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
socket *websocket.Conn
send chan []byte
manager *Manager
}
type Manager struct {
clients map[*Client]bool
broadcast chan Message
broadcast chan []byte
register chan *Client
unregister chan *Client
}
@ -49,50 +30,12 @@ type Manager struct {
func NewManager() *Manager {
return &Manager{
clients: make(map[*Client]bool),
broadcast: make(chan Message),
broadcast: make(chan []byte),
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 {
@ -103,56 +46,25 @@ 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:
switch message.messageType {
case ALL:
man.BroadcastAll(message.data)
case CHANNELS:
man.BroadcastChannels(message.data)
case CONFIG:
man.BroadcastConfig(message.data)
for client := range man.clients {
select {
case client.send <- message:
default:
man.unregister <- client
}
}
}
}
}
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:
log.Println("received: ", message)
}
}
}
func (c *Client) write() {
func (c *Client) writer() {
defer func() {
_ = c.socket.Close()
@ -163,12 +75,13 @@ func (c *Client) write() {
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 {
fmt.Println(err)
c.manager.unregister <- c
return
}
}

View File

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

View File

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

View File

@ -1,7 +1,6 @@
package main
import (
"fmt"
"log"
"os"
"os/signal"
@ -55,7 +54,7 @@ func main() {
defer dbPort.Close()
color.Set(color.FgHiRed)
fmt.Printf("Started on: %s\n", time.Now().Format(time.DateTime))
log.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
@ -64,7 +63,7 @@ func main() {
go apiPort.Run()
color.Set(color.FgHiGreen)
fmt.Printf("TH7 API is live on http://localhost:%d/\n", cfg.API.Port)
log.Printf("TH7 API is live on http://localhost:%d/\n", cfg.API.Port)
color.Unset()
}

View File

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

View File

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