Merge pull request #156 from jwulf/0.23.0-alpha.5

Fixes #151

Author: s3than

package-lock.json

@@ -1,6 +1,6 @@
 {
 	"name": "zeebe-node",
-	"version": "v0.23.0-alpha.4",
+	"version": "v0.23.0-alpha.5",
 	"description": "A Node.js client library for the Zeebe Microservices Orchestration Engine.",
 	"keywords": [
 		"zeebe",

package.json

@@ -6,10 +6,10 @@ jest.setTimeout(15000)
 describe('StdOut Substitution', () => {
 	it('uses an injected stdout', done => {
 		const mockStd = new MockStdOut()
-		const z = new ZBClient({ stdout: mockStd })
+		const z = new ZBClient({ stdout: mockStd, eagerConnection: false })
 
 		// tslint:disable-next-line: no-console
-		z.createWorker(null, 'test', console.log)
+		z.createWorker('test', console.log)
 		setTimeout(() => {
 			z.close()
 		}, 2000)

src/__tests__/StdOut.spec.ts

@@ -440,14 +440,15 @@ export class GrpcClient extends EventEmitter {
 		return metadata
 	}
 
-	private watchGrpcChannel(): Promise<number> {
+	private waitForGrpcChannelReconnect(): Promise<number> {
 		this.emit(MiddlewareSignals.Log.Debug, 'Start watching Grpc channel...')
 		return new Promise(resolve => {
+			const tryToConnect = true
 			const gRPC = this.client
 			if (this.channelClosed) {
 				return
 			}
-			const state = gRPC.getChannel().getConnectivityState(false)
+			const state = gRPC.getChannel().getConnectivityState(tryToConnect)
 			this.emit(
 				MiddlewareSignals.Log.Error,
 				`Grpc Channel State: ${connectivityState[state]}`
@@ -473,7 +474,7 @@ export class GrpcClient extends EventEmitter {
 					}
 					const newState = gRPC
 						.getChannel()
-						.getConnectivityState(false)
+						.getConnectivityState(tryToConnect)
 					this.emit(
 						MiddlewareSignals.Log.Error,
 						`Grpc Channel State: ${connectivityState[newState]}`
@@ -484,16 +485,15 @@ export class GrpcClient extends EventEmitter {
 					)
 					if (
 						newState === GrpcState.READY ||
-						newState === GrpcState.IDLE ||
-						newState === GrpcState.TRANSIENT_FAILURE
+						newState === GrpcState.IDLE
 					) {
 						return resolve(newState)
 					} else {
 						this.emit(
 							MiddlewareSignals.Log.Error,
 							`Grpc Retry count: ${this.gRPCRetryCount}`
 						)
-						return resolve(await this.watchGrpcChannel())
+						return resolve(await this.waitForGrpcChannelReconnect())
 					}
 				}
 			)
@@ -562,17 +562,18 @@ export class GrpcClient extends EventEmitter {
 	private handleGrpcError = (stream: any) => async (err: any) => {
 		this.emit(MiddlewareSignals.Event.Error)
 		this.emit(MiddlewareSignals.Log.Error, `Grpc Error: ${err.message}`)
-		const channelState = await this.watchGrpcChannel()
+		const channelState = await this.waitForGrpcChannelReconnect()
 		this.emit(
 			MiddlewareSignals.Log.Debug,
 			`Grpc Channel state: ${connectivityState[channelState]}`
 		)
 		stream.removeAllListeners()
+		this.emit(MiddlewareSignals.Event.Ready)
 		// if (
 		// 	channelState === GrpcState.READY ||
 		// 	channelState === GrpcState.IDLE
 		// ) {
-		// 	this.emit(MiddlewareSignals.Log.Info, 'Grpc channel connected')
+		// 	this.emit(MiddlewareSignals.Log.Info, 'Grpc channel reconnected')
 		// 	// tslint:disable-next-line: no-console
 		// 	console.log('handleGrpc', channelState) // @DEBUG
 

src/lib/GrpcClient.ts

@@ -203,8 +203,8 @@ export class ZBWorkerBase<
 				await this.grpcClient.close(timeout)
 				this.grpcClient.removeAllListeners()
 				Object.keys(this.jobStreams).forEach(key => {
-					this.jobStreams[key].removeAllListeners()
-					this.jobStreams[key].cancel()
+					this.jobStreams[key]?.removeAllListeners?.()
+					this.jobStreams[key]?.cancel?.()
 					delete this.jobStreams[key]
 					this.logger.logDebug('Removed Job Stream Listeners')
 				})
@@ -399,7 +399,7 @@ export class ZBWorkerBase<
 			this.jobStreams[id] = jobStream.stream
 			// This event happens when the server cancels the call after the deadline
 			// And when it has completed a response with work
-			jobStream.stream.on('end', () => {
+			jobStream.stream.on?.('end', () => {
 				this.logger.logDebug(
 					`Stream ended after ${(Date.now() - start) / 1000} seconds`
 				)
@@ -478,7 +478,7 @@ export class ZBWorkerBase<
 			}
 		}
 
-		stream.on('data', (res: ActivateJobsResponse) => {
+		stream?.on?.('data', (res: ActivateJobsResponse) => {
 			// If we are closing, don't start working on these jobs. They will have to be timed out by the server.
 			if (this.closing) {
 				return