device_home/signaling_server/gen/service.js

   1 const path = require('path');
   2 const fs = require('fs');
   3 const express = require('express');
   4 const QRCode = require('qrcode');
   5 const Edge = require('./edge');
   6 const SocketTizen = require('./socket-tizen');
   7 const { getMyAddress } = require('./util');
   8
   9 const TAG = 'service.js';
  10
  11 const app = express();
  12
  13 const options = {
  14   key: fs.readFileSync(path.resolve(__dirname, 'key.pem')),
  15   cert: fs.readFileSync(path.resolve(__dirname, 'cert.pem'))
  16 };
  17
  18 console.log(TAG, `platform : ${process.platform}`);
  19
  20 const httpPort = process.env.HTTP_PORT || 9559;
  21 const httpsPort = process.env.PORT || process.env.HTTPS_PORT || 5443;
  22 const httpsServer = require('https').createServer(options, app);
  23 const httpServer = require('http').createServer(app);
  24 const io = require('socket.io')();
  25 const isTizen = process.platform === 'tizen';
  26 const supportMessagePort = isTizen;
  27 // Implementation for edge orchestration
  28 const supportEdgeOrchestration =
  29   typeof webapis !== 'undefined' && Object.hasOwnProperty.call(webapis, 'edge');
  30 console.log(TAG, `supportEdgeOrchestration : ${supportEdgeOrchestration}`);
  31
  32 io.attach(httpServer);
  33 io.attach(httpsServer);
  34
  35 const clients = new Set();
  36 const workers = new Map();
  37 const sockets = new Map();
  38 let edgeForCastanets = null;
  39 let forceQuitTimer = null;
  40 let isMeerkatStarted = false;
  41 let statusIntervalId = null;
  42 let localPort = null;
  43
  44 app.set('host', '0.0.0.0');
  45
  46 if (isTizen) {
  47   app.use(express.static(path.join(__dirname, './public')));
  48   app.use(express.static(path.join(__dirname, '../../shared/res')));
  49 } else {
  50   app.use(
  51     '/offload.html',
  52     express.static(path.join(__dirname, '../../offload/apps/web/offload.html'))
  53   );
  54   // Host offload-worker
  55   app.use(
  56     '/offload-worker',
  57     express.static(path.join(__dirname, '../../offload-worker/apps/web/'))
  58   );
  59   app.use(
  60     '/offload-worker/offload-worker.js',
  61     express.static(path.join(__dirname, '../../dist/offload-worker.js'))
  62   );
  63   app.use(
  64     '/face.webm',
  65     express.static(path.join(__dirname, '../../offload-worker/apps/web/face.webm'))
  66   );
  67   const serveIndex = require('serve-index');
  68   app.use(
  69     '/',
  70     express.static(path.join(__dirname, '../../dist')),
  71     express.static(path.join(__dirname, '../../offload/sample')),
  72     serveIndex(path.join(__dirname, '../../offload/sample'))
  73   );
  74   app.use(
  75     '/test',
  76     express.static(path.join(__dirname, '../../test')),
  77     serveIndex(path.join(__dirname, '../../test'))
  78   );
  79 }
  80
  81 function onConnection(socket) {
  82   if (isTizen && !isMeerkatStarted) {
  83     try {
  84       console.log(TAG, 'Try to start Meerkat client.');
  85       tizen.application.launch(
  86         'org.tizen.meerkat.client',
  87         () => {
  88           console.log(TAG, 'Meerkat client is started.');
  89           isMeerkatStarted = true;
  90         },
  91         err => console.error(TAG, 'Failed to launch Meerkat client. ' + err)
  92       );
  93     } catch (err) {
  94       console.error(TAG, 'Failed to launch Meerkat client. ' + err);
  95     }
  96   }
  97   console.log(TAG, `connection from '${socket.id}.`);
  98   sockets.set(socket.id, socket);
  99
 100   // client creates a session.
 101   socket.on('create', async function () {
 102     if (forceQuitTimer !== null) {
 103       clearTimeout(forceQuitTimer);
 104     }
 105
 106     if (clients.has(socket.id)) {
 107       console.log(TAG, `already created by ${socket.id}.`);
 108       return;
 109     }
 110     clients.add(socket.id);
 111
 112     let qr = null;
 113     const myAddress = getMyAddress();
 114     if (myAddress) {
 115       try {
 116         qr = await QRCode.toDataURL(
 117           'https://' + myAddress + ':5443/offload-worker.html'
 118         );
 119       } catch (err) {
 120         console.error(TAG, 'unabled to generate QR: ' + err);
 121       }
 122     }
 123
 124     socket.emit('greeting', {
 125       qrCode: qr,
 126       workers: Array.from(workers)
 127     });
 128
 129     console.log(
 130       TAG,
 131       `[client] session created by ${socket.id}. workers.size : ${workers.size}`
 132     );
 133
 134     if (supportEdgeOrchestration) {
 135       socket.emit(
 136         'capabilities',
 137         Array.from(edgeForCastanets.getCapabilities())
 138       );
 139     }
 140   });
 141
 142   socket.on('getcapabilities', function () {
 143     console.log(TAG, 'getcapabilities');
 144     if (supportEdgeOrchestration) {
 145       socket.emit(
 146         'capabilities',
 147         Array.from(edgeForCastanets.getCapabilities())
 148       );
 149     } else {
 150       socket.emit('capabilities', []);
 151     }
 152   });
 153
 154   socket.on('requestService', function (workerId) {
 155     if (supportEdgeOrchestration) {
 156       edgeForCastanets.requestService(workerId);
 157     }
 158   });
 159
 160   // new worker has been joined.
 161   socket.on('join', function (worker) {
 162     if (supportEdgeOrchestration) {
 163       let deviceIp = socket.request.connection.remoteAddress;
 164       if (deviceIp.indexOf('::ffff:') !== -1) {
 165         deviceIp = deviceIp.substr(7, deviceIp.length);
 166       }
 167
 168       if (deviceIp) {
 169         edgeForCastanets.joinDevice(deviceIp);
 170       }
 171     }
 172
 173     workers.set(worker.id, {
 174       socketId: socket.id,
 175       name: worker.name,
 176       features: worker.features,
 177       mediaDeviceInfos: worker.mediaDeviceInfos,
 178       compute_tasks: 0
 179     });
 180     console.log(
 181       TAG,
 182       `worker[${workers.size}] join: '${worker.id}' - '${socket.id}', '${worker.name}'`,
 183       worker.features
 184     );
 185
 186     for (const client of clients) {
 187       const clientSocket = sockets.get(client);
 188       clientSocket.emit('worker', {
 189         event: 'join',
 190         workerId: worker.id,
 191         socketId: socket.id,
 192         name: worker.name,
 193         features: worker.features,
 194         mediaDeviceInfos: worker.mediaDeviceInfos
 195       });
 196     }
 197   });
 198
 199   // route message between clients.
 200   socket.on('message', function (data) {
 201     console.log(TAG, `message ${JSON.stringify(data)}`);
 202     let socketId = null;
 203     if (workers.has(data.to)) {
 204       socketId = workers.get(data.to).socketId;
 205     } else if (clients.has(data.to)) {
 206       socketId = data.to;
 207     }
 208
 209     if (socketId) {
 210       const socket = sockets.get(socketId);
 211       socket.emit('message', data);
 212     }
 213   });
 214
 215   socket.on('disconnect', function (reason) {
 216     const socketId = socket.id;
 217     sockets.delete(socketId);
 218     if (clients.has(socketId)) {
 219       console.log(TAG, `[client] session terminated by client: ${socketId}`);
 220       clients.delete(socketId);
 221
 222       // broadcast to offload-worker
 223       for (const socket of sockets.values()) {
 224         socket.emit('client', {
 225           event: 'bye',
 226           socketId: socketId
 227         });
 228       }
 229
 230       if (clients.size === 0) {
 231         if (supportMessagePort) {
 232           closeServer();
 233         }
 234         forceQuitTimer = setTimeout(function () {
 235           console.log(
 236             TAG,
 237             'All clients are destroyed. Broadcast \'forceQuit\' to workers'
 238           );
 239           for (const socket of sockets.values()) {
 240             socket.emit('client', {
 241               event: 'forceQuit',
 242               socketId: socketId
 243             });
 244           }
 245         }, 5000);
 246       }
 247     } else {
 248       if (supportEdgeOrchestration) {
 249         let deviceIp = socket.request.connection.remoteAddress;
 250         if (deviceIp.indexOf('::ffff:') !== -1) {
 251           deviceIp = deviceIp.substr(7, deviceIp.length);
 252         }
 253
 254         if (deviceIp) {
 255           edgeForCastanets.disconnectDevice(deviceIp);
 256         }
 257       }
 258
 259       let workerId = null;
 260       workers.forEach(function (value, key, map) {
 261         if (value.socketId === socket.id) {
 262           workerId = key;
 263         }
 264       });
 265
 266       if (workerId) {
 267         for (const client of clients) {
 268           const socket = sockets.get(client);
 269           socket.emit('worker', {
 270             event: 'bye',
 271             workerId: workerId,
 272             socketId: socket.id
 273           });
 274         }
 275         workers.delete(workerId);
 276         console.log(TAG, `worker[${workers.size}] bye: '${workerId}'`);
 277       }
 278     }
 279   });
 280 }
 281
 282 io.of('/offload-js').on('connection', onConnection);
 283
 284 if (supportEdgeOrchestration) {
 285   edgeForCastanets = new Edge(
 286     'castanets',
 287     'android',
 288     'com.samsung.android.castanets'
 289   );
 290 }
 291
 292 function startServer() {
 293   console.log(TAG, 'starting server...');
 294
 295   if (!httpsServer.listening) {
 296     httpsServer.listen(httpsPort, function () {
 297       console.log(TAG, `server is listening on https ${httpsPort} port.`);
 298     });
 299   }
 300
 301   if (!httpServer.listening) {
 302     httpServer.listen(httpPort, function () {
 303       console.log(TAG, `server is listening on http ${httpPort} port.`);
 304     });
 305   }
 306 }
 307
 308 function closeServer() {
 309   console.log(TAG, 'closing server...');
 310
 311   if (httpsServer.listening) {
 312     httpsServer.close(err => {
 313       if (err) {
 314         console.error('failed to close the https server:', err);
 315       }
 316     });
 317   }
 318
 319   if (httpServer.listening) {
 320     httpServer.close(err => {
 321       if (err) {
 322         console.error('failed to close the http server:', err);
 323       }
 324     });
 325   }
 326 }
 327
 328 // Check the client status
 329 function checkConnectionStatus() {
 330   for (const client of clients) {
 331     const socket = sockets.get(client);
 332     if (socket.constructor === SocketTizen) {
 333       try {
 334         socket.emit('status');
 335       } catch (e) {
 336         console.error(TAG, `Failed to check ${client} status`);
 337         socket.handleEvents('disconnect');
 338       }
 339     }
 340   }
 341 }
 342
 343 function handleMessagePort(messages) {
 344   if (messages.length === 0) {
 345     console.error(TAG, 'Not found message');
 346     return;
 347   }
 348
 349   const message = messages[0];
 350   const event = message.key;
 351   const value = JSON.parse(message.value);
 352   const id = value.id;
 353
 354   if (event === 'connect') {
 355     // FIXME: The message port does not guarantee that the connection has
 356     //  been disconnected when the page is reloaded. Therefore, if a new
 357     //  connection occurs with the same id, the existing connection is
 358     //  disconnected.
 359     if (sockets.has(id)) {
 360       console.log(TAG, `Disconnect already connected socket: ${id}`);
 361       const socket = sockets.get(id);
 362       socket.handleEvents('disconnect');
 363     }
 364
 365     const socket = new SocketTizen(id, localPort);
 366     socket.on('connection', onConnection);
 367     socket.connect();
 368     sockets.set(id, socket);
 369     startServer();
 370   } else {
 371     const socket = sockets.get(id);
 372     socket.handleEvents(event, value.data);
 373   }
 374 }
 375
 376 module.exports.onStart = () => {
 377   console.log(`${TAG} onStart is called`);
 378   if (supportMessagePort) {
 379     console.log(TAG, 'listening tizen messageport...');
 380     localPort = tizen.messageport.requestLocalMessagePort('offload');
 381     localPort.addMessagePortListener(handleMessagePort);
 382
 383     // Prevent to terminate the process
 384     statusIntervalId = setInterval(checkConnectionStatus, 1000);
 385   } else {
 386     startServer();
 387   }
 388 };
 389
 390 module.exports.onStop = () => {
 391   console.log(`${TAG} onStop is called`);
 392   if (supportMessagePort) {
 393     for (const socket of sockets.values()) {
 394       socket.close();
 395     }
 396     if (localPort !== null) {
 397       localPort.removeMessagePortListener(handleMessagePort);
 398     }
 399     if (statusIntervalId !== null) {
 400       clearInterval(statusIntervalId);
 401     }
 402   }
 403
 404   closeServer();
 405 };
 406
 407 module.exports.onRequest = () => {
 408   console.log(`${TAG} onRequest is called`);
 409 };