ctucx.git: smartied

add accessToken

diff --git a/config.json b/config.json
@@ -96,9 +96,10 @@
 	"httpPort": 5000,
 	"tcpPort": 5001,
 	"wsPort": 5002,
-	"modbusAddr": "127.0.0.1",
-	"modbusPort": 5502,
-	"lacrosseAddr": "127.0.0.1",
+	"modbusAddr": "192.168.1.1",
+	"modbusPort": 502,
+	"lacrosseAddr": "192.168.1.1",
 	"lacrossePort": 2342,
-	"powermeterUpdateIntervalSec": 20
+	"powermeterUpdateIntervalSec": 20,
+	"accessToken": "passwort"
 }

diff --git a/src/frontend_http.nim b/src/frontend_http.nim
@@ -6,11 +6,16 @@ import sequtils
 import json
 import vars
 
-proc processHttpClient(req: Request) {.async.} =
+proc processHttpClient(req: Request) {.async,gcsafe.} =
   if req.reqMethod == HttpGet:
-    await req.respond(Http200, $(%* server.state))
+    if req.headers.hasKey("Authorization") and req.headers["Authorization"] == "Bearer " & server.config.accessToken:
+      await req.respond(Http200, $(%* server.state))
+    else:
+      await req.respond(Http401, "401 Unauthorized")
   elif req.reqMethod == HttpPost:
-    await req.respond(Http200, await tryHandleRequest(req.body))
+    let client = Client(id: lastClientId, sendProc: proc (msg: string) {.async.} = await req.respond(Http200, msg))
+    lastClientId += 1
+    await client.tryHandleRequest(req.body)
   else:
     await req.respond(Http405, "405 Method Not Allowed")
 

diff --git a/src/frontend_tcp.nim b/src/frontend_tcp.nim
@@ -6,39 +6,25 @@ import sequtils
 import json
 import vars
 
-proc broadcast(msg: string) {.locks: true.}=
-  for client in tcpClients:
-    try:
-      asyncCheck client.send(msg)
-    except:
-      client.close()
-      tcpClients.keepIf(proc(x: AsyncSocket): bool = x != client)
-
-proc processTcpClient(client: AsyncSocket) {.async.} =
+proc processTcpClient(sock: AsyncSocket) {.async.} =
+  let client = Client(id: lastClientId, sendProc: proc (msg: string) {.async.} = await sock.send(msg & '\n'))
+  lastClientId += 1
   try:
-    await client.send($(%*server.state))
     while true:
-      let req = await client.recvLine()
+      let req = await sock.recvLine()
       if req == "":
-        client.close()
+        sock.close()
         break
-
-      let resp = await tryHandleRequest(req)
-      await client.send(resp & '\n')
+      await client.tryHandleRequest(req)
   except:
-    client.close()
-    tcpClients.keepIf(proc(x: AsyncSocket): bool = x != client)
+    sock.close()
 
 proc serveTcp*() {.async.} =
-  registerBroadcastHandler(broadcast)
-  tcpClients = @[]
-
   var socket = newAsyncSocket()
   socket.setSockOpt(OptReuseAddr, true)
   socket.bindAddr(Port(server.config.tcpPort))
   socket.listen()
 
   while true:
-    let client = await socket.accept()
-    tcpClients.add(client)
-    asyncCheck processTcpClient(client)
+    let sock = await socket.accept()
+    asyncCheck processTcpClient(sock)

diff --git a/src/frontend_ws.nim b/src/frontend_ws.nim
@@ -7,20 +7,13 @@ import sequtils
 import json
 import vars
 
-proc broadcast(msg: string) {.locks: true.} =
-  for client in wsClients:
-    try:
-      asyncCheck client.send(msg)
-    except:
-      client.close()
-      wsClients.keepIf(proc(x: WebSocket): bool = x != client)
-
-proc processWsClient(req: Request) {.async.} =
+proc processWsClient(req: Request) {.async,gcsafe.} =
   var ws: WebSocket
+  var client: Client
   try:
     ws = await newWebsocket(req)
-    wsClients.add(ws)
-    await ws.send($(%*server.state))
+    lastClientId += 1
+    client = Client(id: lastClientId, sendProc: proc (msg: string) {.async.} = await ws.send(msg))
   except:
     asyncCheck req.respond(Http404, "404")
     return

@@ -28,19 +21,10 @@ proc processWsClient(req: Request) {.async.} =
   try:
     while true:
       let req = await ws.receiveStrPacket()
-      let resp = await tryHandleRequest(req)
-      await ws.send(resp)
+      await client.tryHandleRequest(req)
   except:
     ws.close()
-    wsClients.keepIf(proc(x: WebSocket): bool = x != ws)
 
 proc serveWs*() {.async.} =
-  registerBroadcastHandler(broadcast)
-  wsClients = @[]
-
-  addTimer(1000, false, proc (fd: AsyncFD): bool {.gcsafe.} =
-      broadcast("")
-    )
-
   var httpServer = newAsyncHttpServer()
   await httpServer.serve(Port(server.config.wsPort), processWsClient)

diff --git a/src/smartied.nim b/src/smartied.nim
@@ -13,7 +13,6 @@ import util
 import tables
 
 server = Server(config: parseJson(readFile("./config.json")).to(Config))
-echo server.state
 
 initUtil()
 initModbus()

diff --git a/src/types.nim b/src/types.nim
@@ -1,4 +1,5 @@
 import asyncnet
+import asyncdispatch
 import json
 import tables
 import sequtils

@@ -30,6 +31,7 @@ type Config* = object
   powermeterUpdateIntervalSec*: uint
   devices*: Table[string, DeviceConfig]
   clientConfigs*: Table[string, JsonNode]
+  accessToken*: string
 
 type DeviceState* = object
   case type*: DeviceType

@@ -56,9 +58,11 @@ type Server* = object
 type ActionType* = enum
   SetRelayAction,
   ToggleRelayAction,
-  GetClientConfigAction
+  GetClientConfigAction,
+  SetSubscriptionStateAction
 
 type Action* = object
+  accessToken*: string
   case type*: ActionType
   of SetRelayAction:
     setRelayBoard*: string

@@ -69,6 +73,8 @@ type Action* = object
     toggleRelay*: uint8
   of GetClientConfigAction:
     configName*: string
+  of SetSubscriptionStateAction:
+    subscribed*: bool
 
 type ResponseStatus* = enum
   Err,

@@ -86,3 +92,6 @@ type LacrosseMessage* = object
 
 type modbus* = pointer
 
+type Client* = object
+  sendProc*: proc (msg: string): Future[void] {.gcsafe.}
+  id*: int

diff --git a/src/util.nim b/src/util.nim
@@ -5,21 +5,39 @@ import types
 import tables
 import backend_relayboard
 import vars
+import sequtils
 
-proc initUtil*() =
-  broadcastHandlers = @[]
-
-proc registerBroadcastHandler*(handler: proc (msg: string) {.gcsafe, locks: true.}) =
-  broadcastHandlers.add(handler)
+proc trySend*(client: Client, msg: string) {.async.} =
+  try:
+    await client.sendProc(msg)
+  except:
+    let e = getCurrentException()
+    echo("error while sending data to client: ", e.msg)
+    echo("removing client ", client.id)
+    echo clients.map(proc(x: Client): int = x.id)
+    clients.keepIf(proc(x: Client): bool = x.id != client.id)
+    echo clients.map(proc(x: Client): int = x.id)
 
 proc broadcast*(msg: string) =
-  for broadcastHandler in broadcastHandlers:
-    broadcastHandler(msg)
+  for client in clients.filter(proc (x: Client): bool = true):
+    asyncCheck client.trySend(msg)
 
-proc handleRequest*(req: string): Future[JsonNode] {.async.} =
+proc initUtil*() =
+  addTimer(1000, false, proc (fd: AsyncFD): bool {.gcsafe.} =
+      broadcast("")
+    )
+
+  lastClientId = 1
+  clients = @[]
+
+proc handleRequest*(client: Client, req: string): Future[JsonNode] {.async.} =
   let action = parseJson(req).to(Action)
 
-  if action.type == SetRelayAction:
+  if action.accessToken != server.config.accessToken:
+    raise newException(OsError, "invalid accessToken")
+
+  case action.type
+  of SetRelayAction:
     let config = server.config.devices[action.setRelayBoard]
 
     server.state[action.setRelayBoard].relays[action.setRelay] = action.setValue

@@ -27,7 +45,7 @@ proc handleRequest*(req: string): Future[JsonNode] {.async.} =
 
     broadcast($(%*server.state))
     return JsonNode()
-  elif action.type == ToggleRelayAction:
+  of ToggleRelayAction:
     let config = server.config.devices[action.toggleRelayBoard]
 
     server.state[action.toggleRelayBoard].relays[action.toggleRelay] = not server.state[action.toggleRelayBoard].relays[action.toggleRelay]

@@ -35,16 +53,31 @@ proc handleRequest*(req: string): Future[JsonNode] {.async.} =
 
     broadcast($(%*server.state))
     return JsonNode()
-  elif action.type == GetClientConfigAction:
+  of GetClientConfigAction:
     let clientConfig: JsonNode = server.config.clientConfigs[action.configName]
     return clientConfig
+  of SetSubscriptionStateAction:
+    if action.subscribed:
+      echo("adding client ", client.id)
+      echo clients.map(proc(x: Client): int = x.id)
+      clients.keepIf(proc(x: Client): bool = x.id != client.id)
+      echo clients.map(proc(x: Client): int = x.id)
+      clients.add(client)
+      echo clients.map(proc(x: Client): int = x.id)
+      await client.sendProc($(%*server.state))
+    else:
+      echo("removing client ", client.id)
+      echo clients.map(proc(x: Client): int = x.id)
+      clients.keepIf(proc(x: Client): bool = x.id != client.id)
+      echo clients.map(proc(x: Client): int = x.id)
+    return JsonNode()
 
-proc tryHandleRequest*(req: string): Future[string] {.async.} =
+proc tryHandleRequest*(client: Client, req: string): Future[void] {.async.} =
   GC_fullCollect()
   var resp: Response
   try:
-    resp = Response(status: Ok, data: await handleRequest(req))
+    resp = Response(status: Ok, data: await client.handleRequest(req))
   except:
     let e = getCurrentException()
     resp = Response(status: Err, data: newJString(e.msg))
-  return $(%*resp)
+  await client.sendProc($(%*resp))

diff --git a/src/vars.nim b/src/vars.nim
@@ -2,8 +2,7 @@ import types
 import ws
 import asyncnet
 
-var wsClients* {.threadvar.}: seq[WebSocket]
-var tcpClients* {.threadvar.}: seq[AsyncSocket]
+var clients* {.threadvar.}: seq[Client]
 var server* {.threadvar.}: Server
 var mb* {.threadvar.}: modbus
-var broadcastHandlers* {.threadvar.}: seq[proc (msg: string) {.gcsafe.}]
+var lastClientId* {.threadvar.}: int