sdnr/wt/odlux/framework/src/services/notificationService.ts

   1 import * as X2JS from 'x2js';\r
   2 \r
   3 const socketUrl = [location.protocol === 'https:' ? 'wss://' : 'ws://', 'admin', ':', 'admin', '@', location.hostname, ':', location.port, '/websocket'].join('');\r
   4 const subscriptions: { [scope: string]: SubscriptionCallback[] } = { };\r
   5 \r
   6 export interface IFormatedMessage {\r
   7   notifType: string | null;\r
   8   time: string;\r
   9 }\r
  10 \r
  11 export type SubscriptionCallback<TMessage extends IFormatedMessage = IFormatedMessage> = (msg: TMessage) => void;\r
  12 \r
  13 function formatData(event: MessageEvent) : IFormatedMessage | undefined {\r
  14 \r
  15   var x2js = new X2JS();\r
  16   var jsonObj: { [key: string]: IFormatedMessage } = x2js.xml2js(event.data);\r
  17   if (jsonObj && typeof (jsonObj) === 'object') {\r
  18 \r
  19     const notifType = Object.keys(jsonObj)[0];\r
  20     const formated = jsonObj[notifType];\r
  21     formated.notifType = notifType ;\r
  22     formated.time = new Date().toISOString();\r
  23     return formated;\r
  24   }\r
  25   return undefined;\r
  26 \r
  27 }\r
  28 \r
  29 export function subscribe<TMessage extends IFormatedMessage = IFormatedMessage>(scope: string | string[], callback: SubscriptionCallback<TMessage>): Promise<boolean> {\r
  30   return socketReady.then((notificationSocket) => {\r
  31     const scopes = scope instanceof Array ? scope : [scope];\r
  32 \r
  33     // send all new scopes to subscribe\r
  34     const newScopesToSubscribe: string[] = scopes.reduce((acc: string[], cur: string) => {\r
  35       const currentCallbacks = subscriptions[cur];\r
  36       if (currentCallbacks) {\r
  37         if (!currentCallbacks.some(c => c === callback)) {\r
  38           currentCallbacks.push(callback);\r
  39         }\r
  40       } else {\r
  41         subscriptions[cur] = [callback];\r
  42         acc.push(cur);\r
  43       }\r
  44       return acc;\r
  45     }, []);\r
  46 \r
  47     if (newScopesToSubscribe.length === 0) {\r
  48       return true;\r
  49     }\r
  50 \r
  51     // send a subscription to all active scopes\r
  52     const scopesToSubscribe = Object.keys(subscriptions);\r
  53     if (notificationSocket.readyState === notificationSocket.OPEN) {\r
  54       const data = {\r
  55         'data': 'scopes',\r
  56         'scopes': scopesToSubscribe\r
  57       };\r
  58       notificationSocket.send(JSON.stringify(data));\r
  59       return true;\r
  60     }\r
  61     return false;\r
  62   });\r
  63 }\r
  64 \r
  65 export function unsubscribe<TMessage extends IFormatedMessage = IFormatedMessage>(scope: string | string[], callback: SubscriptionCallback<TMessage>): Promise<boolean> {\r
  66   return socketReady.then((notificationSocket) => {\r
  67     const scopes = scope instanceof Array ? scope : [scope];\r
  68     scopes.forEach(s => {\r
  69       const callbacks = subscriptions[s];\r
  70       const index = callbacks && callbacks.indexOf(callback);\r
  71       if (index > -1) {\r
  72         callbacks.splice(index, 1);\r
  73       }\r
  74       if (callbacks.length === 0) {\r
  75         subscriptions[s] === undefined;\r
  76       }\r
  77     });\r
  78 \r
  79     // send a subscription to all active scopes\r
  80     const scopesToSubscribe = Object.keys(subscriptions);\r
  81     if (notificationSocket.readyState === notificationSocket.OPEN) {\r
  82       const data = {\r
  83         'data': 'scopes',\r
  84         'scopes': scopesToSubscribe\r
  85       };\r
  86       notificationSocket.send(JSON.stringify(data));\r
  87       return true;\r
  88     }\r
  89     return false;\r
  90   });\r
  91 }\r
  92 \r
  93 const connect = (): Promise<WebSocket> => {\r
  94   return new Promise((resolve, reject) => {\r
  95     const notificationSocket = new WebSocket(socketUrl);\r
  96 \r
  97     notificationSocket.onmessage = (event) => {\r
  98       // process received event\r
  99       if (typeof event.data === 'string') {\r
 100         const formated = formatData(event);\r
 101         if (formated && formated.notifType) {\r
 102           const callbacks = subscriptions[formated.notifType];\r
 103           if (callbacks) {\r
 104             callbacks.forEach(cb => {\r
 105               // ensure all callbacks will be called\r
 106               try {\r
 107                 return cb(formated);\r
 108               } catch (reason) {\r
 109                 console.error(reason);\r
 110               }\r
 111             });\r
 112           }\r
 113         }\r
 114       }\r
 115     };\r
 116 \r
 117     notificationSocket.onerror = function (error) {\r
 118       console.log("Socket error: " + error);\r
 119       reject("Socket error: " + error);\r
 120     };\r
 121 \r
 122     notificationSocket.onopen = function (event) {\r
 123       console.log("Socket connection opened.");\r
 124       resolve(notificationSocket);\r
 125     };\r
 126 \r
 127     notificationSocket.onclose = function (event) {\r
 128       socketReady = connect();\r
 129     };\r
 130   });\r
 131 }\r
 132 \r
 133 let socketReady = connect();\r
 134 \r
 135 \r
 136 \r
 137 \r