@highoutput/amqp - npm Package Compare versions

Comparing version 0.0.13 to 0.0.14

build/lib/client.js

@@ -10,3 +10,5 @@ "use strict";
 const async_group_1 = __importDefault(require("@highoutput/async-group"));
+const error_1 = __importDefault(require("@highoutput/error"));
 const logger_1 = __importDefault(require("./logger"));
+const util_1 = require("./util");
 class Client {
@@ -35,8 +37,12 @@ constructor(connection, scope, options) {
         const correlationId = v4_1.default();
+        const body = {
+            correlationId,
+            arguments: this.options.serialize ? util_1.serialize(args) : args,
+            timestamp: Date.now(),
+        };
+        logger_1.default.tag(['client', 'request']).verbose(body);
         this.sender.send({
             reply_to: (_a = this.receiver) === null || _a === void 0 ? void 0 : _a.source.address,
             correlation_id: correlationId,
-            body: {
-                parameters: args,
-            },
+            body,
         });
@@ -69,21 +75,20 @@ if (this.options.noResponse) {
     async start() {
-        this.sender = this.connection.open_sender({
-            target: {
-                address: this.scope,
-            },
-        });
-        this.sender.on('sender_error', (context) => {
-            var _a;
-            logger_1.default.tag('client').error((_a = context.error) === null || _a === void 0 ? void 0 : _a.message);
-        });
-        this.receiver = this.connection.open_receiver({
-            source: {
-                address: `${this.scope}:${this.id}`,
-                dynamic: true,
-            },
-        });
+        const [sender, receiver] = await Promise.all([
+            util_1.openSender(this.connection, {
+                target: {
+                    address: this.scope,
+                },
+            }),
+            util_1.openReceiver(this.connection, {
+                source: {
+                    address: `${this.scope}:${this.id}`,
+                    dynamic: true,
+                },
+            }),
+        ]);
+        this.sender = sender;
+        this.receiver = receiver;
         this.receiver.on('message', (context) => {
             var _a, _b;
-            const body = (_a = context.message) === null || _a === void 0 ? void 0 : _a.body;
-            logger_1.default.tag(['client', 'message']).info(body);
+            let body = (_a = context.message) === null || _a === void 0 ? void 0 : _a.body;
             const callback = this.callbacks.get((_b = context.message) === null || _b === void 0 ? void 0 : _b.correlation_id);
@@ -93,4 +98,15 @@ if (!callback) {
             }
+            body = Object.assign(Object.assign({}, body), { result: this.options.deserialize ? util_1.deserialize(body.result) : body.result });
+            logger_1.default.tag(['client', 'response']).verbose(body);
             if (body.error) {
-                callback.reject(body.error);
+                let error;
+                const deserialized = util_1.deserialize(body.error);
+                const meta = Object.assign(Object.assign({}, ramda_1.default.omit(['id', 'name', 'message', 'stack', 'service'])(deserialized)), { original: deserialized });
+                if (body.error.name === 'AppError') {
+                    error = new error_1.default(body.error.code, body.error.message, meta);
+                }
+                else {
+                    error = new error_1.default('WORKER_ERROR', body.error.message, meta);
+                }
+                callback.reject(error);
                 return;
@@ -100,28 +116,10 @@ }
         });
-        await Promise.all([
-            new Promise((resolve) => {
-                this.sender.once('sender_open', () => {
-                    resolve();
-                });
-            }),
-            new Promise((resolve) => {
-                this.receiver.once('receiver_open', () => {
-                    resolve();
-                });
-            }),
-        ]);
     }
     async stop() {
         if (this.sender && this.sender.is_open()) {
-            this.sender.close();
-            await new Promise((resolve) => {
-                this.connection.once('sender_close', resolve);
-            });
+            await util_1.closeSender(this.sender);
         }
         await this.asyncGroup.wait();
         if (this.receiver && this.receiver.is_open()) {
-            this.receiver.close();
-            await new Promise((resolve) => {
-                this.connection.once('receiver_close', resolve);
-            });
+            await util_1.closeReceiver(this.receiver);
         }
@@ -128,0 +126,0 @@ }

build/lib/worker.js

@@ -10,2 +10,3 @@ "use strict";
 const logger_1 = __importDefault(require("./logger"));
+const util_1 = require("./util");
 class Worker {
@@ -29,15 +30,7 @@ constructor(connection, scope, handler, options) {
         if (!promise) {
-            promise = (async () => {
-                const sender = this.connection.open_sender({
-                    target: {
-                        address,
-                    },
-                });
-                await new Promise((resolve) => {
-                    sender.once('sender_open', () => {
-                        resolve();
-                    });
-                });
-                return sender;
-            })();
+            promise = util_1.openSender(this.connection, {
+                target: {
+                    address,
+                },
+            });
             this.senders.set(address, promise);
@@ -52,3 +45,4 @@ }
         }
-        logger_1.default.tag(['worker', 'message']).info(message.body);
+        const request = Object.assign(Object.assign({}, message.body), { arguments: this.options.deserialize ? util_1.deserialize(message.body.arguments) : message.body.arguments });
+        logger_1.default.tag(['worker', 'request']).info(request);
         const sender = await this.getSender(message.reply_to);
@@ -58,17 +52,21 @@ let result = null;
         try {
-            result = await this.handler(...message.body.parameters);
+            result = await this.handler(...request.arguments);
         }
         catch (err) {
-            error = serialize_error_1.serializeError(err);
+            error = util_1.serialize(serialize_error_1.serializeError(err));
         }
+        const response = {
+            correlationId: message.correlation_id,
+            result: this.options.serialize ? util_1.serialize(result) : result,
+            error,
+            timestamp: Date.now(),
+        };
+        logger_1.default.tag(['worker', 'response']).info(request);
         sender.send({
             correlation_id: message.correlation_id,
-            body: {
-                result,
-                error,
-            },
+            body: response,
         });
     }
     async start() {
-        this.receiver = this.connection.open_receiver({
+        this.receiver = await util_1.openReceiver(this.connection, {
             source: {
@@ -91,6 +89,3 @@ address: this.scope,
         if (this.receiver && this.receiver.is_open()) {
-            this.receiver.close();
-            await new Promise((resolve) => {
-                this.connection.once('receiver_close', resolve);
-            });
+            await util_1.closeReceiver(this.receiver);
         }
@@ -101,6 +96,3 @@ await this.asyncGroup.wait();
             if (sender.is_open()) {
-                sender.close();
-                await new Promise((resolve) => {
-                    this.connection.once('sender_close', resolve);
-                });
+                await util_1.closeSender(sender);
             }
@@ -107,0 +99,0 @@ }));

package.json

 {
   "name": "@highoutput/amqp",
-  "version": "0.0.13",
+  "version": "0.0.14",
   "description": "A simplified abstraction of the AMQP 1.0 protocol",
@@ -51,3 +51,3 @@ "keywords": [
   },
-  "gitHead": "e3454394788771b2fdd1d8d6ca6dee5c4deec630"
+  "gitHead": "3732cb2574dd1f18903bf623aea7406438f10a76"
 }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

38263

increased by24.86%

Number of package files

24

increased by20%

Lines of code

549

increased by23.09%

No dependency changes