telebit-relay.js/lib/pipe-ws.js

58 lines
2.2 KiB
JavaScript
Raw Normal View History

2018-06-01 06:41:32 +00:00
'use strict';
var Packer = require('proxy-packer');
2018-06-19 23:43:28 +00:00
module.exports = function pipeWs(servername, service, srv, conn, serviceport) {
2018-06-01 06:41:32 +00:00
var browserAddr = Packer.socketToAddr(conn);
var cid = Packer.addrToId(browserAddr);
browserAddr.service = service;
browserAddr.serviceport = serviceport;
browserAddr.name = servername;
conn.tunnelCid = cid;
2018-06-19 23:43:28 +00:00
var rid = Packer.socketToId(srv.upgradeReq.socket);
2018-06-01 06:41:32 +00:00
//if (state.debug) { console.log('[pipeWs] client', cid, '=> remote', rid, 'for', servername, 'via', service); }
function sendWs(data, serviceOverride) {
2018-06-19 23:43:28 +00:00
if (srv.ws && (!conn.tunnelClosing || serviceOverride)) {
2018-06-01 06:41:32 +00:00
try {
2018-06-19 23:43:28 +00:00
srv.ws.send(Packer.pack(browserAddr, data, serviceOverride), { binary: true });
2018-06-01 06:41:32 +00:00
// If we can't send data over the websocket as fast as this connection can send it to us
// (or there are a lot of connections trying to send over the same websocket) then we
// need to pause the connection for a little. We pause all connections if any are paused
// to make things more fair so a connection doesn't get stuck waiting for everyone else
// to finish because it got caught on the boundary. Also if serviceOverride is set it
// means the connection is over, so no need to pause it.
2018-06-19 23:43:28 +00:00
if (!serviceOverride && (srv.pausedConns.length || srv.ws.bufferedAmount > 1024*1024)) {
2018-06-01 06:41:32 +00:00
// console.log('pausing', cid, 'to allow web socket to catch up');
conn.pause();
2018-06-19 23:43:28 +00:00
srv.pausedConns.push(conn);
2018-06-01 06:41:32 +00:00
}
} catch (err) {
2018-06-19 23:43:28 +00:00
console.warn('[pipeWs] srv', rid, ' => client', cid, 'error sending websocket message', err);
2018-06-01 06:41:32 +00:00
}
}
}
2018-06-19 23:43:28 +00:00
srv.clients[cid] = conn;
conn.servername = servername;
conn.serviceport = serviceport;
conn.service = service;
2018-06-01 06:41:32 +00:00
conn.on('data', function (chunk) {
2018-06-19 23:43:28 +00:00
//if (state.debug) { console.log('[pipeWs] client', cid, ' => srv', rid, chunk.byteLength, 'bytes'); }
2018-06-01 06:41:32 +00:00
sendWs(chunk);
});
conn.on('error', function (err) {
console.warn('[pipeWs] client', cid, 'connection error:', err);
});
conn.on('close', function (hadErr) {
//if (state.debug) { console.log('[pipeWs] client', cid, 'closing'); }
sendWs(null, hadErr ? 'error': 'end');
2018-06-19 23:43:28 +00:00
delete srv.clients[cid];
2018-06-01 06:41:32 +00:00
});
};