Make sailor to use object-storage-client

lib/amqp.js

@@ -1,7 +1,7 @@
 const log = require('./logging.js');
 const amqplib = require('amqplib');
 const encryptor = require('./encryptor.js');
-const ObjectStorage = require('./objectStorage.js');
+const { ObjectStorage } = require('@elastic.io/object-storage-client');
 const co = require('co');
 const _ = require('lodash');
 const eventToPromise = require('event-to-promise');
@@ -12,7 +12,13 @@ const HEADER_ERROR_RESPONSE = 'x-eio-error-response';
 class Amqp {
     constructor(settings) {
         this.settings = settings;
-        this.objectStorage = new ObjectStorage(settings);
+        this.objectStorage = new ObjectStorage({
+            uri: settings.OBJECT_STORAGE_URI,
+            cipher: {
+                key: process.env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD,
+                iv: process.env.ELASTICIO_MESSAGE_CRYPTO_IV
+            }
+        });
     }
 
     connect(uri) {
@@ -61,7 +67,7 @@ class Amqp {
             return encryptor.decryptMessageContent(message.content);
         }
 
-        return await this.objectStorage.getObject(objectId);
+        return await this.objectStorage.getAsJSON(objectId, this.settings.OBJECT_STORAGE_TOKEN);
     }
 
     async encryptMessage(data, properties) {
@@ -71,7 +77,7 @@ class Amqp {
 
         let message = '';
         try {
-            properties.headers.objectId = await this.objectStorage.addObject(data);
+            properties.headers.objectId = await this.objectStorage.addAsJSON(data, this.settings.OBJECT_STORAGE_TOKEN);
         } catch (e) {
             log.error('Failed to add message to object storage: %s', e);
             message = encryptor.encryptMessageContent(data);

mocha_spec/integration_helpers.js

@@ -6,7 +6,9 @@ const { EventEmitter } = require('events');
 const PREFIX = 'sailor_nodejs_integration_test';
 const nock = require('nock');
 const getStream = require('get-stream');
-const encryptor = require('../lib/encryptor.js');
+const { Readable } = require('stream');
+const { Message } = require('@elastic.io/object-storage-client');
+const encryptor = require('../lib/encryptor');
 
 const env = process.env;
 
@@ -134,7 +136,7 @@ class AmqpHelper extends EventEmitter {
         this.publishChannel.ack(message);
 
         const content = message.content.toString();
-        const emittedMessage = content ? JSON.parse(content) : content;
+        const emittedMessage = content ? encryptor.decryptMessageContent(message.content) : content;
 
         const data = {
             properties: message.properties,
@@ -184,6 +186,9 @@ function prepareEnv() {
     env.ELASTICIO_OBJECT_STORAGE_URI = 'http://ma.es.ter';
     env.ELASTICIO_OBJECT_STORAGE_ENABLED = '';
     env.ELASTICIO_OBJECT_STORAGE_TOKEN = 'jwt';
+
+    env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD = 'testCryptoPassword';
+    env.ELASTICIO_MESSAGE_CRYPTO_IV = 'iv=any16_symbols';
 }
 
 function mockApiTaskStepResponse(response) {
@@ -203,7 +208,14 @@ function mockApiTaskStepResponse(response) {
 }
 
 async function encryptForObjectStorage(input) {
-    const stream = encryptor.encryptMessageContentStream(input);
+    const dataStream = new Readable();
+    dataStream.push(JSON.stringify(input));
+    dataStream.push(null);
+    const message = new Message({
+        key: env.ELASTICIO_MESSAGE_CRYPTO_PASSWORD,
+        iv: env.ELASTICIO_MESSAGE_CRYPTO_IV
+    });
+    const stream = message.packStream(dataStream);
     return await getStream.buffer(stream);
 }
 

mocha_spec/run.spec.js

@@ -245,7 +245,6 @@ describe('Integration Test', () => {
                 .get('/customers')
                 .reply(200, customers);
 
-            const log = sinon.stub(logging, 'warn');
             const objectStorageGet = nock(process.env.ELASTICIO_OBJECT_STORAGE_URI)
                 .get(`/objects/${objectId}`)
                 .matchHeader('authorization', /Bearer/)
@@ -282,7 +281,6 @@ describe('Integration Test', () => {
                 expect(body).to.be.null;
                 expect(objectStorageGet.isDone()).to.be.true;
                 expect(objectStoragePut.isDone()).to.be.true;
-                expect(log.getCall(1).args[1].toString()).to.include('400');
                 done();
             });