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});
+    }
+
+  });
+
+})();
diff --git a/src/controllers/redis.js b/src/controllers/redis.js
deleted file mode 100644
index 7e86000..0000000
--- a/src/controllers/redis.js
+++ /dev/null
@@ -1,77 +0,0 @@
-
-/*
- * Copyright 2017-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 request = require('superagent');
-  const socket = socketIo.get();
-  const _ = require('lodash');
-
-  var redis = require('redis');
-  
-  var client = redis.createClient({
-    host: config.redis.host,
-    port: config.redis.port
-  });
-
-  client.on('error', function (err) {
-    logger.log('error', err);
-  });
-
-  client.on('ready', function () {
-    logger.log('info', 'Redis connected');
-  });
-
-  client.on('subscribe', function (channel) {
-    logger.log('debug', `Subscribed to channel: ${channel}`);
-  });
-
-  client.on('pmessage', function (pattern, channel, message) {
-    if (channel === 'Diag') {
-      return;
-    }
-
-
-    let msg;
-    try {
-      msg = JSON.parse(message);
-      // TODO find the user that needs to be notified for msg.object update
-    }
-    catch(e) {
-      // send the event also if it is not JSON
-      msg = message;
-    }
-
-    if (msg.deleted) {
-      logger.log('warn', 'Remove on: ' + channel + ': ' + message);
-      socket.emit('remove', {model: channel, msg: msg, deleted: true});
-    }
-    else {
-      logger.log('warn', 'Update on: ' + channel + ': ' + message);
-      socket.emit('update', {model: channel, msg: msg});
-    }
-  });
-
-  // subscribe to all channels
-  client.psubscribe('*');
-
-})();
\ No newline at end of file
diff --git a/src/controllers/websocket.js b/src/controllers/websocket.js
index 30a7885..e04e13a 100644
--- a/src/controllers/websocket.js
+++ b/src/controllers/websocket.js
@@ -1,4 +1,3 @@
-
 /*
  * Copyright 2017-present Open Networking Foundation
 
@@ -15,7 +14,6 @@
  * limitations under the License.
  */
 
-
 (function () {
   'use strict';
 
@@ -54,4 +52,4 @@
   // const socket = socketIo.get();
   // socket.emit('eventName', data);
 
-})(); 
\ No newline at end of file
+})();