web-socket/wss-server.js

'use strict';  

const encoding = require("encoding");
const zlib = require('zlib');

function stringToArray(bufferString) {
 // let uint8Array = new TextEncoder("utf-8").encode(bufferString);
//var resultBuffer = encoding.convert(nameString, 'ASCII', 'UTF-8');
var uint8Array = encoding.convert(bufferString, 'ASCII', 'UTF-8');
 return uint8Array;
}

function getDateTime() {

    var date = new Date();

    var hour = date.getHours();
    hour = (hour < 10 ? "0" : "") + hour;

    var min  = date.getMinutes();
    min = (min < 10 ? "0" : "") + min;

    var sec  = date.getSeconds();
    sec = (sec < 10 ? "0" : "") + sec;

    var year = date.getFullYear();

    var month = date.getMonth() + 1;
    month = (month < 10 ? "0" : "") + month;

    var day  = date.getDate();
    day = (day < 10 ? "0" : "") + day;

    return year + ":" + month + ":" + day + ":" + hour + ":" + min + ":" + sec;

}

function intToUint8Array(value) {
  console.log('value:' + value);
  var arr = new Uint8Array(2);
  arr[0] = (value >> 8);
  arr[1] = value % 0x100;
  return arr;
}

function intToUint8ArrayCRC(value) {
  console.log('value:' + value);
  var arr = new Uint8Array(2);
  arr[0] = value % 0x100;
  arr[1] = (value >> 8);
  return arr;
}

function calc_crc16(data, len) {
  var crc = 0xffff;
  var LSB;
  for (var i = 0; i < len; i++) {
    crc ^= data[i];
    for (var j = 0; j < 8; j++) {
      LSB = crc & 1;
      crc = crc >> 1;
      if (LSB != 0) crc ^= 0xa001;
    }
  }
  return crc;
}

//wss start

var fs = require('fs'); 

var cfg = { 
ssl: true, 
port: 443, 
ssl_key: 'ssl.key', ssl_cert: 'ssl.crt' 
}; 

var httpServ = (cfg.ssl) ? require('https') : require('http');
 const  WebSocketServer = require('ws').Server;
 var app = null; 

 var processRequest = function(req, res)
 { 
res.writeHead(200);
 res.end('All glory to WebSockets!\n');
 }; 

if (cfg.ssl) {
 app = httpServ.createServer({ 
 key: fs.readFileSync(cfg.ssl_key),
 cert: fs.readFileSync(cfg.ssl_cert) 
}, processRequest).listen(cfg.port); 
} else {
 app = httpServ.createServer(processRequest).listen(cfg.port);
 } 

const wss = new WebSocketServer({ server: app });


 wss.on('connection', function(wsConnect)
 {

//0

//
//mqtt
const mqtt = require('mqtt');
var opt = {
  port:1883,
  clientId: 'phonetest'
};

function uint8ArrayToInt(arr) {
  console.log('arr[0]:' + arr[0]);
  console.log('arr[1]:' + arr[1]);
  return arr[0] * 0x100 + arr[1];
}

const client  = mqtt.connect('tcp://iot.twgigatech.cn:1883');
client.on('connect', function () 
{
//fs.appendFile('./log/wss_conn.log',getDateTime()+' ' +'wss connect in'+'\r\n');
//fs.appendFile('./log/all.log',getDateTime()+' ' +'wss connect in \r\n');
console.log('connect MQTT server');

//client.subscribe("CAD9A3D3EUWWNBKY111A/iotk");

client.subscribe("CAD9A3D3EUWWNBKY111A/wxxk/#");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/#");

/*
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/5");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/6");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/7");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/4");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/3");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/2");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/1");
client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/8");
*/
// client.publish('CAD9A3D3EUWWNBKY/iotk','test');

});

client.on('message', function (topic, message) {
console.log(' ');
console.log(' topic = '+topic.toString());
console.log('==============mqtt  rece start ');

console.log('mqtt rece: '+message.toString());
console.log('=======================');

//fs.appendFile('./log/all.log',getDateTime()+' ' +'mqtt_rece: '+message.toString()+'\r\n');
//fs.appendFile('./log/mqtt_rece.log',getDateTime()+' ' +'mqtt_rece: '+message.toString()+'\r\n');

  console.log('messagelen:' + message.length.toString());

  let lenBin = message.subarray(26, 28);
  let len = uint8ArrayToInt(lenBin);
  console.log('len:' + len);
  let blockData = message.subarray(32, message.length);
 console.log('blockData:' + blockData);
  let dataType = blockData[0];
  console.log('dataType:' + dataType);
  var rawData = blockData.subarray(25, len-2); 
  console.log('rawData:' + rawData);
 
  console.log('rawData len:' + rawData.length.toString());
  var strData;
  if(dataType == 0)
  {
    strData = arrayToString(rawData);
  } else if (dataType == 1) 
{
   // strData = zlib.inflate(rawData, function(err, buffer) 

   zlib.inflate(rawData, function(err, buffer) 
{
if (!err) 
{
  console.log('buffer= '+buffer);
  //fs.appendFile('./log/all.log',getDateTime()+' ' +'buffer: '+buffer+'\r\n');
  //buffer.length.toString();
  console.log('mqtt inflate OK ');
  console.log('wss.clients size~~~~~~~ '+ wss.clients.size);	
  //fs.appendFile('./log/all.log',getDateTime()+' ' +'wss client size: '+wss.clients.size+'\r\n');

  //wss.send(buffer);
  //wssclients[0].send(buffer);
  var jsdata = buffer.toString();
  
  if (topic.toString().indexOf('wxxm')!=-1)
  {
    console.log(' wxxm find ');
    var jssdata = {
      "cmd": "wxxIotM"//,
      //"data":""
    };
    //jssdata['data']=jsdata ;
    jssdata['data']=JSON.parse(jsdata) ;



  try
{
//fs.appendFile('./log/all.log',getDateTime()+' ' +'wss_tomobile wxxIotM: '+JSON.stringify(jssdata )+'\r\n' );
wsConnect.send(JSON.stringify(jssdata ));
//fs.appendFile('./log/wss_tomobile.log',getDateTime()+' ' +'wss_tomobile wxxIotM: '+JSON.stringify(jssdata )+'\r\n');
}
catch(e)
{
   console.log(e)
   }
console.log(JSON.stringify(jssdata ));
}



 
if (topic.toString().indexOf('wxxk')!=-1)
{
  console.log(' wxxk find');
  var jssdata = {
    "cmd": "wxxIotK"//,
        //"data":""
  };
  //jssdata['data']=jsdata ;
  jssdata['data']=JSON.parse(jsdata) ;

  try
  {
  //fs.appendFile('./log/all.log',getDateTime()+' ' +'wss_tomobile wxxIotK: '+jssdata+'\r\n' );
  wsConnect.send(JSON.stringify(jssdata ));
  //fs.appendFile('./log/wss_tomobile.log',getDateTime()+' ' +'wss_tomobile wxxIotK: '+jssdata +'\r\n');
  }
  catch(e)
  {
    console.log(e)
  }
console.log(JSON.stringify(jssdata ));
}


 console.log(' wssclients[0].send ');
}
});

//number = 5
  
 } else {
    strData = '';
  }  


console.log('============mqtt rece finish');
console.log(' ');

//wsConnect.send(Buffer.from(message));
//  client.end();
});
//


//0


 wsConnect.on('message', function(message)
 { 
      console.log('wss rece: '+message); 
      //to k or m
//fs.appendFile('./log/wss_recemsg.log',getDateTime()+' ' +'wss rece: '+message+'\r\n');
//fs.appendFile('./log/all.log',getDateTime()+' ' +'wss rece: '+message+'\r\n');
var obj = JSON.parse(message.toString());
var vdata =obj ['data'];
console.log( 'obj.data:' + vdata);
console.log( '~~~~~~~~~~~~~~~~~~~~~');
message = vdata ;


  var rawData;
  var dataType;
  var event = 0;

  var bufferID = stringToArray('GA300_1_A1001001_5');
       console.log('message.length:' + message.length);

try
{
 if (message.length < 128) {
    dataType = 0;
    rawData = stringToArray(message );
 console.log('rawData:' + rawData);
  var blockData = new Uint8Array(25 + rawData.length + 2);
  blockData[0] = dataType;
  blockData.set(bufferID,1);
  blockData.set(rawData, 25);
  var crc16 = calc_crc16(blockData, 25 + rawData.length)
  blockData.set(intToUint8ArrayCRC(crc16), 25 + rawData.length);
  console.log('blockData: ' + blockData);
var packedData = new Uint8Array(32 + blockData.length);
  packedData.set(bufferID);
  packedData.set(intToUint8Array(event), 24);
  packedData.set(intToUint8Array(25+rawData.length+2), 26);
  packedData.set(intToUint8Array(0), 28);
  packedData[30] = 1;
  packedData[31] = 1;
  packedData.set(blockData, 32);

 console.log('~~~~~~~~~~~~~~~~~~');
 console.log('~~~~~~~~~~~~~~~~~~');
 console.log('~~~~~~~~~~~~~~~~~~');

 client.publish('CAD9A3D3EUWWNBKY111A/iotm/5',Buffer.from(packedData));
 client.publish('CAD9A3D3EUWWNBKY111A/iotk/5',Buffer.from(packedData));
//fs.File('./log/mqtt_puablish.log',getDateTime()+' ' +'mqtt publish: '+packedData+'\r\n');
//fs.appendFile('./log/all.log',getDateTime()+' ' +'mqtt publish: '+packedData+'\r\n');
 console.log('~~~~~~~~~~~~~~~~~~');

  } else {
    dataType = 1;

zlib.deflate(message, (err, buffer) => {
try
{
  if (!err) {

    console.log('deflate OK'+buffer.toString('base64'));
 console.log('rawData:' + buffer);
rawData=buffer;
  var blockData = new Uint8Array(25 + rawData.length + 2);
  blockData[0] = dataType;
  blockData.set(bufferID,1);
  blockData.set(rawData, 25);
  var crc16 = calc_crc16(blockData, 25 + rawData.length)
  blockData.set(intToUint8ArrayCRC(crc16), 25 + rawData.length);
  console.log('blockData: ' + blockData);
var packedData = new Uint8Array(32 + blockData.length);
  packedData.set(bufferID);
  packedData.set(intToUint8Array(event), 24);
  packedData.set(intToUint8Array(25+rawData.length+2), 26);
  packedData.set(intToUint8Array(0), 28);
  packedData[30] = 1;
  packedData[31] = 1;
  packedData.set(blockData, 32);
 console.log('~~~~~~~~~~~~~~~~~~');
 console.log('~~~~~~~~~~~~~~~~~~');
 console.log('~~~~~~~~~~~~~~~~~~');
 client.publish('CAD9A3D3EUWWNBKY111A/iotm/5',Buffer.from(packedData));
 client.publish('CAD9A3D3EUWWNBKY111A/iotk/5',Buffer.from(packedData));
//fs.appendFile('./log/mqtt_puablish.log',getDateTime()+' ' +'mqtt publish: '+packedData+'\r\n');
//fs.appendFile('./log/all.log',getDateTime()+' ' +'mqtt publish: '+packedData+'\r\n');
 console.log('~~~~~~~~~~~~~~~~~~');
  } 

}
catch(e){
            console.log(e)
}


});

}


 }
catch(e){
            console.log(e)
}

   //client.publish('CAD9A3D3EUWWNBKY111A/iotm/5',rawData);
   //client.publish('CAD9A3D3EUWWNBKY111A/iotk/5',rawData);
  
   //client.subscribe("CAD9A3D3EUWWNBKY111A/wxxm/#");

   //wsConnect.send(message);
 }); 
});

//wss finish


//0



//0