Fix segfault caused by socket I/O on a closed io_service

Fixes apache#167

### Motivation

Here are some debugging info when the segfault happened in
`testCloseClient`. The outputs have been trimmed to make them clear.

An example crash at `async_write`:

```
#12 0x00007ffff7496dad in basic_stream_socket<...>::boost::asio::async_write /usr/include/boost/asio/impl/write.hpp:512
#13 0x00007ffff748e003 in ClientConnection::asyncWrite lib/ClientConnection.h:245
#14 0x00007ffff746e0b6 in ClientConnection::handleHandshake (this=0x555555e689d0) lib/ClientConnection.cc:502
```

Another example crash at `async_receive`:

```
#6  0x00007ffff7497247 in basic_stream_socket<...>::async_receive /usr/include/boost/asio/basic_stream_socket.hpp:677
#7  0x00007ffff748e647 in ClientConnection::asyncReceive lib/ClientConnection.h:258
#8  0x00007ffff746fa5d in ClientConnection::readNextCommand lib/ClientConnection.cc:606
```

The frame where it crashed:

```
245       if (descriptor_data->shutdown_)
(gdb) p descriptor_data
$2 = (boost::asio::detail::epoll_reactor::per_descriptor_data &) @0x555555e4a780: 0x0
```

We can see the socket descriptor is `nullptr`. The root cause is when
`async_receive` or `async_write` is called, the `io_service` object
might be closed. This case happened when `createProducerAsync` is
called, the actual producer creation continues in another thread, while
the `client.close()` happens in the current thread.

### Modifications

Check if the `ClientConnection` is closed before `async_receive` or
`async_write`. To avoid the use of lock, changing the `state_` field to
atomic.

### Verifications

```bash
./tests/pulsar-tests --gtest_filter='ClientTest.testCloseClient' --gtest_repeat=20
```

It never crashed after applying this patch.

Author: BewareMyPower

lib/ClientConnection.cc

@@ -812,7 +812,7 @@ void ClientConnection::handleIncomingMessage(const proto::CommandMessage& msg, b
 void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
     LOG_DEBUG(cnxString_ << "Handling incoming command: " << Commands::messageType(incomingCmd.type()));
 
-    switch (state_) {
+    switch (state_.load()) {
         case Pending: {
             LOG_ERROR(cnxString_ << "Connection is not ready yet");
             break;

lib/ClientConnection.h

@@ -22,6 +22,7 @@
 #include <pulsar/ClientConfiguration.h>
 #include <pulsar/defines.h>
 
+#include <atomic>
 #include <boost/any.hpp>
 #include <boost/asio/bind_executor.hpp>
 #include <boost/asio/deadline_timer.hpp>
@@ -235,6 +236,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
 
     template <typename ConstBufferSequence, typename WriteHandler>
     inline void asyncWrite(const ConstBufferSequence& buffers, WriteHandler handler) {
+        if (isClosed()) {
+            return;
+        }
         if (tlsSocket_) {
 #if BOOST_VERSION >= 106600
             boost::asio::async_write(*tlsSocket_, buffers, boost::asio::bind_executor(strand_, handler));
@@ -248,6 +252,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
 
     template <typename MutableBufferSequence, typename ReadHandler>
     inline void asyncReceive(const MutableBufferSequence& buffers, ReadHandler handler) {
+        if (isClosed()) {
+            return;
+        }
         if (tlsSocket_) {
 #if BOOST_VERSION >= 106600
             tlsSocket_->async_read_some(buffers, boost::asio::bind_executor(strand_, handler));
@@ -259,7 +266,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
         }
     }
 
-    State state_ = Pending;
+    std::atomic<State> state_{Pending};
     TimeDuration operationsTimeout_;
     AuthenticationPtr authentication_;
     int serverProtocolVersion_;