Compare commits
3 Commits
187238c7e4
...
eda649c65f
Author | SHA1 | Date | |
---|---|---|---|
eda649c65f | |||
dcd1a03c69 | |||
ea3c096ea9 |
41
api/api.go
41
api/api.go
@ -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
|
||||
|
@ -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)
|
||||
}
|
||||
|
||||
|
115
api/websocket.go
115
api/websocket.go
@ -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
|
||||
}
|
||||
}
|
||||
|
@ -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
|
||||
}
|
||||
|
||||
|
@ -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
|
||||
}
|
||||
|
5
main.go
5
main.go
@ -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()
|
||||
}
|
||||
|
||||
|
@ -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
|
||||
}
|
||||
|
||||
|
@ -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");
|
||||
});
|
||||
|
Loading…
Reference in New Issue
Block a user