add support of object storage on error

lib/amqp.js

@@ -202,9 +202,11 @@ class Amqp {
         return this.prepareMessageAndSendToExchange(data, properties, routingKey);
     }
 
-    async sendError(err, properties, originalMessageContent, throttle) {
+    async sendError(err, properties, originalMessage, throttle) {
         const settings = this.settings;
         const headers = properties.headers;
+        const messageContent = _.get(originalMessage , 'content');
+        const messageObjectId = _.get(originalMessage, 'properties.headers.objectId');
 
         const encryptedError = encryptor.encryptMessageContent({
             name: err.name,
@@ -215,9 +217,10 @@ class Amqp {
         const payload = {
             error: encryptedError
         };
-
-        if (originalMessageContent && originalMessageContent !== '') {
-            payload.errorInput = originalMessageContent.toString();
+        if (messageObjectId) {
+            payload.objectId = messageObjectId;
+        } else if (messageContent && messageContent !== '') {
+            payload.errorInput = messageContent.toString();
         }
         const errorPayload = JSON.stringify(payload);
 
@@ -245,7 +248,7 @@ class Amqp {
             return this.sendError(
                 new Error('Rebound limit exceeded'),
                 properties,
-                originalMessage.content
+                originalMessage
             );
         } else {
             properties.expiration = getExpiration(reboundIteration);

lib/sailor.js

@@ -337,7 +337,7 @@ class Sailor {
                 }, 'processMessage emit error');
                 headers.end = new Date().getTime();
                 const props = createAmqpProperties(headers);
-                return self.amqpConnection.sendError(err, props, message.content, self.throttles.error);
+                return self.amqpConnection.sendError(err, props, message, self.throttles.error);
             }
 
             async function onRebound(err) {
@@ -450,9 +450,11 @@ class Sailor {
         }
 
         function onModuleNotFound(err) {
+            const headers = _.clone(outgoingMessageHeaders);
             console.error(err.stack);
-            outgoingMessageHeaders.end = new Date().getTime();
-            self.amqpConnection.sendError(err, outgoingMessageHeaders, message.content);
+            headers.end = new Date().getTime();
+            const props = createAmqpProperties(headers);
+            self.amqpConnection.sendError(err, props, message);
             self.amqpConnection.reject(message);
         }
 

mocha_spec/integration_component/actions/error_action.js

@@ -0,0 +1,18 @@
+'use strict';
+
+exports.process = processAction;
+
+function processAction(msg, cfg) {
+
+    //eslint-disable-next-line no-invalid-this
+    this.emit('error', {
+        statusCode: 200,
+        body: 'Ok',
+        headers: {
+            'content-type': 'text/plain'
+        }
+    });
+    //eslint-disable-next-line no-invalid-this
+    this.emit('end');
+
+}

mocha_spec/integration_component/component.json

@@ -33,6 +33,10 @@
     "http_reply_action": {
       "main": "./actions/http_reply_action.js",
       "title": "Http Reply Action"
+    },
+    "error_action": {
+      "main": "./actions/error_action.js",
+      "title": "Error Action"
     }
   }
 }

mocha_spec/run.spec.js

@@ -201,6 +201,37 @@ describe('Integration Test', () => {
             return new Promise((resolve) => { done = resolve; });
         });
 
+        it('should send objectId on error instead of original message', async () => {
+
+            process.env.ELASTICIO_OBJECT_STORAGE_ENABLED = true;
+            env.ELASTICIO_FUNCTION = 'error_action';
+
+            helpers.mockApiTaskStepResponse();
+
+            const objectStorageGet = nock(process.env.ELASTICIO_OBJECT_STORAGE_URI)
+                .get(`/objects/${objectId}`)
+                .matchHeader('authorization', /Bearer/)
+                .reply(200, await helpers.encryptForObjectStorage(inputMessage), {
+                    'content-type': 'application/octet-stream'
+                });
+
+            let done;
+            amqpHelper.on('data', ({ emittedMessage }) => {
+                expect(emittedMessage.objectId).to.equal(objectId);
+                expect(objectStorageGet.isDone()).to.be.true;
+                done();
+            });
+
+            run = requireRun();
+
+            amqpHelper.publishMessage('', {
+                parentMessageId,
+                traceId
+            }, { objectId });
+
+            return new Promise((resolve) => { done = resolve; });
+        });
+
         it('should get object storage message if error repeat succeed', async () => {
             process.env.ELASTICIO_OBJECT_STORAGE_ENABLED = true;
 

spec/amqp.spec.js

@@ -312,7 +312,7 @@ describe('AMQP', () => {
             }
         };
 
-        amqp.sendError(new Error('Test error'), props, message.content);
+        amqp.sendError(new Error('Test error'), props, message);
 
         expect(amqp.publishChannel.publish).toHaveBeenCalled();
         expect(amqp.publishChannel.publish.callCount).toEqual(1);
@@ -371,7 +371,7 @@ describe('AMQP', () => {
         };
 
         async function test() {
-            await amqp.sendError(new Error('Test error'), props, message.content);
+            await amqp.sendError(new Error('Test error'), props, message);
         }
 
         test().then(() => {
@@ -497,6 +497,52 @@ describe('AMQP', () => {
         expect(publishParameters[3]).toEqual(props);
     });
 
+    it('Should send objectId instead of message when process error', () => {
+
+        const amqp = new Amqp(settings);
+        const objectId = 'aef703e0-7ebf-456f-a7b9-20b647ab43cd';
+        amqp.publishChannel = jasmine.createSpyObj('publishChannel', ['publish']);
+
+
+        const props = {
+            contentType: 'application/json',
+            contentEncoding: 'utf8',
+            mandatory: true,
+            headers: {
+                taskId: 'task1234567890',
+                stepId: 'step_456'
+            }
+        };
+
+        const msgWithObjectId = _.cloneDeep(message);
+        msgWithObjectId.properties.headers.objectId = objectId;
+
+        amqp.sendError(new Error('Test error'), props, msgWithObjectId);
+
+        expect(amqp.publishChannel.publish).toHaveBeenCalled();
+        expect(amqp.publishChannel.publish.callCount).toEqual(1);
+
+        const publishParameters = amqp.publishChannel.publish.calls[0].args;
+        expect(publishParameters).toEqual([
+            settings.PUBLISH_MESSAGES_TO,
+            settings.ERROR_ROUTING_KEY,
+            jasmine.any(Object),
+            props
+        ]);
+
+        const payload = JSON.parse(publishParameters[2].toString());
+        payload.error = encryptor.decryptMessageContent(payload.error);
+
+        expect(payload).toEqual({
+            error: {
+                name: 'Error',
+                message: 'Test error',
+                stack: jasmine.any(String)
+            },
+            objectId
+        });
+    });
+
     it('Should send message to rebounds when rebound happened', () => {
 
         const amqp = new Amqp(settings);

spec/sailor.spec.js

@@ -590,7 +590,7 @@ describe('Sailor', () => {
                     expect(fakeAMQPConnection.sendError).toHaveBeenCalled();
                     expect(fakeAMQPConnection.sendError.calls[0].args[0].message).toEqual('Some component error');
                     expect(fakeAMQPConnection.sendError.calls[0].args[0].stack).not.toBeUndefined();
-                    expect(fakeAMQPConnection.sendError.calls[0].args[2]).toEqual(message.content);
+                    expect(fakeAMQPConnection.sendError.calls[0].args[2]).toEqual(message);
 
                     expect(fakeAMQPConnection.reject).toHaveBeenCalled();
                     expect(fakeAMQPConnection.reject.callCount).toEqual(1);
@@ -654,7 +654,7 @@ describe('Sailor', () => {
                         /* eslint-enable */
                     );
                     expect(fakeAMQPConnection.sendError.calls[0].args[0].stack).not.toBeUndefined();
-                    expect(fakeAMQPConnection.sendError.calls[0].args[2]).toEqual(message.content);
+                    expect(fakeAMQPConnection.sendError.calls[0].args[2]).toEqual(message);
 
                     expect(fakeAMQPConnection.reject).toHaveBeenCalled();
                     expect(fakeAMQPConnection.reject.callCount).toEqual(1);