latest updates @APR 20, 2022

kafka-consumer.js

@@ -0,0 +1,98 @@
+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();
+        });
+    });
+}