[SEBA-233]
Change to using kafka for GUI events, instead of redis
Change-Id: I205f18632f15b44a5447530af33e94dd8c88674e
diff --git a/src/controllers/kafka.js b/src/controllers/kafka.js
new file mode 100644
index 0000000..f754a65
--- /dev/null
+++ b/src/controllers/kafka.js
@@ -0,0 +1,84 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+(function () {
+ 'use strict';
+
+ const socketIo = require('./websocket.js');
+ const config = require('../config/config.js');
+ const logger = require('../config/logger.js');
+ const socket = socketIo.get();
+
+ // docs: https://github.com/Blizzard/node-rdkafka
+ var kafka = require('node-rdkafka');
+
+ logger.log('debug',`Using librdkafka version: ${kafka.librdkafkaVersion}, kafka features: '${kafka.features}'`);
+ logger.log('debug',`Connecting to broker: ${config.kafka_bootstrap_servers}`);
+
+ var stream = kafka.KafkaConsumer.createReadStream({
+ 'metadata.broker.list': config.kafka_bootstrap_servers,
+ 'group.id': 'xos-ws',
+ 'socket.keepalive.enable': true,
+ 'enable.auto.commit': false
+ }, {}, {
+ topics: ['xos.gui_events'],
+ });
+
+ stream.on('ready', function () {
+ logger.log('info', 'Kafka connected');
+ });
+
+ stream.on('error', function (err) {
+ logger.log('error', err);
+ });
+
+ stream.consumer.on('event.error', function (err) {
+ logger.log('error', err);
+ });
+
+ stream.on('data', function (msg) {
+ logger.log('debug', `Topic: ${msg.topic}, Key: ${msg.key}, Timestamp: ${msg.timestamp}`);
+
+ // strip diag messages
+ // NOTE: have to coerce to string (due to FFI?)
+ if (msg.key.toString() === 'Diag') {
+ return;
+ }
+
+ let msgobj;
+
+ try {
+ // TODO find the user that needs to be notified for msg.object update
+ msgobj = JSON.parse(msg.value)
+ }
+
+ catch(e) {
+ // stringify the event if it is not JSON
+ msgobj = msg.value.toString()
+ }
+
+ if (msgobj.deleted) {
+ logger.log('info', 'Remove on: ' + msg.key + ': ' + msg.value);
+ socket.emit('remove', {model: msg.key.toString(), msg: msgobj, deleted: true});
+ }
+ else {
+ logger.log('info', 'Update on: ' + msg.key + ': ' + msg.value);
+ socket.emit('update', {model: msg.key.toString(), msg: msgobj});
+ }
+
+ });
+
+})();