- Add test coverage for transaction mirroring. In the process, I simplified the implementation to remove a check that was intended to prevent looping transactions but in practice would only be effective in limited cases, and would actually prevent mirrors running on different ports behind the same NAT address from replicating.

- Add another protocol test case to catch all of the branches in client_stream_processor

Author: stephen-palmer

lib/cache/cache_base.js

@@ -10,6 +10,7 @@ const { promisify } = require('util');
 const loki = require('lokijs');
 const ReliabilityManager = require('./reliability_manager');
 const _ = require('lodash');
+const { Writable } = require('stream');
 
 const kDbName = 'cache.db';
 
@@ -176,7 +177,7 @@ class CacheBase extends EventEmitter {
      * @returns {Promise<any>}
      */
     createPutTransaction(guid, hash) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve(new PutTransaction(guid, hash));
     }
 
     /**
@@ -300,7 +301,9 @@ class PutTransaction extends EventEmitter {
      * @returns {Promise<any>}
      */
     getWriteStream(type, size) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve(new Writable({
+            write(chunk, encoding, cb){ setImmediate(cb); }
+        }));
     }
 
     async invalidate() {
@@ -313,7 +316,7 @@ class PutTransaction extends EventEmitter {
      * @returns {Promise<any>}
      */
     async writeFilesToPath(targetPath) {
-        return Promise.reject(new Error("Not implemented"));
+        return Promise.resolve();
     }
 }
 

lib/server/server.js

@@ -20,13 +20,9 @@ class CacheServer {
             this._port = consts.DEFAULT_PORT;
 
         this._server = null;
+        this._mirrorConfig = options.mirror ? [].concat(options.mirror) : [];
         this._mirrors = [];
 
-        if(options.mirror) {
-            options.mirror = [].concat(options.mirror);
-            this._mirrors = options.mirror.map(m => new TransactionMirror(m, cache));
-        }
-
         this.allowIpv6 = options.allowIpv6;
         this._errCallback = null;
     }
@@ -57,6 +53,14 @@ class CacheServer {
         return this._server;
     }
 
+    /**
+     *
+     * @returns {Array|*}
+     */
+    get mirrors() {
+        return this._mirrors;
+    }
+
     /**
      *
      * @param {Function} cb
@@ -79,12 +83,10 @@ class CacheServer {
             const cmdProc = new CommandProcessor(this.cache);
             const streamProc = new ClientStreamProcessor({clientAddress: `${socket.remoteAddress}:${socket.remotePort}`});
 
-            const mirrors = this._mirrors;
-            if(mirrors.length > 0) {
+            if(this._mirrors.length > 0) {
                 cmdProc.on('onTransactionEnd', (trx) => {
-                    mirrors.forEach(m => {
-                        if(m.address !== socket.remoteAddress)
-                            m.queueTransaction(trx);
+                    this._mirrors.forEach(m => {
+                        m.queueTransaction(trx);
                     });
                 });
             }
@@ -103,14 +105,16 @@ class CacheServer {
                 .pipe(socket);     // Connect back to socket to send files
 
             socket['commandProcessor'] = cmdProc;
-        });
-
-        this._server.on('error', err => {
+        }).on('error', err => {
             if (err.code === 'EADDRINUSE') {
                 helpers.log(consts.LOG_ERR, `Port ${this.port} is already in use...`);
             }
 
             if (this._errCallback && typeof(this._errCallback === 'function')) { this._errCallback(err); }
+        }).on('listening', () => {
+            this._mirrors = this._mirrorConfig
+                .filter(m => m && m.port && m.host)
+                .map(m => new TransactionMirror(m, this.cache));
         });
 
         return new Promise(resolve => {

lib/server/transaction_mirror.js

@@ -37,6 +37,10 @@ class TransactionMirror {
         return this._connectOptions.host;
     }
 
+    get port() {
+        return this._connectOptions.port;
+    }
+
     _connect() {
         helpers.log(consts.LOG_INFO, `[TransactionMirror] Connecting to ${this._connectOptions.host}:${this._connectOptions.port}`);
         return this._client.connect();

test/cache_base.js

@@ -8,6 +8,7 @@ const path = require('path');
 const randomBuffer = require('./test_utils').randomBuffer;
 const consts = require('../lib/constants');
 const sinon = require('sinon');
+const { Writable } = require('stream');
 
 describe("Cache: Base Class", () => {
     let cache;
@@ -124,9 +125,9 @@ describe("Cache: Base Class", () => {
     });
 
     describe("createPutTransaction", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return cache.createPutTransaction()
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return an instance of a PutTransaction", async () => {
+            const t = await cache.createPutTransaction();
+            assert.ok(t instanceof PutTransaction);
         });
     });
 
@@ -264,16 +265,16 @@ describe("PutTransaction: Base Class", () => {
     });
 
     describe("getWriteStream", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return trx.getWriteStream(consts.FILE_TYPE.INFO, 0)
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return a Writable stream", async() => {
+            const s = await trx.getWriteStream(consts.FILE_TYPE.INFO, 0);
+            assert.ok(s instanceof Writable);
         });
     });
 
     describe("writeFilesToPath", () => {
-        it("should require override implementation in subclasses by returning an error", () => {
-            return trx.writeFilesToPath()
-                .then(() => { throw new Error("Expected error!"); }, () => {});
+        it("should return a promise", () => {
+            const p = trx.writeFilesToPath();
+            assert.ok(p instanceof Promise);
         });
     });
 });

test/protocol.js

@@ -323,6 +323,7 @@ describe("Protocol", () => {
                 });
 
                 const tests = [
+                    {cmd: cmd.getAsset, blob: self.data.bin, type: 'bin', packetSize: 1},
                     {cmd: cmd.getAsset, blob: self.data.bin, type: 'bin', packetSize: SMALL_PACKET_SIZE},
                     {cmd: cmd.getInfo, blob: self.data.info, type: 'info', packetSize: MED_PACKET_SIZE},
                     {cmd: cmd.getResource, blob: self.data.resource, type: 'resource', packetSize: LARGE_PACKET_SIZE}

test/server.js

@@ -6,15 +6,61 @@ const os = require('os');
 const helpers = require('../lib/helpers');
 const consts = require('../lib/constants');
 const CacheServer = require('../lib/server/server');
-const Cache = require('../lib/cache/cache_base').CacheBase;
-const sleep = require('./test_utils').sleep;
-const cmd = require('./test_utils').cmd;
+const CacheBase = require('../lib/cache/cache_base').CacheBase;
+const TransactionMirror = require('../lib/server/transaction_mirror');
+const { generateCommandData, encodeCommand, clientWrite, sleep, cmd } = require('./test_utils');
+const sinon = require('sinon');
 
-const cache = new Cache();
-const server = new CacheServer(cache, {port: 0});
+const cache = new CacheBase();
 let client;
 
+describe("Server constructor", function() {
+    it("should use the default port if no port is specified in options", () => {
+        const s = new CacheServer(cache, { mirror:[] });
+        assert.strictEqual(s.port, consts.DEFAULT_PORT);
+    });
+});
+
+describe("Server mirroring", function() {
+    const server = new CacheServer(cache, {
+        port: 0,
+        mirror: [{host: "127.0.0.1", port: 8126}, {host: "1.2.3.4", port: 8126}, {host: "4.3.2.1", port: 8126}]
+    });
+
+    before(function () {
+        return server.start(err => { return Promise.reject(err); });
+    });
+
+    after(function() {
+        server.stop();
+    });
+
+    beforeEach(function (done) {
+        client = net.connect({port: server.port}, done);
+    });
+
+    afterEach(() => client.end());
+
+    it("should mirror transactions to the configured list of mirrors", async () => {
+        const spies = server.mirrors.map(m => {
+            return sinon.spy(m, "queueTransaction");
+        });
+
+        const testData = generateCommandData();
+
+        const buf = Buffer.from(helpers.encodeInt32(consts.PROTOCOL_VERSION) +
+            encodeCommand(cmd.transactionStart, testData.guid, testData.hash) +
+            encodeCommand(cmd.putAsset, null, null, testData.bin) +
+            encodeCommand(cmd.transactionEnd), 'ascii');
+
+        await clientWrite(client, buf);
+
+        spies.forEach(s => assert(s.calledOnce));
+    });
+});
+
 describe("Server common", function() {
+    const server = new CacheServer(cache, {port: 0});
 
     before(function () {
         this._defaultErrCallback = err => assert(!err, `Cache Server reported error! ${err}`);
@@ -71,7 +117,7 @@ describe("Server common", function() {
     describe("Ipv6", function() {
         const ipv6Server = new CacheServer(cache, {port: 0, allowIpv6: true});
 
-        before(function () {  
+        before(function () {
             const interfaces = os.networkInterfaces();
             let ipv6Available = false;
             Object.keys(interfaces).forEach(function (interfaceName){
@@ -85,15 +131,15 @@ describe("Server common", function() {
             if(!ipv6Available){
                 console.log("Skipping IPv6 tests because IPv6 is not available on this machine");
                 this.skip();
-            }   
+            }
 
             return ipv6Server.start(err => assert(!err, `Cache Server reported error! ${err}`));
         });
-    
+
         after(function() {
             ipv6Server.stop();
         });
-    
+
         it("should bind to ipv6 when allowed", function(done) {
             const serverAddress = ipv6Server._server.address();
             assert.strictEqual(serverAddress.family, "IPv6");
@@ -108,7 +154,7 @@ describe("Server common", function() {
         before(function () {
             return ipv4Server.start(err => assert(!err, `Cache Server reported error! ${err}`));
         });
-    
+
         after(function() {
             ipv4Server.stop();
         });

test/unity_cache_server.js

@@ -208,6 +208,9 @@ describe("Unity Cache Server bootstrap", () => {
                             cachePath: tmpPath
                         }
                     }
+                },
+                Server: {
+                    port: 0
                 }
             });