Не удается отправить полученные данные в мой поток socket.io?

Question

Я пытаюсь переключиться с одного mysql-запроса на соединение mysql-pool, так что пользователи могут использовать одно соединение mysql, но я не знаком с этим вообще (также новичок в nodejs/socket.io).

Следующий код это то, что я сделал до сих пор для передачи данных каждый второй к разъему в массиве:

var 
port    = process.env.OPENSHIFT_NODEJS_PORT || 8000, 
ip     = process.env.OPENSHIFT_NODEJS_IP || '127.0.0.1', 
app     = require('http').createServer(handler), 
fs     = require('fs'), 
request    = require('request'), 
mysql    = require('mysql'), 
moment    = require('moment'), 
tz     = require('moment-timezone'), 
pool    = mysql.createPool({ 
connectionLimit: 100, 
host:    'xxx', 
user:    'xxx', 
password:   'xxx', 
database:   'xxx', 
debug:    false, 
port:    3306}), 
socketArray   = [], 
POLLING_INTERVAL = 1000, 
pollingTimer; 
moment.tz.setDefault("Europe/Berlin"); 

var io = require('socket.io').listen(app); 
io.set('origins', '*:*'); 

function time() 
{ 
output = new Date(); 
output = moment().format('(H:mm:ss.SS) '); 
return output; 
} 

function handler(req,res) 
    { 
    res.setHeader("Access-Control-Allow-Origin", "*"); 
    res.setHeader("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept"); 
    res.statusCode = 200; 
    res.connection.setTimeout(0); 
    res.end(); 
} 
app.listen(port,ip); 

function pollingLoop() { 
    if (socketArray.length === 0) { 
     // no connections, wait and try again 
     setTimeout(pollingLoop, POLLING_INTERVAL); 
     return; // continue without sending mysql query 
    } 
    pool.getConnection(function(err,connection){ 
    if (err) { console.log({"code" : 100, "status" : "connection-db error"}); return; } 
    console.log('connected as id ' + connection.threadId); 
    console.log('socketArray length: ' + socketArray.length); 

    var selection = 
     "SELECT\ 
     a.`id`,a.`product_id` AS pid,a.`random` AS nr,a.`price`,a.`price_end` AS pe,\ 
     TIMESTAMPDIFF(SECOND,NOW(),a.`datetime`) AS duration,\ 
     ABS(TIMESTAMPDIFF(SECOND,NOW(),b.`date`)) AS hb\ 
     FROM `auctions` AS a\ 
     LEFT JOIN `auctions_bids` AS b ON b.`auction_id` = a.`id`\ 
     WHERE TIMESTAMPDIFF(SECOND,NOW(),a.`datetime`) > '-1'\ 
     GROUP BY a.`id`\ 
     ORDER BY `duration` DESC,`id` DESC LIMIT 15"; 
    var streamArray = [], lg = '';      

    var query = connection.query(selection, function(err, results, rows){ 
    lg += ('id: '+results[0].id+' ('+results[0].duration+') ');   

    if 
    (
    ((results[0].duration < 2 || results[0].duration <= results[0].nr) && (results[0].price <= results[0].pe)) 
    || 
    ((results[0].duration < 2 || results[0].duration <= results[0].nr) && (results[0].hb > 0 && results[0].hb < 30)) 
    ) 
    { 
    min = 3; 
    max = 5; 
    rand = Math.floor(Math.random()*(max-min+1)+min); 
    price = results[0].price+0.01; 
    price = price.toFixed(2); 

    pool.query('UPDATE `auctions` SET `random` = ?,`price` = ?, `datetime` = DATE_ADD(`datetime`,INTERVAL(17-TIMESTAMPDIFF(SECOND,NOW(),`datetime`))SECOND) WHERE `id` = ?',[rand, price, results[0].id]); 
    console.log(time()+'UPDATED id '+results[0].id+': random ('+rand+') price ('+price+'€)'); 
    }  
    streamArray.push(results[0]);   

    updateSockets({ streamArray: streamArray });  
    console.log("auctions pushed: " + streamArray); 
    connection.release(); 
    setTimeout(pollingLoop, POLLING_INTERVAL); 
    }); 
    console.log(time()+lg+' C: '+socketArray.length); 
}); 
} 
pollingLoop(); 

io.sockets.on('connection', function(socket) { 
    socket.on('disconnect', function() { 
    clearTimeout(pollingTimer); 
    var socketIndex = socketArray.indexOf(socket); 
    console.log(time()+'SOCKET-ID = %s DISCONNECTED', socketIndex); 
    if (~socketIndex) { socketArray.splice(socketIndex, 1); } 
    }); 
    console.log(time()+'NEW SOCKET CONNECTED!'); 
    socketArray.push(socket); 
}); 

var updateSockets = function(data) { 
    socketArray.forEach(function(tmpSocket) { tmpSocket.volatile.emit('stream', data); }); 
}; 

console.log(time()+'server.js executed\n'); 

Но это не посылает мне никаких данных в WebSocket. Является ли этот подход (кодовая структура) даже правильным? Раньше я использовал query.on('results'), чтобы получить данные, как это:

var selection = "SELECT * FROM auctions";    
var query = mysql.query(selection), auctions = []; 
query.on('result', function(auction) { 
console.log('id: '+auction.id+' ('+auction.duration+') '); 
}); 

Это работало отлично данные, показывающие, с , но как это сделать в моей связи тузд пула?

Также через несколько секунд я получаю сообщение об ошибке, которое release() даже не определено, но оно указано в документации по модулю mysql ... поэтому я считаю, что мой логический процесс как-то неверен.

1. Должен ли я использовать connection.end() и .release() вообще? Поскольку соединение
   никогда не должно заканчиваться.
2. Должен ли я использовать setInterval(function() { mysql.query('SELECT
1'); }, 5000);, как было сказано в другом вопросе StackOverflow, чтобы поддерживать соединение здесь? (nodejs mysql Error: Connection lost The server closed the connection)

(Цените какие-либо советы или ответы даже некоторые из моих вопросов лучше, чем некоторые ответы нет, потому что я испытал, что эта тема не отвечает много на всех!).

EDIT: Обновлен весь мой код (см. Выше). Результат выглядит следующим образом: http://s21.postimg.org/avsxa87rb/output.jpg

Итак, поток получает данные, но в console.log ничего, и есть эта ошибка JavaScript?

Answer 1

Вы должны создать пул и использовать getConnection в этом пуле. Затем, когда вы закончите соединение, отпустите его. Кроме того, вам не нужно останавливать pollingLoop или запускать его для каждого соединения, достаточно одного цикла.

Я не понимал утверждения if с условиями, поэтому я его пропустил. Скорее всего, нужно пойти куда-то еще.

var socketArr = []; 

function handler(req, res) { 
    res.statusCode = 200; 
    res.connection.setTimeout(0); 
    res.end(); 
} 
app.listen(port, ip); 
var pool = mysql.createPool({ 
    host  : 'example.org', 
    user  : 'bob', 
    password : 'secret' 
}); 

function pollingLoop() { 
    if (socketArr.length === 0) { 
     // no connections, wait and try again 
     setTimeout(pollingLoop, 1000); 
     return; // continue without sending mysql query 
    } 
    pool.getConnection(function (err, connection) { 
     if (err) { 
      console.log({ 
       "code": 100, 
       "status": "Error in connection database" 
      }); 
      return; 
     } 
     console.log('connected as id ' + connection.threadId); 
     var selection = "SELECT * FROM auctions"; 
     var streamArray = [], 
      lg = ''; 
     var query = connection.query(selection, function (err, results, fields, rows) { 
      lg += ('id: ' + results[0].id + ' (' + results[0].duration + ') '); 
      /*if (conditions) { 
       var query_update = connection.query('UPDATE `auctions` SET `price` = ? WHERE `id` = ?', [price, auction.id]); 
       console.log(time() + 'UPDATED id ' + auction.id + ': price (' + price + '€)'); 
      }*/ 
      streamArray.push(results); 
      updateSockets({ 
       streamArray: streamArray 
      }); 
      console.log("auctions pushed: " + streamArray); 
      connection.release(); 
      setTimeout(pollingLoop, 1000); 
     }); 
     console.log(time() + lg + ' C: ' + socketArr.length); 
    }); 
} 
// start loop 
pollingLoop(); 

io.sockets.on('connection', function (socket) { 
    socket.on('disconnect', function() { 
     var socketIndex = socketArr.indexOf(socket); 
     console.log(time() + 'SOCKET-ID = %s DISCONNECTED', socketIndex); 
     if (~socketIndex) { 
      socketArr.splice(socketIndex, 1); 
     } 
    }); 
    console.log(time() + 'NEW SOCKET CONNECTED!'); 
    socketArr.push(socket); 
}); 
var updateSockets = function (data) { 
    socketArr.forEach(function (tmpSocket) { 
     tmpSocket.volatile.emit('stream', data); 
    }); 
};