Merge branch 'kafka' into 'develop'

add apache Kafka

See merge request !3

package.json

@@ -37,6 +37,7 @@
         "cross-env": "7.0.0",
         "express": "4.17.1",
         "express-boom": "3.0.0",
+        "kafka-node": "^5.0.0",
         "morgan": "1.9.1",
         "pg": "^8.5.1",
         "pg-hstore": "^2.3.3",

src/config.json

@@ -2,7 +2,7 @@
     "development": {
         "username": "postgres",
         "password": "meu@sds12@!#gh",
-        "database": "node_sequelize",
+        "database": "chinguyen_nodejs-practice",
         "host": "27.74.255.96",
         "port": 5430,
         "dialect": "postgres"

src/dao/_index.ts

 import * as LanguagesDao from './languages'
 import * as AppUserDao from './appusers'
 import * as StudentDao from './students'
+import * as ConsumerDao from '../kafka/consumerMessage'
+import * as ProducerDao from '../kafka/messageProducer'
 
 export { LanguagesDao }
 export { AppUserDao }
 export { StudentDao }
+export { ConsumerDao }
+export { ProducerDao }

src/endpoints/_index.ts

 import * as LanguageController from './languages/_index'
 import * as AppUserController from './appusers/_index'
 import * as StudentsController from './students/_index'
+import * as ConsumerController from './consumer/_index'
+import * as ProducerController from './producer/_index'
 
-export { LanguageController, AppUserController, StudentsController }
+export { LanguageController, AppUserController, StudentsController, ConsumerController, ProducerController }

src/endpoints/consumer/_index.ts

+import * as ConsumerGet from './consumerMessage.get'
+
+
+export { ConsumerGet }
+

src/endpoints/consumer/consumerMessage.get.ts

+import { Request, Response } from 'express'
+import { ConsumerDao } from '../../dao/_index'
+
+export function getMessage(req: Request, res: Response){
+    return ConsumerDao.receiveMessage();
+}
\ No newline at end of file

src/endpoints/producer/_index.ts

+import * as ProducerPost from './producerMessage.post'
+
+
+export { ProducerPost }
\ No newline at end of file

src/endpoints/producer/producerMessage.post.ts

+import { ProducerDao } from '../../dao/_index'
+import { Request, Response } from 'express'
+
+
+export function producer(req: Request, res: Response){
+    return ProducerDao.produceMessage();
+}
\ No newline at end of file

src/kafka/consumerMessage.ts

+import * as kafka from 'kafka-node';
+//import {getConsumer} from './message'
+ 
+
+
+export function receiveMessage(): any{
+    const _that = this;
+    let topicName = 'message-chinguyen';
+    let produceTopic = topicName;
+    let kafkaConsumer = kafka.Consumer;
+    let Client = kafka.KafkaClient;
+    let client = new Client({kafkaHost: 'localhost:9092'})
+
+    let consumer = new kafkaConsumer(client,[{topic: produceTopic}],{autoCommit: false, fetchMaxWaitMs: 1000, fetchMaxBytes: 1024 * 1024})
+    
+    consumer.on('message', function(message){
+        console.log(message);
+    });
+    consumer.on('error', function (err) {
+        console.log('error', err);
+      });
+}
\ No newline at end of file

src/kafka/message.ts

+import * as kafka from 'kafka-node';
+//import {} from 'optimist'
+
+let clientOpts = {
+    kafkaHost: 'localhost:9092',
+    //sasl: configuration.kafkaClient.sasl,
+    connectTimeout: 10000,
+    maxAsyncRequests: 10,
+    requestTimeout: 30000
+};
+// Create Producer
+let MessageProducer = kafka.Producer, 
+aaClient = new kafka.KafkaClient(clientOpts),
+messageProducer = new MessageProducer(aaClient,{requireAcks: 1})
+
+messageProducer.on('ready', function () {
+    setMessageProducer(messageProducer);
+});
+
+export function getMessageProducer(): kafka.Producer {
+    return messageProducer;
+}
+
+export function setMessageProducer(producerObj: kafka.Producer): void {
+    messageProducer = producerObj;
+} 
+///
+

src/kafka/messageProducer.ts

+import { Producer } from 'kafka-node'
+import {getMessageProducer} from './message'
+
+    export function produceMessage() {
+            const _that = this;
+            let producer = getMessageProducer();
+            let topicName = 'message-chinguyen';
+            let produceTopic = topicName;
+
+            let payloads: {
+                topic: any;
+                messages: any;
+                key: any;
+                attributes: any;
+                timestamp: any;
+              }[] = [];
+              let curTimestamp = Date.now();
+              payloads.push({topic: produceTopic, messages: 'hahahahaha', key: 1, attributes: 'haha', timestamp: curTimestamp});
+              
+              if (payloads.length > 0) {
+                producer.send(payloads, function (err, data) {
+                  if (err) {
+                    console.log("error");
+                  }
+                  else {
+                    console.log("success");
+                    
+                  }
+                });
+              }
+        
+        
+        }
\ No newline at end of file

src/routes/kafka.ts

+import { Express } from 'express'
+import { ConsumerController, ProducerController } from '../endpoints/_index'
+
+export function routes(app: Express){
+    app.post('/api/producerMessage', ProducerController.ProducerPost.producer)
+    app.get('/api/consumerMessage', ConsumerController.ConsumerGet.getMessage)
+}

src/sqlz/config/config.json

 {
     "development": {
         "username": "postgres",
-        "password": "chinguyen@123",
-        "database": "node_sequelize",
-        "host": "127.0.0.1",
+        "password": "meu@sds12@!#gh",
+        "database": "chinguyen_nodejs-practice",
+        "host": "27.74.255.96",
         "port": 5432,
         "dialect": "postgres"
     }