98 lines
No EOL
4.2 KiB
JavaScript
98 lines
No EOL
4.2 KiB
JavaScript
module.exports = function(socketNamespaces) {
|
|
const kafka = require("kafka-node");
|
|
|
|
const client = new kafka.Client("localhost:2181");
|
|
|
|
const topics = [
|
|
{
|
|
topic: "tcp.realtime"
|
|
}
|
|
];
|
|
/* const options = {
|
|
autoCommit: true,
|
|
fetchMaxWaitMs: 1000,
|
|
fetchMaxBytes: 1024 * 1024,
|
|
encoding: "buffer"
|
|
}; */
|
|
|
|
var options = {
|
|
//host: 'zookeeper:2181', // zookeeper host omit if connecting directly to broker (see kafkaHost below)
|
|
kafkaHost: 'localhost:9092', // connect directly to kafka broker (instantiates a KafkaClient)
|
|
zk : undefined, // put client zk settings if you need them (see Client)
|
|
batch: undefined, // put client batch settings if you need them (see Client)
|
|
ssl: false, // optional (defaults to false) or tls options hash
|
|
groupId: 'ws.realtime',
|
|
sessionTimeout: 15000,
|
|
// An array of partition assignment protocols ordered by preference.
|
|
// 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
|
|
protocol: ['roundrobin'],
|
|
|
|
// Offsets to use for new groups other options could be 'earliest' or 'none' (none will emit an error if no offsets were saved)
|
|
// equivalent to Java client's auto.offset.reset
|
|
fromOffset: 'latest', // default
|
|
commitOffsetsOnFirstJoin: true, // on the very first time this consumer group subscribes to a topic, record the offset returned in fromOffset (latest/earliest)
|
|
// how to recover from OutOfRangeOffset error (where save offset is past server retention) accepts same value as fromOffset
|
|
outOfRangeOffset: 'latest', // default
|
|
migrateHLC: false, // for details please see Migration section below
|
|
migrateRolling: false,
|
|
// Callback to allow consumers with autoCommit false a chance to commit before a rebalance finishes
|
|
// isAlreadyMember will be false on the first connection, and true on rebalances triggered after that
|
|
onRebalance: null//(isAlreadyMember, callback) => { callback(); } // or null
|
|
};
|
|
|
|
const consumer = new kafka.ConsumerGroup(options, 'tcp.realtime');
|
|
consumer.on("message", function (message) {
|
|
/* Print latest offset. */
|
|
/* var offset = new kafka.Offset(client);
|
|
|
|
offset.fetch([{ topic: 'tcp.realtime', partition: 0, time: -1 }], function (err, data) {
|
|
var latestOffset = data['tcp.realtime']['0'][0];
|
|
console.log("Consumer current offset: " + latestOffset);
|
|
}); */
|
|
// Read string into a buffer.
|
|
var buf = new Buffer(message.value, "binary");
|
|
var decodedMessage = JSON.parse(buf.toString());
|
|
//console.log(decodedMessage);
|
|
|
|
switch (decodedMessage.namespace) {
|
|
case 'gpsio':
|
|
if (decodedMessage.channel.endsWith('acc')) {
|
|
socketNamespaces[decodedMessage.namespace]
|
|
.to(decodedMessage.room)
|
|
.emit(decodedMessage.channel, decodedMessage.data[0], decodedMessage.data[1], decodedMessage.data[2], decodedMessage.data[3]);
|
|
}
|
|
else {
|
|
socketNamespaces[decodedMessage.namespace]
|
|
.to(decodedMessage.room)
|
|
.emit(decodedMessage.channel, decodedMessage.data[0], decodedMessage.data[1], decodedMessage.data[2]);
|
|
}
|
|
break;
|
|
|
|
case 'notifIO':
|
|
socketNamespaces[decodedMessage.namespace]
|
|
.emit(decodedMessage.channel, decodedMessage.data[0]);
|
|
break;
|
|
|
|
case 'sbNotifIO':
|
|
socketNamespaces[decodedMessage.namespace]
|
|
.emit(decodedMessage.channel, decodedMessage.data[0]);
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
});
|
|
|
|
consumer.on("error", function(err) {
|
|
console.log("error", err);
|
|
});
|
|
|
|
consumer.on('offsetOutOfRange', function (err) {
|
|
console.log("offsetOutOfRange", err);
|
|
});
|
|
|
|
process.on("SIGINT", function() {
|
|
consumer.close(true, function() {
|
|
process.exit();
|
|
});
|
|
});
|
|
} |