Fix possible zombie consumer when closing after reconnection

[BewareMyPower]

Fixes #483

Motivation

When the consumer is during reconnection (e.g. after seek is done), the Subscribe request might be sent to the broker but the connection is not bound with the consumer. In this case, closeAsync won't send a CloseConsumer RPC because connection is not available, see

pulsar-client-cpp/lib/ConsumerImpl.cc

Line 1355 in 3be5267

return;

Modifications

Call setCnx in connectionOpened before sending the Subscribe RPC. However, after that, seek or hasMessageAvailable might fail with ResultNotConnected after reconnection. Hence, add a subscribe promise to delay these RPCs after subscribe RPC is done.

Add testSeekAfterSeek and testCloseAfterSeek to verify the behavior.

[BewareMyPower]

Mark it as drafted first, I think a better solution for seek after seek case is to add the backoff, which could be much simpler. I will improve this PR next week.