Merge branch 'develop' of...

Merge branch 'develop' of http://27.74.255.96:8088/chinguyen/chinguyen_nodejs_managestudents into develop

# Conflicts:
#	package.json
#	src/sqlz/config/config.json

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();
+}

src/endpoints/producer/_index.ts

+import * as ProducerPost from './producerMessage.post'
+
+
+export { ProducerPost }

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();
+}

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);
+  });
+}

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");
+
+      }
+    });
+  }
+
+
+}

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)
+}