1114 reliabletopic terminate test failure [AP-1893] (#1122)

* ReliableTopic lifecycle is checked in `cancel()`

Author: akeles85

hazelcast/include/hazelcast/client/reliable_topic.h

@@ -49,7 +49,9 @@ namespace client {
  * process m1, m2, m3...mn in order.
  *
  */
-class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
+class HAZELCAST_API reliable_topic
+  : public proxy::ProxyImpl
+  , public std::enable_shared_from_this<reliable_topic>
 {
     friend class spi::ProxyManager;
     friend class hazelcast_client;
@@ -106,7 +108,7 @@ class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
                                       logger_,
                                       execution_service_,
                                       executor_,
-                                      runners_map_));
+                                      shared_from_this()));
         runners_map_.put(id, runner);
         runner->next();
         return std::to_string(id);
@@ -149,10 +151,9 @@ class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
                       int batch_size,
                       logger& lg,
                       std::shared_ptr<spi::impl::ClientExecutionServiceImpl>
-                        execution_service,                      
+                        execution_service,
                       util::hz_thread_pool& executor,
-                      util::SynchronizedMap<int, util::concurrent::Cancellable>&
-                        runners_map)
+                      std::weak_ptr<reliable_topic> topic)
           : listener_(listener)
           , id_(id)
           , ringbuffer_(rb)
@@ -163,7 +164,7 @@ class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
           , executor_(executor)
           , serialization_service_(service)
           , batch_size_(batch_size)
-          , runners_map_(runners_map)
+          , topic_(std::move(topic))
         {
             // we are going to listen to next publication. We don't care about
             // what already has been published.
@@ -334,7 +335,10 @@ class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
         bool cancel() override
         {
             cancelled_.store(true);
-            runners_map_.remove(id_);
+            auto topic_ptr = topic_.lock();
+            if (topic_ptr) {
+                topic_ptr->runners_map_.remove(id_);
+            }
             return true;
         }
 
@@ -414,7 +418,7 @@ class HAZELCAST_API reliable_topic : public proxy::ProxyImpl
         util::hz_thread_pool& executor_;
         serialization::pimpl::SerializationService& serialization_service_;
         int batch_size_;
-        util::SynchronizedMap<int, util::concurrent::Cancellable>& runners_map_;
+        std::weak_ptr<reliable_topic> topic_;
     };
 
     util::SynchronizedMap<int, util::concurrent::Cancellable> runners_map_;