feat(general): add heartbeat to websocket control (#460)

* chore: ignore .idea

* feat: add heartbeat event to websocket

* feat: add HeartbeatInterval to websocket config

* feat: send HeartbeatInterval in SYSTEM_INIT message

* feat: add heartbeat logic

* fix: client events is not in WebSocketEvents type

.gitignore

@@ -33,3 +33,6 @@ bin
 
 # Environment files
 *.env
+
+# Code Editors
+.idea

client/src/neko/base.ts

@@ -20,6 +20,7 @@ export interface BaseEvents {
 
 export abstract class BaseClient extends EventEmitter<BaseEvents> {
   protected _ws?: WebSocket
+  protected _ws_heartbeat?: number
   protected _peer?: RTCPeerConnection
   protected _channel?: RTCDataChannel
   protected _timeout?: number
@@ -80,6 +81,11 @@ export abstract class BaseClient extends EventEmitter<BaseEvents> {
       this._timeout = undefined
     }
 
+    if (this._ws_heartbeat) {
+      clearInterval(this._ws_heartbeat)
+      this._ws_heartbeat = undefined
+    }
+
     if (this._ws) {
       // reset all events
       this._ws.onmessage = () => {}

client/src/neko/events.ts

@@ -14,6 +14,9 @@ export const EVENT = {
     DISCONNECT: 'system/disconnect',
     ERROR: 'system/error',
   },
+  CLIENT: {
+    HEARTBEAT: 'client/heartbeat'
+  },
   SIGNAL: {
     OFFER: 'signal/offer',
     ANSWER: 'signal/answer',
@@ -69,6 +72,7 @@ export type Events = typeof EVENT
 
 export type WebSocketEvents =
   | SystemEvents
+  | ClientEvents
   | ControlEvents
   | MemberEvents
   | SignalEvents
@@ -87,6 +91,7 @@ export type ControlEvents =
   | typeof EVENT.CONTROL.KEYBOARD
 
 export type SystemEvents = typeof EVENT.SYSTEM.DISCONNECT
+export type ClientEvents = typeof EVENT.CLIENT.HEARTBEAT
 export type MemberEvents = typeof EVENT.MEMBER.LIST | typeof EVENT.MEMBER.CONNECTED | typeof EVENT.MEMBER.DISCONNECTED
 
 export type SignalEvents =

client/src/neko/index.ts

@@ -134,7 +134,7 @@ export class NekoClient extends BaseClient implements EventEmitter<NekoEvents> {
   /////////////////////////////
   // System Events
   /////////////////////////////
-  protected [EVENT.SYSTEM.INIT]({ implicit_hosting, locks, file_transfer }: SystemInitPayload) {
+  protected [EVENT.SYSTEM.INIT]({ implicit_hosting, locks, file_transfer, heartbeat_interval }: SystemInitPayload) {
     this.$accessor.remote.setImplicitHosting(implicit_hosting)
     this.$accessor.remote.setFileTransfer(file_transfer)
 
@@ -145,6 +145,11 @@ export class NekoClient extends BaseClient implements EventEmitter<NekoEvents> {
         id: locks[resource],
       })
     }
+
+    if (heartbeat_interval > 0) {
+      if (this._ws_heartbeat) clearInterval(this._ws_heartbeat)
+      this._ws_heartbeat = window.setInterval(() => this.sendMessage(EVENT.CLIENT.HEARTBEAT), heartbeat_interval * 1000)
+    }
   }
 
   protected [EVENT.SYSTEM.DISCONNECT]({ message }: SystemMessagePayload) {

client/src/neko/messages.ts

@@ -61,6 +61,7 @@ export interface SystemInitPayload {
   implicit_hosting: boolean
   locks: Record<string, string>
   file_transfer: boolean
+  heartbeat_interval: number
 }
 
 // system/disconnect

server/internal/config/websocket.go

@@ -14,6 +14,8 @@ type WebSocket struct {
 
 	ControlProtection bool
 
+	HeartbeatInterval int
+
 	FileTransferEnabled bool
 	FileTransferPath    string
 }
@@ -39,6 +41,11 @@ func (WebSocket) Init(cmd *cobra.Command) error {
 		return err
 	}
 
+	cmd.PersistentFlags().Int("heartbeat_interval", 120, "heartbeat interval in seconds")
+	if err := viper.BindPFlag("heartbeat_interval", cmd.PersistentFlags().Lookup("heartbeat_interval")); err != nil {
+		return err
+	}
+
 	// File transfer
 
 	cmd.PersistentFlags().Bool("file_transfer_enabled", false, "enable file transfer feature")
@@ -61,6 +68,8 @@ func (s *WebSocket) Set() {
 
 	s.ControlProtection = viper.GetBool("control_protection")
 
+	s.HeartbeatInterval = viper.GetInt("heartbeat_interval")
+
 	s.FileTransferEnabled = viper.GetBool("file_transfer_enabled")
 	s.FileTransferPath = viper.GetString("file_transfer_path")
 	s.FileTransferPath = filepath.Clean(s.FileTransferPath)

server/internal/types/event/events.go

@@ -6,6 +6,10 @@ const (
 	SYSTEM_ERROR      = "system/error"
 )
 
+const (
+	CLIENT_HEARTBEAT = "client/heartbeat"
+)
+
 const (
 	SIGNAL_OFFER     = "signal/offer"
 	SIGNAL_ANSWER    = "signal/answer"

server/internal/types/message/messages.go

@@ -11,10 +11,11 @@ type Message struct {
 }
 
 type SystemInit struct {
-	Event           string            `json:"event"`
+	Event             string            `json:"event"`
-	ImplicitHosting bool              `json:"implicit_hosting"`
+	ImplicitHosting   bool              `json:"implicit_hosting"`
-	Locks           map[string]string `json:"locks"`
+	Locks             map[string]string `json:"locks"`
-	FileTransfer    bool              `json:"file_transfer"`
+	FileTransfer      bool              `json:"file_transfer"`
+	HeartbeatInterval int               `json:"heartbeat_interval"`
 }
 
 type SystemMessage struct {

server/internal/websocket/handler/handler.go

@@ -74,6 +74,11 @@ func (h *MessageHandler) Message(id string, raw []byte) error {
 	}
 
 	switch header.Event {
+	// Client Events
+	case event.CLIENT_HEARTBEAT:
+		// do nothing
+		return nil
+
 	// Signal Events
 	case event.SIGNAL_OFFER:
 		payload := &message.SignalOffer{}

server/internal/websocket/handler/session.go

@@ -6,7 +6,7 @@ import (
 	"m1k1o/neko/internal/types/message"
 )
 
-func (h *MessageHandler) SessionCreated(id string, session types.Session) error {
+func (h *MessageHandler) SessionCreated(id string, heartbeatInterval int, session types.Session) error {
 	// send sdp and id over to client
 	if err := h.signalProvide(id, session); err != nil {
 		return err
@@ -14,10 +14,11 @@ func (h *MessageHandler) SessionCreated(id string, session types.Session) error
 
 	// send initialization information
 	if err := session.Send(message.SystemInit{
-		Event:           event.SYSTEM_INIT,
+		Event:             event.SYSTEM_INIT,
-		ImplicitHosting: h.webrtc.ImplicitControl(),
+		ImplicitHosting:   h.webrtc.ImplicitControl(),
-		Locks:           h.state.AllLocked(),
+		Locks:             h.state.AllLocked(),
-		FileTransfer:    h.state.FileTransferEnabled(),
+		FileTransfer:      h.state.FileTransferEnabled(),
+		HeartbeatInterval: heartbeatInterval,
 	}); err != nil {
 		h.logger.Warn().Str("id", id).Err(err).Msgf("sending event %s has failed", event.SYSTEM_INIT)
 		return err

server/internal/websocket/websocket.go

@@ -111,7 +111,7 @@ func (ws *WebSocketHandler) Start() {
 
 			switch e.Type {
 			case types.SESSION_CREATED:
-				if err := ws.handler.SessionCreated(e.Id, e.Session); err != nil {
+				if err := ws.handler.SessionCreated(e.Id, ws.conf.HeartbeatInterval, e.Session); err != nil {
 					ws.logger.Warn().Str("id", e.Id).Err(err).Msg("session created with and error")
 				} else {
 					ws.logger.Debug().Str("id", e.Id).Msg("session created")