Fix close() returns ResultAlreadyClosed after unsubscribe

Fixes #88

### Motivation

When `close` is called if the consumer has already called `unsubscribe`
or `close`, it should not fail. See

https://github.com/apache/pulsar/blob/428c18c8d0c3d135189920740192982e11ffb2bf/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1034

### Modifications

Use the same close logic with Java client.

Add `testCloseAgainBeforeCloseDone` and `testCloseAfterUnsubscribe` to
verify the new behaviors of `Consumer::close`.

Author: BewareMyPower

lib/ConsumerImpl.cc

@@ -1250,8 +1250,10 @@ void ConsumerImpl::closeAsync(ResultCallback originalCallback) {
         }
     };
 
-    if (state_ != Ready) {
-        callback(ResultAlreadyClosed);
+    auto state = state_.load();
+    if (state == Closing || state == Closed) {
+        shutdown();
+        callback(ResultOk);
         return;
     }
 

tests/ConsumerTest.cc

@@ -1401,4 +1401,29 @@ TEST(ConsumerTest, testNoListenerThreadBlocking) {
     client.close();
 }
 
+TEST(ConsumerTest, testCloseAfterUnsubscribe) {
+    Client client{lookupUrl};
+    Consumer consumer;
+    ASSERT_EQ(ResultOk, client.subscribe("test-close-after-unsubscribe", "sub", consumer));
+    ASSERT_EQ(ResultOk, consumer.unsubscribe());
+    ASSERT_EQ(ResultOk, consumer.close());
+}
+
+TEST(ConsumerTest, testCloseAgainBeforeCloseDone) {
+    Client client{lookupUrl};
+    Consumer consumer;
+    ASSERT_EQ(ResultOk, client.subscribe("test-close-again-before-close-done", "sub", consumer));
+    auto done = std::make_shared<std::atomic_bool>(false);
+    auto result = std::make_shared<std::atomic<Result>>(ResultOk);
+    consumer.closeAsync([done, result](Result innerResult) {
+        result->store(innerResult);
+        done->store(true);
+    });
+    ASSERT_EQ(ResultOk, consumer.close());
+    ASSERT_FALSE(*done);
+    waitUntil(std::chrono::seconds(3), [done] { return done->load(); });
+    ASSERT_EQ(ResultOk, *result);
+    ASSERT_TRUE(*done);
+}
+
 }  // namespace pulsar