Skip to content
Permalink
Browse files

fix(grpc): clean up disconnected streams

  • Loading branch information...
mrfelton committed Apr 24, 2019
1 parent 5cd1a5b commit 8c67eff7b6d1b686db51cb52e056b410ba6eed36
Showing with 18 additions and 7 deletions.
  1. +17 −7 services/grpc/grpcService.js
  2. +1 −0 services/grpc/lightning/lightning.js
@@ -145,32 +145,42 @@ class GrpcService extends EventEmitter {
*/
subscribe() {
// this.subscriptions['something'] = this.service.subscribeToSomething()
// super.subscribe()

// Close and clear subscriptions when they emit an end event.
const activeSubKeys = Object.keys(this.subscriptions)
activeSubKeys.forEach(key => {
const call = this.subscriptions[key]
call.on('end', () => {
grpcLog.info(`gRPC subscription "${this.serviceName}.${key}" ended.`)
delete this.subscriptions[key]
})
})
}

/**
* Unsubscribe from all streams.
*/
async unsubscribe() {
const activeSubKeys = Object.keys(this.subscriptions)
if (activeSubKeys.length) {
grpcLog.info(`Unsubscribing from all ${this.serviceName} gRPC streams: %o`, activeSubKeys)
const cancellations = activeSubKeys.map(key => this._cancelSubscription(key))
return Promise.all(cancellations)
}
grpcLog.info(`Unsubscribing from all ${this.serviceName} gRPC streams: %o`, activeSubKeys)
const cancellations = activeSubKeys.map(key => this._cancelSubscription(key))
await Promise.all(cancellations)
}

/**
* Unsubscribe from a single stream.
*/
async _cancelSubscription(key) {
grpcLog.info(`Unsubscribing from ${key} gRPC stream`)
grpcLog.info(`Unsubscribing from ${this.serviceName}.${key} gRPC stream`)
const call = this.subscriptions[key]

// Cancellation status callback handler.
const result = new Promise(resolve => {
call.on('status', callStatus => {
if (callStatus.code === status.CANCELLED) {
grpcLog.info(`Unsubscribed from ${key} gRPC stream`)
delete this.subscriptions[key]
grpcLog.info(`Unsubscribed from ${this.serviceName}.${key} gRPC stream`)
resolve()
}
})
@@ -61,6 +61,7 @@ class Lightning extends LndGrpcService {
this.subscriptions['channelGraph'] = this.subscribeChannelGraph()
this.subscriptions['invoices'] = this.subscribeInvoices()
this.subscriptions['transactions'] = this.subscribeTransactions()
super.subscribe()
}
}

0 comments on commit 8c67eff

Please sign in to comment.
You can’t perform that action at this time.