How to handle microservice exit on client side?

[cncolder]

I'm submitting a...

[ ] Regression 
[ ] Bug report
[x] Feature request
[ ] Documentation issue or request
[ ] Support request => Please do not submit support request here, instead post your question on Stack Overflow.

Current behavior

After microservice exit (reboot or upgrade). The Observable return from client.send won't receive next notification again. The error and complete not be called.

Expected behavior

Client need receive an ended notify. Then retry or return error to user.

Minimal reproduction of the problem with instructions

// microservice side
@Controller('timer')
export class TimerController {
    @MessagePattern({ cmd: 'interval' })
    interval() {
        return rxjs.interval(1000)
    }
}

// client side
@Injectable()
export class TimerService {
    constructor() {
        process.nextTick(() => this.watch())
    }

    @Client({
        transport: Transport.REDIS,
        options: { url: `redis://${REDIS_HOST}:${REDIS_PORT}` },
    })
    private client: ClientProxy

    watch() {
        this.client.send({ cmd: 'interval' }).subscribe(
            n => console.log(n), // stop after reboot microservice
            err => console.error('error', err), // no run
            () => console.log('complete'), // no run
        )
    }
}

Output:

1
2
3
# reboot microservice

What is the motivation / use case for changing the behavior?

Image a more complex example: Microservice watch mongoose model change stream. return rxjs.fromEvent(this.User.watch(), 'change') GraphQL receive changes and publish to PubSub

Environment

Nest version: 5.4.1


For Tooling issues:
- Node version: XX  
- Platform:  

Others:

[cncolder]

I have resolve this in a dirty way.

In microservice. I return a merged observable.

    watch (resumeAfter) {
        // User is a mongoose model
        const changeStream = this.User.watch({ fullDocument: 'updateLookup', resumeAfter })
        // Send keep alive ping to client every 1s
        const ka$ = interval(1000).pipe(map(() => ({ eventName: 'ka' })))
        const change$ = fromEvent(changeStream, 'change').pipe(
            map(data => ({ eventName: 'change', ...data })),
        )
        // Change stream will close if MongoDB server reboot
        const close$ = fromEvent(changeStream, 'close').pipe(
            map(data => ({ eventName: 'close', ...data })),
        )
        return merge(ka$, change$, close$)
    }

In gateway. We can throw timeout if keep alive is missing.

    private resumeAfter: MongoChangeEvent['_id']
    watch() {
        let res$ = this.client.send<MongoChangeEvent>(
            { srv: 'user', cmd: 'watch' },
            { resumeAfter: this.resumeAfter },
        )
        res$.pipe(timeout(1100))
            .pipe(filter(event => event.eventName !== 'ka'))
            .subscribe(
                event => {
                    if (event.eventName === 'close') {
                        return this.watch()
                    }
                    this.resumeAfter = event._id
                    // Your logical code to handle changed document
                },
                error => this.watch()
            )
    }

Client will throw TimeoutError if no ka coming again. It handle close event also. And re-watch again.

[kamilmysliwiec]

Actually (to be honest), I think that your solution is good in this case. However, instead of recursively calling this.watch(), I'd rather suggest applying retry operator that would try to automatically reconnect after some timeout.

[cncolder]

Ye. retry is better choice.

But I think we need timeout in all scenes that microservice return Observable. Every time microservice exit without error. The client will hung up.

Maybe interval is not a good example.

If microservice return of(1,2,3) (like mongodb cursor). The service crash after return 1 and 2. Client will never complete. Because there isn't a timeout option for microservice communication layer.

[kamilmysliwiec]

It seems that merging a typical ping endpoint should be enough to handle such a thing. https://github.com/nestjs/nest/issues/1414#issuecomment-451449323

[lock[bot]]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.