fix send

Author: thanhngodinh

.env

@@ -6,5 +6,5 @@ AMQHOST=localhost
 AMQPORT=61613
 AMQUSERNAME=admin
 AMQPASSWORD=admin
-AMQDESTINATIONNAME=my-topic
+AMQDESTINATIONNAME=topic
 AMQSUBSCRIPTIONNAME=sub-user

src/context.ts

@@ -42,6 +42,7 @@ export const user: Attributes = {
 export interface ApplicationContext {
   handle: (data: User, header?: StringMap) => Promise<number>;
   read: (handle: (data: User, header?: StringMap) => Promise<number>) => void;
+  write: (data: User) => Promise<boolean>;
   health: HealthController;
 }
 
@@ -56,7 +57,7 @@ export function createContext(db: Db, client: Client, config: Config): Applicati
   const writer = new ActiveMQWriter<User>(client, config.destinationName, config.subscriptionName);
   const retryService = new RetryService<User, boolean>(writer.write, log, log);
   const handler = new Handler<User, boolean>(retryWriter.write, validator.validate, retries, errorHandler.error, log, log, retryService.retry, 3, 'retry');
-  const ctx: ApplicationContext = { handle: handler.handle, read: subscriber.subscribe, health };
+  const ctx: ApplicationContext = { handle: handler.handle, read: subscriber.subscribe, health, write: writer.write };
   return ctx;
 }
 export function log(msg: any): void {

src/index.ts

@@ -1,10 +1,12 @@
-import { json } from 'body-parser';
-import dotenv from 'dotenv';
-import express from 'express';
-import http from 'http';
-import { connectToDb } from 'mongodb-extension';
-import { createContext } from './context';
-import { ActiveMQConnection, Config } from './services/activemq';
+import { json } from "body-parser";
+import dotenv from "dotenv";
+import express from "express";
+import http from "http";
+import { connectToDb } from "mongodb-extension";
+import { createContext } from "./context";
+import { ActiveMQConnection, Config } from "./services/activemq";
+import { getBody } from "logger-core";
+
 // import { SendController } from './services/activemq/send';
 // import { printData, retry } from './services/pubsub/retry';
 
@@ -26,8 +28,15 @@ const amqSubscriptionName = process.env.AMQSUBSCRIPTIONNAME;
 app.use(json());
 
 connectToDb(`${mongoURI}`, `${mongoDB}`).then(async (db) => {
-  if (!amqhost || !amqport || !amqUsername || !amqPassword || !amqDestinationName || !amqSubscriptionName) {
-    throw new Error('config wrong!');
+  if (
+    !amqhost ||
+    !amqport ||
+    !amqUsername ||
+    !amqPassword ||
+    !amqDestinationName ||
+    !amqSubscriptionName
+  ) {
+    throw new Error("config wrong!");
   }
   const config: Config = {
     host: amqhost,
@@ -42,13 +51,26 @@ connectToDb(`${mongoURI}`, `${mongoDB}`).then(async (db) => {
   const ctx = createContext(db, client, config);
   ctx.read(ctx.handle);
 
-  app.get('/health', ctx.health.check);
-  app.post('/send', (req, res) => {
-    ctx.handle(req.body)
-    res.writeHead(200, {'Content-Type': 'application/json'});
-    res.end(JSON.stringify({message: 'message was produced'}));
-  });
-  http.createServer(app).listen(port, () => {
-    console.log('Start server at port ' + port);
-  });
+  http
+    .createServer((req, res) => {
+      if (req.url === "/health") {
+        ctx.health.check(req, res);
+      } else if (req.url === "/send") {
+        getBody(req).then((body: any) => {
+          ctx
+            .write(JSON.parse(body))
+            .then(() => {
+              res.writeHead(200, { "Content-Type": "application/json" });
+              res.end(JSON.stringify({ message: "message was produced" }));
+            })
+            .catch((err) => {
+              res.writeHead(500, { "Content-Type": "application/json" });
+              res.end(JSON.stringify({ error: err }));
+            });
+        }).catch(err => console.log(err));
+      }
+    })
+    .listen(port, () => {
+      console.log("Start server at port " + port);
+    });
 });

src/services/activemq/subscriber.ts

@@ -32,7 +32,7 @@ export class ActiveMQSubscriber<T> {
   subscribe(handle: (data: T, attributes?: StringMap) => Promise<number>): void {
     const prefix = this.prefix && this.prefix.length > 0 ? this.prefix : '/';
     const subscribeHeaders = {
-      'destination': `${this.destinationName}${prefix}${this.subscriptionName}`,
+      'destination': `${this.destinationName}${prefix}${this.subscriptionName}`,   
       'ack': this.ackMode,
       'subscription-type': 'ANYCAST'
     };

src/services/activemq/writer.ts

@@ -27,16 +27,19 @@ export class ActiveMQWriter<T> {
     return new Promise((resolve, reject) => {
       const frame = this.client.send(sendHeaders);
       if (this.log) {
-        this.log('produce send data : ' + JSON.stringify(message));
+        this.log('produce send data : ' + JSON.stringify(message.data));
       }
-      const result = frame.write(JSON.stringify(message));
+      const result = frame.write(JSON.stringify(message.data));
       frame.end((err: any) => {
         if (err) {
           console.log('Error activemq: ', err);
-          reject(err);
+          reject(err); 
         }
       });
       resolve(result);
-    });
+    })
+    
+    
+    ;
   }
 }