diff --git a/fabric-client/lib/EventHub.js b/fabric-client/lib/EventHub.js index 086ea15239..048ef97e58 100644 --- a/fabric-client/lib/EventHub.js +++ b/fabric-client/lib/EventHub.js @@ -171,7 +171,6 @@ var EventHub = class { // grpc chat streaming interface this._stream = null; // fabric connection state of this eventhub - this._connect_called = false; this._connected = false; this._connect_running = false; // should this event hub reconnect on registrations @@ -255,7 +254,7 @@ var EventHub = class { * the connection to the peer event hub */ _connect(force) { - this._connect_called = true; + logger.debug('_connect - start - %s', new Date()); if(this._connect_running) { logger.debug('_connect - connect is running'); return; @@ -344,6 +343,7 @@ var EventHub = class { logger.debug('on.error - event stream:%s _current_stream:%s',stream_id, self._current_stream); if(stream_id != self._current_stream) { logger.debug('on.error - incoming event was from a canceled stream'); + logger.debug('on.error - %s %s',new Date(),err); return; } @@ -387,7 +387,6 @@ var EventHub = class { this._stream.end(); this._stream = null; } - this._connect_called = false; //we can turn off since we closed out all callbacks } /* @@ -477,7 +476,8 @@ var EventHub = class { } } - if(force_reconnect && this._connect_called) { + //reconnect will only happen when there is error callback + if(force_reconnect) { try { if(this._stream) { var is_paused = this._stream.isPaused(); @@ -503,7 +503,6 @@ var EventHub = class { logger.error('_checkConnection - error ::' + error.stack ? error.stack : error); var err = new Error('Event hub is not connected '); this._disconnect(err); - throw err; } } } diff --git a/test/integration/e2e/e2eUtils.js b/test/integration/e2e/e2eUtils.js index 6219bdeca8..a2ad3d5ee1 100644 --- a/test/integration/e2e/e2eUtils.js +++ b/test/integration/e2e/e2eUtils.js @@ -377,6 +377,10 @@ function instantiateChaincode(userOrg, chaincode_path, version, language, upgrad t.pass('The chaincode ' + type + ' transaction was valid.'); resolve(); } + }, (err) => { + t.fail('The was a problem with the instantiate event '+err); + clearTimeout(handle); + eh.unregisterTxEvent(deployId); }); }); logger.debug('register eventhub %s with tx=%s',eh.getPeerAddr(),deployId); @@ -571,7 +575,8 @@ function invokeChaincode(userOrg, version, chaincodeId, t, useStore){ { pem: Buffer.from(data).toString(), 'ssl-target-name-override': ORGS[userOrg].peer1['server-hostname'], - 'grpc.http2.keepalive_time' : 15 + 'grpc.keepalive_timeout_ms' : 3000, // time to respond to the ping, 3 seconds + 'grpc.keepalive_time_ms' : 360000, // time to wait for ping response, 6 minutes } ); eh.connect(); @@ -610,7 +615,7 @@ function invokeChaincode(userOrg, version, chaincodeId, t, useStore){ } } t.comment('*****************************************************************************'); - t.comment('stop and start the peer event hub ---- N O W ----- you have ' + sleep_time + ' millis'); + t.comment('stop and start the peer event hub ---- N O W ----- you have ' + sleep_time + ' millis ' + (new Date()).toString()); t.comment('*****************************************************************************'); return sleep(sleep_time); }).then((nothing) => { diff --git a/test/unit/event-hub.js b/test/unit/event-hub.js index c17a98a8d5..32c132c7b0 100644 --- a/test/unit/event-hub.js +++ b/test/unit/event-hub.js @@ -21,7 +21,11 @@ var _test = require('tape-promise'); var test = _test(tape); var testutil = require('./util.js'); +var User = require('fabric-client/lib/User.js'); +var utils = require('fabric-client/lib/utils.js'); +var test_user = require('./user.js'); +var Client = require('fabric-client'); var EventHub = require('fabric-client/lib/EventHub.js'); var sdkUtils = require('fabric-client/lib/utils.js'); @@ -491,3 +495,352 @@ test('\n\n** Test the add and remove utilty used by the EventHub to add a settin t.end(); }); + +// test actions after connect fails +// 1. register for event with no delay and no error callback +// 2. register for event with no delay and error callback +// 3. register for event with delay and no error callback +// 4. register for event with delay and error callback +test('\n\n** EventHub test actions when connect failures on transaction registration \n\n', (t) => { + var client = new Client(); + var event_hub = null; + var member = new User('user1'); + var crypto_suite = utils.newCryptoSuite(); + crypto_suite.setCryptoKeyStore(utils.newCryptoKeyStore()); + member.setCryptoSuite(crypto_suite); + crypto_suite.generateKey() + .then(function (key) { + return member.setEnrollment(key, test_user.TEST_CERT_PEM, 'DEFAULT'); + }).then(() => { + var id = member.getIdentity(); + client.setUserContext(member, true); + + // tx test 1 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9999'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerTxEvent('123', (tx_id, code) => { + t.fail('Failed callback should not have been called - tx test 1') + }); + }, + null, + 'Check for The event hub has not been connected to the event source - tx test 1' + ); + + // tx test 2 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9999'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerTxEvent('123', + (tx_id, code) => { + t.fail('Failed callback should not have been called - tx test 2') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back tx test 2 ::'+error); + } else { + t.failed('Failed to get connection failed error tx test 2 ::'+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - tx test 2' + ); + + // tx test 3 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9999'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.throws( + () => { + event_hub.registerTxEvent('123', (tx_id, code) => { + t.fail('Failed callback should not have been called - tx test 3') + }); + }, + /The event hub has not been connected to the event source/, + 'Check for The event hub has not been connected to the event source - tx test 3' + ); + + // test 4 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9999'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.doesNotThrow( + () => { + event_hub.registerTxEvent('123', + (tx_id, code) => { + t.fail('Failed callback should not have been called - tx test 4') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back tx test 4 ::'+error); + } else { + t.failed('Failed to get connection failed error tx test 4 :: '+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - tx test 4' + ); + + t.end(); + }).catch((err) => { + t.fail(err.stack ? err.stack : err); + t.end(); + }); + +}); + +// test actions after connect fails +// 1. register for event with no delay and no error callback +// 2. register for event with no delay and error callback +// 3. register for event with delay and no error callback +// 4. register for event with delay and error callback +test('\n\n** EventHub test actions when connect failures on block registration \n\n', (t) => { + var client = new Client(); + var event_hub = null; + var member = new User('user1'); + var crypto_suite = utils.newCryptoSuite(); + crypto_suite.setCryptoKeyStore(utils.newCryptoKeyStore()); + member.setCryptoSuite(crypto_suite); + crypto_suite.generateKey() + .then(function (key) { + return member.setEnrollment(key, test_user.TEST_CERT_PEM, 'DEFAULT'); + }).then(() => { + var id = member.getIdentity(); + client.setUserContext(member, true); + + // test 1 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9997'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerBlockEvent((tx_id, code) => { + t.fail('Failed callback should not have been called - block test 1') + }); + }, + null, + 'Check for The event hub has not been connected to the event source - block test 1' + ); + + // block test 2 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9997'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerBlockEvent( + (tx_id, code) => { + t.fail('Failed callback should not have been called - block test 2') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back block test 2 ::'+error); + } else { + t.failed('Failed to get connection failed error block test 2 ::'+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - block test 2' + ); + + // block test 3 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9997'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.throws( + () => { + event_hub.registerBlockEvent((tx_id, code) => { + t.fail('Failed callback should not have been called - block test 3') + }); + }, + /The event hub has not been connected to the event source/, + 'Check for The event hub has not been connected to the event source - block test 3' + ); + + // block test 4 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9997'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.doesNotThrow( + () => { + event_hub.registerBlockEvent( + (tx_id, code) => { + t.fail('Failed callback should not have been called - block test 4') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back block test 4 ::'+error); + } else { + t.failed('Failed to get connection failed error block test 4 :: '+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - block test 4' + ); + + t.end(); + }).catch((err) => { + t.fail(err.stack ? err.stack : err); + t.end(); + }); + +}); + +// chaincode test actions after connect fails +// 1. register for event with no delay and no error callback +// 2. register for event with no delay and error callback +// 3. register for event with delay and no error callback +// 4. register for event with delay and error callback +test('\n\n** EventHub test actions when connect failures on chaincode registration \n\n', (t) => { + var client = new Client(); + var event_hub = null; + var member = new User('user1'); + var crypto_suite = utils.newCryptoSuite(); + crypto_suite.setCryptoKeyStore(utils.newCryptoKeyStore()); + member.setCryptoSuite(crypto_suite); + crypto_suite.generateKey() + .then(function (key) { + return member.setEnrollment(key, test_user.TEST_CERT_PEM, 'DEFAULT'); + }).then(() => { + var id = member.getIdentity(); + client.setUserContext(member, true); + + // chaincode test 1 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9998'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerChaincodeEvent('123', 'event', (tx_id, code) => { + t.fail('Failed callback should not have been called - chaincode test 1') + }); + }, + null, + 'Check for The event hub has not been connected to the event source - chaincode test 1' + ); + + // chaincode test 2 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9998'); + event_hub.connect(); + t.doesNotThrow( + () => { + event_hub.registerChaincodeEvent('123', 'event', + (tx_id, code) => { + t.fail('Failed callback should not have been called - chaincode test 2') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back chaincode test 2 ::'+error); + } else { + t.failed('Failed to get connection failed error chaincode test 2 ::'+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - chaincode test 2' + ); + + // chaincode test 3 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9998'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.throws( + () => { + event_hub.registerChaincodeEvent('123', 'event', (tx_id, code) => { + t.fail('Failed callback should not have been called - chaincode test 3') + }); + }, + /The event hub has not been connected to the event source/, + 'Check for The event hub has not been connected to the event source - chaincode test 3' + ); + + // chaincode test 4 + event_hub = client.newEventHub(); + event_hub.setPeerAddr('grpc://localhost:9998'); + event_hub.connect(); + + let sleep_time = 3000; + t.comment('about to sleep '+sleep_time); + return sleep(sleep_time); + }).then((nothing) => { + t.pass('Sleep complete'); + // eventhub is now actually not connected + + t.doesNotThrow( + () => { + event_hub.registerChaincodeEvent('123', 'event', + (tx_id, code) => { + t.fail('Failed callback should not have been called - chaincode test 4') + }, + (error) =>{ + if(error.toString().indexOf('Connect Failed')) { + t.pass('Successfully got the error call back chaincode test 4 ::'+error); + } else { + t.failed('Failed to get connection failed error chaincode test 4 :: '+error); + } + }); + }, + null, + 'Check for The event hub has not been connected to the event source - chaincode test 4' + ); + + t.end(); + }).catch((err) => { + t.fail(err.stack ? err.stack : err); + t.end(); + }); + +}); + +function sleep(ms) { + return new Promise(resolve => setTimeout(resolve, ms)); +}