Merge pull request #3 from topcoder-platform/30065431

Merge winning submission

Author: callmekatootie

README.md

@@ -3,7 +3,7 @@
 ## Dependencies
 
 - nodejs https://nodejs.org/en/ (v8+)
-
+- Kafka v2.11+
 
 
 ## Configuration
@@ -13,12 +13,16 @@ The following parameters can be set in config files or in env variables:
 - LOG_LEVEL: the log level
 - PORT: the server port
 - KAFKA_OPTIONS: Kafka consumer options, see https://www.npmjs.com/package/no-kafka for available options
+- MAX_MESSAGE_COUNT: max message count to cache per topic
 
 For the Kafka connection options:
 
 - connectionString is comma delimited list of initial brokers list
 - secure connection may be achieved via ssl field, see https://www.npmjs.com/package/no-kafka#ssl for details
 
+For front end config, see ui/README.md.
+
+
 ## Local Kafka setup
 
 - `http://kafka.apache.org/quickstart` contains details to setup and manage Kafka server,
@@ -33,6 +37,10 @@ For the Kafka connection options:
 - note that the zookeeper server is at localhost:2181, and Kafka server is at localhost:9092
 - use another terminal, go to same directory, create a topic:
   `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic challenge.notification.create`
+  similarly, you may create more topics, e.g.
+  `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic1`
+  `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic2`
+  `bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic topic3`
 - verify that the topic is created:
   `bin/kafka-topics.sh --list --zookeeper localhost:2181`,
   it should list out the created topics
@@ -63,20 +71,35 @@ For the Kafka connection options:
 
 ## Heroku Deployment
 
+- heroku login
+- heroku create
+- heroku config:set KAFKA_URL=some-public-kafka-url
+- heroku config:set API_URL=API-URL-according-to-the-heroku-app-URL
+- heroku config:set WS_URL=web-socket-URL-according-to-the-heroku-app-URL
 - git init
 - git add .
 - git commit -m message
-- heroku create
-- heroku config:set KAFKA_CONSUMER_URL=some-public-kafka-url
-- heroku config:set KAFKA_PRODUCER_URL=some-public-kafka-url
 - git push heroku master
 
 ## Verification
 
 - setup stuff following above deployment
 - in the UI, select a topic to view topic data stream
 - use the kafka-console-producer to generate some messages as above,
-  then watch the UI, it should got some messages
+  then watch the UI, it should get some messages
 - filter the messages and see results
 - use the UI to post message to Kafka, see above for example message, the data stream table should also show the posted message
 - you may also use the above kafka-console-consumer to view the Kafka messages
+
+
+## Notes
+
+- To keep the web socket connection alive, the following approaches are used:
+  (a) the server will handle both `error` and `close` events to terminate the web socket connection,
+      so that client side will re-start a new connection
+  (b) client side will handle onerror and onclose to re-start a new connection to server
+- The get/view topic, send message operations will show loading indicator,
+  but they are too fast because local back end is used, and especially for get/view topic operations web socket is used,
+  you may hardly see the indicator
+
+

config/default.js

@@ -12,4 +12,6 @@ module.exports = {
       key: process.env.KAFKA_CLIENT_CERT_KEY,
     },
   },
+  // max message count to cache per topic
+  MAX_MESSAGE_COUNT: process.env.MAX_MESSAGE_COUNT || 10000,
 };

set-env.js

@@ -4,8 +4,9 @@ const targetPath = './ui/src/config/config.js';
 
 const envConfigFile = `
 const config = {
-  API_URL: '${process.env.API_URL}/api/v1',
-  WS_URL: '${process.env.WS_URL}'
+  API_URL: '${process.env.API_URL || 'http://localhost:4000'}/api/v1',
+  WS_URL: '${process.env.WS_URL || 'ws://localhost:4000'}',
+  DEFAULT_MESSAGE_COUNT: ${process.env.DEFAULT_MESSAGE_COUNT || 20}
 };
 
 export default config;

src/dataStreamWS.js

@@ -8,10 +8,26 @@ const logger = require('./common/logger');
 const WebSocket = require('ws');
 const WebSocketServer = WebSocket.Server;
 const helper = require('./common/helper');
+const config = require('config');
 
 // all web socket client data
 const allWS = [];
 
+// all cached messages, key is topic, value is array of messages of the topic
+const allMessages = {};
+
+// max cache message count per topic
+const maxMsgCount = Number(config.MAX_MESSAGE_COUNT);
+
+// send data to client via web socket
+const sendData = (ws, payload) => {
+  try {
+    ws.send(JSON.stringify(payload));
+  } catch (err) {
+    logger.error(err);
+  }
+};
+
 /**
  * Setup web socket.
  */
@@ -28,41 +44,76 @@ const setup = (server) => {
     };
     allWS.push(clientData);
 
-    // got message from client
+    // got message from client, the message is string representation of JSON containing fields: topic and count,
+    // where count is the last count of messages of the topic to retrieve
     ws.on('message', (message) => {
       logger.debug(`web socket message: ${message}`);
-      if (message.startsWith('topic:') && message.length > 'topic:'.length) {
-        clientData.topic = message.substring('topic:'.length);
-      } else {
-        logger.error(`invalid web socket message: ${message}`);
+      let msgJSON;
+      try {
+        msgJSON = JSON.parse(message);
+      } catch (err) {
+        logger.err('invalid message', message, err);
+        return;
       }
+      clientData.topic = msgJSON.topic;
+      const topicMsgs = allMessages[msgJSON.topic] || [];
+      let startIndex = topicMsgs.length - msgJSON.count;
+      if (startIndex < 0) startIndex = 0;
+      const messages = topicMsgs.slice(startIndex);
+      // the 'full' flag is true, indicating the messages are full latest messages for client side,
+      // client side should clear the existing messages if any for the topic
+      sendData(ws, { full: true, topic: msgJSON.topic, messages });
     });
 
-    // close event handler
-    ws.on('close', () => {
+    // terminate web socket
+    const terminateWS = () => {
+      if (clientData.terminated) {
+        return;
+      }
+      clientData.terminated = true;
+
       for (let i = 0; i < allWS.length; i += 1) {
         if (id === allWS[i].id) {
           // remove the current client data
           allWS.splice(i, 1);
           break;
         }
       }
+      ws.close();
+    };
+
+    // close event handler
+    ws.on('close', () => {
       logger.debug('web socket closed');
+      terminateWS();
+    });
+
+    // error event handler
+    ws.on('error', (err) => {
+      logger.error('there is error for the web socket', err);
+      terminateWS();
     });
   });
+
+  wss.on('error', (err) => {
+    logger.error('there is error for the web socket server', err);
+  });
 };
 
 /**
- * Send message to all applicable web socket clients.
+ * Send message to all applicable web socket clients. The message will be cached to be retrieved by clients.
  */
 const sendMessage = (topic, message) => {
+  // cache message
+  if (!allMessages[topic]) allMessages[topic] = [];
+  allMessages[topic].push(message);
+  if (allMessages[topic].length > maxMsgCount) allMessages[topic].shift();
+
+  // send message to clients
   _.each(allWS, (clientData) => {
     if (topic === clientData.topic) {
-      try {
-        clientData.ws.send(message);
-      } catch (e) {
-        logger.error(e);
-      }
+      // the 'full' flag is false, indicating the message is to be added to client side
+      sendData(clientData.ws, { full: false, topic, messages: [message] });
     }
   });
 };

src/services/DataStreamService.js

@@ -38,11 +38,13 @@ function* getAllTopics() {
       topics.push(tp);
       // if the topic was not handled yet, then let the consumer handle the topic
       if (!consumers[tp]) {
-        consumer.subscribe(tp, { time: Kafka.LATEST_OFFSET }, dataHandler);
+        consumer.subscribe(tp, { time: Kafka.EARLIEST_OFFSET }, dataHandler);
         consumers[tp] = consumer;
       }
     }
   });
+  // sort topics
+  topics.sort();
   return topics;
 }
 

test/datastream.test.js

@@ -18,6 +18,10 @@ describe('Topcoder Data Stream REST API Tests', () => {
           return done(err);
         }
         expect(res.body.length > 0).to.equal(true);
+        // ensures the topics are sorted properly
+        for (let i = 0; i + 1 < res.body.length; i += 1) {
+          expect(res.body[i] < res.body[i + 1]).to.equal(true);
+        }
         return done();
       });
   });

ui/.gitignore

@@ -13,6 +13,7 @@ The following parameters can be set in config file:
 
 - API_URL: the API URL prefix
 - WS_URL: the Web Socket URL
+- DEFAULT_MESSAGE_COUNT: the default message count per topic
 
 ## Local deployment
 

ui/README.md

@@ -1,47 +1,103 @@
-.App {
-  text-align: center;
+.app {
+  width: 100%;
+}
+
+.left-nav {
+  width: 300px;
+  background-color: #444444;
+  color: #EEEEEE;
+  font-size: 13px;
 }
 
-.App-logo {
-  animation: App-logo-spin infinite 20s linear;
+.logo {
   height: 80px;
+  width: 100%;
+  padding-left: 60px;
+  padding-right: 60px;
+  padding-top: 22px;
+  padding-bottom: 22px;
+  background-color: #333333;
 }
 
-.App-header {
-  background-color: #222;
-  height: 150px;
-  padding: 20px;
-  color: white;
+.left-nav-item {
+  padding-left: 30px;
+  margin-top: 30px;
+  margin-bottom: 20px;
+}
+
+.topic-select {
+  width: 65%;
+  display: inline-block;
+  margin-right: 10px;
 }
 
-.App-title {
-  font-size: 1.5em;
+.filter {
+  padding-left: 0px;
+  text-align: center;
+  font-style: italic;
 }
 
-.App-item {
-  font-size: large;
-  margin: 10px;
+.filter-control {
+  width: 90%;
 }
 
-.topic-select {
-  width: 50%;
-  display: inline-block;
-  margin: 10px;
+.timestamp-control {
+  color: #000000;
 }
 
-.label {
-  width: 150px;
-  display: inline-block;
+.apply-button {
   margin-left: 30px;
-  margin-right: 10px;
 }
 
-.filter {
-  width: 250px;
-  display: inline-block;
+.main-content {
+  background-color: #F5F5F5;
+  padding: 20px;
+  width: calc(100% - 300px);
+}
+
+.error-msg {
+  color: red;
+  margin-bottom: 30px;
+}
+
+.message-input {
+  width: 75%;
+  margin-right: 20px;
+  display: inline;
+}
+
+.extra-message-count {
+  width: 100px;
+  display: inline;
+}
+
+.view-button {
+  vertical-align: top;
+}
+
+.get-button {
+  vertical-align: top;
+  margin-left: 10px;
+}
+
+.send-message-button {
+  vertical-align: text-bottom;
+}
+
+.loading-img-container {
+  position: fixed;
+  top: 0;
+  bottom: 0;
+  left: 0;
+  right: 0;
+  background: rgba(0, 0, 0, 0.8);
+  z-index: 9999;
+  opacity: 0.2;
 }
 
-@keyframes App-logo-spin {
-  from { transform: rotate(0deg); }
-  to { transform: rotate(360deg); }
+.loading-img {
+  position: fixed;
+  top: 50%;
+  left: 50%;
+  z-index: 99999;
 }