[FABCN-409] Chaincode gRPC server w/o TLS (#159)

This patch adds the ChaincodeServer class to support the server mode of
chaincode.
It also adds the server() method, which creates a new instance of
the ChaincodeServer class, to the Shim class, the entrypoint of the
fabric-shim library.

Signed-off-by: Taku Shimosawa <taku.shimosawa@hal.hitachi.com>

libraries/fabric-shim/lib/chaincode.js

@@ -15,6 +15,7 @@ const utils = require('./utils/utils');
 
 const logger = Logger.getLogger('lib/chaincode.js');
 const {ChaincodeSupportClient} = require('./handler');
+const ChaincodeServer = require('./server');
 const Iterators = require('./iterators');
 const ChaincodeStub = require('./stub');
 const KeyEndorsementPolicy = require('./utils/statebased');
@@ -194,6 +195,22 @@ class Shim {
 
         return Logger.getLogger(name);
     }
+
+    /**
+     * Returns a new Chaincode server. Should be called when the chaincode is launched in a server mode.
+     * @static
+     * @param {ChaincodeInterface} chaincode User-provided object that must implement <code>ChaincodeInterface</code>
+     * @param {ChaincodeSeverOpts} serverOpts Chaincode server options
+     */
+    static server(chaincode, serverOpts) {
+        return new ChaincodeServer(chaincode, serverOpts);
+    }
+    /**
+     * @typedef {Object} ChaincodeServerOpts
+     * @property {string} ccid Chaincode ID
+     * @property {string} address Listen address for the server
+     * @property {Object} tlsProps TLS properties. To be implemented. Should be null if TLS is not used.
+     */
 }
 
 // special OID used by Fabric to save attributes in X.509 certificates

libraries/fabric-shim/lib/server.js

@@ -0,0 +1,111 @@
+/*
+# Copyright Hitachi America, Ltd. All Rights Reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+*/
+
+'use strict';
+
+const protoLoader = require('@grpc/proto-loader');
+const grpc = require('@grpc/grpc-js');
+const path = require('path');
+
+const fabprotos = require('../bundle');
+const {ChaincodeMessageHandler} = require('./handler');
+const logger = require('./logger').getLogger('lib/server.js');
+
+const PROTO_PATH = path.resolve(__dirname, '..', 'protos', 'peer', 'chaincode_shim.proto');
+const packageDefinition = protoLoader.loadSync(
+    PROTO_PATH,
+    {
+        keepCase: true,
+        longs: String,
+        enums: String,
+        defaults: true,
+        oneofs: true,
+        includeDirs: [
+            path.resolve(__dirname, '..', 'google-protos'),
+            path.resolve(__dirname, '..', 'protos')
+        ]
+    }
+);
+const protoDescriptor = grpc.loadPackageDefinition(packageDefinition);
+
+/**
+ * The ChaincodeServer class represents a chaincode gRPC server, which waits for connections from peers.
+ */
+class ChaincodeServer {
+    constructor(chaincode, serverOpts) {
+        // Validate arguments
+        if (typeof chaincode !== 'object' || chaincode === null) {
+            throw new Error('Missing required argument: chaincode');
+        }
+        if (typeof serverOpts !== 'object' || serverOpts === null) {
+            throw new Error('Missing required argument: serverOpts');
+        }
+        if (typeof chaincode.Init !== 'function' || typeof chaincode.Invoke !== 'function') {
+            throw new Error('The "chaincode" argument must implement Init() and Invoke() methods');
+        }
+        if (typeof serverOpts.ccid !== 'string') {
+            throw new Error('Missing required property in severOpts: ccid');
+        }
+        if (typeof serverOpts.address !== 'string') {
+            throw new Error('Missing required property in severOpts: address');
+        }
+
+        // Create GRPC Server and register RPC handler
+        this._server = new grpc.Server();
+        const self = this;
+
+        this._server.addService(protoDescriptor.protos.Chaincode.service, {
+            connect: (stream) => {
+                self.connect(stream);
+            }
+        });
+
+        this._serverOpts = serverOpts;
+        this._chaincode = chaincode;
+    }
+
+    start() {
+        return new Promise((resolve, reject) => {
+            logger.debug('ChaincodeServer trying to bind to ' + this._serverOpts.address);
+
+            // TODO: TLS Support
+            this._server.bindAsync(this._serverOpts.address, grpc.ServerCredentials.createInsecure(), (error, port) => {
+                if (!error) {
+                    logger.debug('ChaincodeServer successfully bound to ' + port);
+
+                    this._server.start();
+                    logger.debug('ChaincodeServer started.');
+
+                    resolve();
+                } else {
+                    logger.error('ChaincodeServer failed to bind to ' + this._serverOpts.address);
+                    reject(error);
+                }
+            });
+        });
+    }
+
+    connect(stream) {
+        logger.debug('ChaincodeServer.connect called.');
+
+        try {
+            const client = new ChaincodeMessageHandler(stream, this._chaincode);
+            const chaincodeID = {
+                name: this._serverOpts.ccid
+            };
+
+            logger.debug('Start chatting with a peer through a new stream. Chaincode ID = ' + this._serverOpts.ccid);
+            client.chat({
+                type: fabprotos.protos.ChaincodeMessage.Type.REGISTER,
+                payload: fabprotos.protos.ChaincodeID.encode(chaincodeID).finish()
+            });
+        } catch (e) {
+            logger.warn('connection from peer failed: ' + e);
+        }
+    }
+}
+
+module.exports = ChaincodeServer;

libraries/fabric-shim/test/unit/chaincode.js

@@ -371,4 +371,23 @@ describe('Chaincode', () => {
             Logger.getLogger.restore();
         });
     });
+
+    describe('server()', () => {
+        before(() => {
+            Chaincode = rewire(chaincodePath);
+        });
+
+        it ('should create a ChaincodeServer instance', () => {
+            const mockObj = {_chaincode: {}, _serverOpts: {}};
+            const serverStub = sinon.stub().returns(mockObj);
+            Chaincode.__set__('ChaincodeServer', serverStub);
+
+            const mockChaincode = new Chaincode.ChaincodeInterface();
+            const serverOpts = {ccid: 'example-cc-id:1', address: '0.0.0.0:9999'};
+
+            expect(Chaincode.server(mockChaincode, serverOpts)).to.deep.equal(mockObj);
+            expect(serverStub.calledOnce).to.be.true;
+            expect(serverStub.firstCall.args).to.deep.equal([mockChaincode, serverOpts]);
+        });
+    });
 });

libraries/fabric-shim/test/unit/server.js

@@ -0,0 +1,155 @@
+/*
+# Copyright Hitachi America, Ltd. All Rights Reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+*/
+/* global describe it beforeEach afterEach before after */
+'use strict';
+
+const sinon = require('sinon');
+const chai = require('chai');
+chai.use(require('chai-as-promised'));
+const expect = chai.expect;
+const rewire = require('rewire');
+const fabprotos = require('../../bundle');
+const grpc = require('@grpc/grpc-js');
+
+const serverPath = '../../lib/server';
+let ChaincodeServer = rewire(serverPath);
+
+const mockChaincode = {Init: () => {}, Invoke: () => {}};
+
+describe('ChaincodeServer', () => {
+    const mockGrpcServerInstance = {
+        addService: sinon.stub()
+    };
+    let grpcServerStub;
+    const serverOpts = {
+        ccid: 'example-chaincode-id',
+        address: '0.0.0.0:9999',
+        serverOpts: {}
+    };
+
+    beforeEach(() => {
+        grpcServerStub = sinon.stub(grpc, 'Server').returns(mockGrpcServerInstance);
+    });
+    afterEach(() => {
+        grpcServerStub.restore();
+    });
+
+    describe('constructor', () => {
+        it('should create a gRPC server instance and call addService in the constructor', () => {
+            const server = new ChaincodeServer(mockChaincode, serverOpts);
+
+            expect(grpcServerStub.calledOnce).to.be.ok;
+            expect(server._server).to.deep.equal(mockGrpcServerInstance);
+            expect(server._server.addService.calledOnce).to.be.ok;
+            expect(server._chaincode).to.deep.equal(mockChaincode);
+            expect(server._serverOpts).to.deep.equal(serverOpts);
+        });
+
+        it('should throw an error when chaincode is missing', () => {
+            expect(() => new ChaincodeServer(null, serverOpts)).to.throw('Missing required argument: chaincode');
+        });
+        it('should throw an error when chaincode implements only Invoke', () => {
+            expect(() => new ChaincodeServer({Invoke: sinon.stub()}, serverOpts))
+                .to.throw('The "chaincode" argument must implement Init() and Invoke() methods');
+        });
+        it('should throw an error when chaincode implements only Init', () => {
+            expect(() => new ChaincodeServer({Init: sinon.stub()}, serverOpts))
+                .to.throw('The "chaincode" argument must implement Init() and Invoke() methods');
+        });
+        it('should throw an error when serverOpts is missing', () => {
+            expect(() => new ChaincodeServer(mockChaincode)).to.throw('Missing required argument: serverOpts');
+        });
+        it('should throw an error when serverOpts.ccid is missing', () => {
+            expect(() => new ChaincodeServer(mockChaincode, {})).to.throw('Missing required property in severOpts: ccid');
+        });
+        it('should throw an error when serverOpts.address is missing', () => {
+            expect(() => new ChaincodeServer(mockChaincode, {ccid: 'some id'})).to.throw('Missing required property in severOpts: address');
+        });
+    });
+
+    describe('start()', () => {
+        const mockCredential = {};
+        let insecureCredentialStub;
+
+        beforeEach(() => {
+            insecureCredentialStub = sinon.stub(grpc.ServerCredentials, 'createInsecure').returns(mockCredential);
+        });
+        afterEach(() => {
+            insecureCredentialStub.restore();
+        });
+
+        it('should call bindAsync and start', async () => {
+            const server = new ChaincodeServer(mockChaincode, serverOpts);
+
+            server._server = {
+                bindAsync: sinon.stub().callsFake((address, credential, callback) => {
+                    callback(null, 9999);
+                }),
+                start: sinon.stub()
+            };
+
+            expect(await server.start()).not.to.throw;
+            expect(server._server.bindAsync.calledOnce).to.be.ok;
+            expect(server._server.bindAsync.firstCall.args[0]).to.equal(serverOpts.address);
+            expect(server._server.bindAsync.firstCall.args[1]).to.equal(mockCredential);
+            expect(server._server.start.calledOnce).to.be.ok;
+        });
+
+        it('should throw if bindAsync fails', async () => {
+            const server = new ChaincodeServer(mockChaincode, serverOpts);
+
+            server._server = {
+                bindAsync: sinon.stub().callsFake((address, credential, callback) => {
+                    callback('failed to bind', 9999);
+                }),
+                start: sinon.stub()
+            };
+            expect(server.start()).to.eventually.be.rejectedWith('failed to bind');
+        });
+    });
+
+    describe('connect()', () => {
+        it('should call connect', () => {
+            const mockHandler = {
+                chat: sinon.stub()
+            };
+            const mockHandlerStub = sinon.stub().returns(mockHandler);
+            ChaincodeServer.__set__('ChaincodeMessageHandler', mockHandlerStub);
+
+            const server = new ChaincodeServer(mockChaincode, serverOpts);
+
+            const serviceImpl = server._server.addService.firstCall.args[1];
+            const mockStream = {on: sinon.stub(), write: sinon.stub()};
+
+            expect(serviceImpl.connect(mockStream)).not.to.throw;
+            expect(mockHandlerStub.calledOnce).to.be.ok;
+            expect(mockHandler.chat.calledOnce).to.be.ok;
+            expect(mockHandler.chat.firstCall.args).to.deep.equal([{
+                type: fabprotos.protos.ChaincodeMessage.Type.REGISTER,
+                payload: fabprotos.protos.ChaincodeID.encode({
+                    name: 'example-chaincode-id'
+                }).finish()
+            }]);
+        });
+
+        it('should not throw even if chat fails', () => {
+            const mockHandler = {
+                chat: sinon.stub().throws(new Error('Some error from chat'))
+            };
+            const mockHandlerStub = sinon.stub().returns(mockHandler);
+            ChaincodeServer.__set__('ChaincodeMessageHandler', mockHandlerStub);
+
+            const server = new ChaincodeServer(mockChaincode, serverOpts);
+
+            const serviceImpl = server._server.addService.firstCall.args[1];
+            const mockStream = {on: sinon.stub(), write: sinon.stub()};
+
+            expect(serviceImpl.connect(mockStream)).not.to.throw;
+            expect(mockHandlerStub.calledOnce).to.be.ok;
+            expect(mockHandler.chat.calledOnce).to.be.ok;
+        });
+    });
+});