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,20 +37,15 @@ func NewGinAdapter(corePort ports.CorePort, cfg config.Config) *GinAdapter {
 	go manager.start()
 
 	// API
-	v1 := adapter.router.Group("/v1")
+	adapter.router.GET("/v1/data/channel/:id", adapter.GetChannelByIDHandler)
-	v1.GET("/data/channel/:id", adapter.GetChannelByIDHandler)
+	adapter.router.GET("/v1/data/channels", adapter.GetChannelsHandler)
-	v1.GET("/data/channels", adapter.GetChannelsHandler)
+	adapter.router.GET("/v1/data/ratio", adapter.GetRatioHandler)
-	v1.GET("/data/ratio", adapter.GetRatioHandler)
+	adapter.router.GET("/v1/data", adapter.GetDataHandler)
-	v1.GET("/data", adapter.GetDataHandler)
 
-	v1.GET("/config/channel/:id", adapter.GetConfigChannelByIdHandler)
+	adapter.router.GET("/v1/config/channel/:id", adapter.GetConfigChannelByIdHandler)
-	v1.GET("/config/channels", adapter.GetConfigChannelsHandler)
+	adapter.router.GET("/v1/config/channels", adapter.GetConfigChannelsHandler)
-	v1.GET("/config/device", adapter.GetConfigDeviceHandler)
+	adapter.router.GET("/v1/config/device", adapter.GetConfigDeviceHandler)
-	v1.GET("/config/db", adapter.GetConfigDBHandler)
+	adapter.router.GET("/v1/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)
@@ -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

api/service.go

@@ -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()
+		chwrap.Channels = core.GetChannels()
-		wrap.Ratio = core.GetRatio()
+		marshaled, err := json.Marshal(chwrap)
-		wrap.Timestamp = time.Now().Format(time.RFC1123)
-
-		marshaled, err := json.Marshal(wrap)
 		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)
 	}
 

api/websocket.go

@@ -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 {
+			switch message.messageType {
-				select {
+			case ALL:
-				case client.send <- message:
+				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) write() {
-}
-
-func (c *Client) writer() {
 
 	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
 			}
 		}

config/config.go

@@ -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
 		}
 

db/influxdb.go

@@ -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
 }

main.go

@@ -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()
 	}
 

pcb/dummy.go

@@ -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
 }
 

static/data-updater.js

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