Outgoing cleanup (#1838)

Author: bernardnormier

cpp/include/Ice/Communicator.h

@@ -317,10 +317,7 @@ class ICE_CLASS(ICE_API) Communicator
      * @param compress Specifies whether or not the queued batch requests should be compressed before being sent over
      * the wire.
      */
-    virtual void flushBatchRequests(CompressBatch compress)
-    {
-        flushBatchRequestsAsync(compress).get();
-    }
+    ICE_MEMBER(ICE_API) void flushBatchRequests(CompressBatch compress);
 
     /**
      * Flush any pending batch requests for this communicator. This means all batch requests invoked on fixed proxies
@@ -345,23 +342,7 @@ class ICE_CLASS(ICE_API) Communicator
      * the wire.
      * @return The future object for the invocation.
      */
-    template<template<typename> class P = ::std::promise>
-    auto flushBatchRequestsAsync(CompressBatch compress)
-        -> decltype(::std::declval<P<void>>().get_future())
-    {
-        using Promise = P<void>;
-        auto promise = ::std::make_shared<Promise>();
-        flushBatchRequestsAsync(compress,
-                                [promise](::std::exception_ptr ex)
-                                {
-                                    promise->set_exception(::std::move(ex));
-                                },
-                                [promise](bool)
-                                {
-                                    promise->set_value();
-                                });
-        return promise->get_future();
-    }
+    ICE_MEMBER(ICE_API) std::future<void> flushBatchRequestsAsync(CompressBatch compress);
 
     /**
      * Add the Admin object with all its facets to the provided object adapter. If <code>Ice.Admin.ServerId</code> is

cpp/include/Ice/Connection.h

@@ -12,7 +12,6 @@
 #include <Ice/Exception.h>
 #include <Ice/StreamHelpers.h>
 #include <Ice/Comparable.h>
-#include <Ice/OutgoingAsync.h>
 #include <optional>
 #include <Ice/ObjectAdapterF.h>
 #include <Ice/Identity.h>
@@ -304,10 +303,7 @@ class ICE_CLASS(ICE_API) Connection
      * @param compress Specifies whether or not the queued batch requests should be compressed before being sent over
      * the wire.
      */
-    virtual void flushBatchRequests(CompressBatch compress)
-    {
-        flushBatchRequestsAsync(compress).get();
-    }
+    ICE_MEMBER(ICE_API) void flushBatchRequests(CompressBatch compress);
 
     /**
      * Flush any pending batch requests for this connection. This means all batch requests invoked on fixed proxies
@@ -330,23 +326,7 @@ class ICE_CLASS(ICE_API) Connection
      * the wire.
      * @return The future object for the invocation.
      */
-    template<template<typename> class P = ::std::promise>
-    auto flushBatchRequestsAsync(CompressBatch compress)
-        -> decltype(::std::declval<P<void>>().get_future())
-    {
-        using Promise = P<void>;
-        auto promise = ::std::make_shared<Promise>();
-        flushBatchRequestsAsync(compress,
-                                [promise](::std::exception_ptr ex)
-                                {
-                                    promise->set_exception(::std::move(ex));
-                                },
-                                [promise](bool)
-                                {
-                                    promise->set_value();
-                                });
-        return promise->get_future();
-    }
+    ICE_MEMBER(ICE_API) std::future<void> flushBatchRequestsAsync(CompressBatch compress);
 
     /**
      * Set a close callback on the connection. The callback is called by the connection when it's closed. The callback
@@ -366,10 +346,7 @@ class ICE_CLASS(ICE_API) Connection
     /**
      * Send a heartbeat message.
      */
-    virtual void heartbeat()
-    {
-        heartbeatAsync().get();
-    }
+    ICE_MEMBER(ICE_API) void heartbeat();
 
     /**
      * Send a heartbeat message.
@@ -385,28 +362,13 @@ class ICE_CLASS(ICE_API) Connection
      * Send a heartbeat message.
      * @return The future object for the invocation.
      */
-    template<template<typename> class P = ::std::promise>
-    auto heartbeatAsync()
-        -> decltype(::std::declval<P<void>>().get_future())
-    {
-        using Promise = P<void>;
-        auto promise = ::std::make_shared<Promise>();
-        heartbeatAsync([promise](::std::exception_ptr ex)
-                       {
-                           promise->set_exception(::std::move(ex));
-                       },
-                       [promise](bool)
-                       {
-                           promise->set_value();
-                       });
-        return promise->get_future();
-    }
+    ICE_MEMBER(ICE_API) std::future<void> heartbeatAsync();
 
     /**
      * Set the active connection management parameters.
      * @param timeout The timeout value in seconds, must be &gt;= 0.
      * @param close The close condition
-     * @param heartbeat The hertbeat condition
+     * @param heartbeat The heartbeat condition
      */
     virtual void setACM(const std::optional<int>& timeout, const std::optional<ACMClose>& close, const std::optional<ACMHeartbeat>& heartbeat) = 0;
 
@@ -481,7 +443,7 @@ class ICE_CLASS(ICE_API) IPConnectionInfo : public ::Ice::ConnectionInfo
 
     /**
      * One-shot constructor to initialize all data members.
-     * @param underlying The information of the underyling transport or null if there's no underlying transport.
+     * @param underlying The information of the underlying transport or null if there's no underlying transport.
      * @param incoming Whether or not the connection is an incoming or outgoing connection.
      * @param adapterName The name of the adapter associated with the connection.
      * @param connectionId The connection id.
@@ -540,7 +502,7 @@ class ICE_CLASS(ICE_API) TCPConnectionInfo : public ::Ice::IPConnectionInfo
 
     /**
      * One-shot constructor to initialize all data members.
-     * @param underlying The information of the underyling transport or null if there's no underlying transport.
+     * @param underlying The information of the underlying transport or null if there's no underlying transport.
      * @param incoming Whether or not the connection is an incoming or outgoing connection.
      * @param adapterName The name of the adapter associated with the connection.
      * @param connectionId The connection id.

cpp/include/Ice/Ice.h

@@ -9,14 +9,12 @@
 #include <Ice/Config.h>
 #include <Ice/Comparable.h>
 #include <Ice/StreamHelpers.h>
-#include <Ice/OutgoingAsync.h>
 #include <Ice/Proxy.h>
 #include <Ice/Current.h>
 #include <Ice/LocalException.h>
 #include <optional>
 #include <Ice/Object.h>
 #include <Ice/SlicedData.h>
-#include <Ice/OutgoingAsync.h>
 #include <Ice/IncomingAsync.h>
 #include <Ice/FactoryTable.h>
 #include <Ice/FactoryTableInit.h>

cpp/include/Ice/OutgoingAsync.h

@@ -6,7 +6,6 @@
 #define ICE_OUTGOING_ASYNC_H
 
 #include <IceUtil/Timer.h>
-#include <Ice/OutgoingAsyncF.h>
 #include <Ice/CommunicatorF.h>
 #include <Ice/ConnectionIF.h>
 #include <Ice/ObjectAdapterF.h>
@@ -23,6 +22,7 @@
 namespace IceInternal
 {
 
+class OutgoingAsyncBase;
 class RetryException;
 class CollocatedRequestHandler;
 
@@ -136,6 +136,8 @@ class ICE_API OutgoingAsyncBase : public virtual OutgoingAsyncCompletionCallback
     static const unsigned char Sent;
 };
 
+using OutgoingAsyncBasePtr = ::std::shared_ptr<OutgoingAsyncBase>;
+
 //
 // Base class for proxy based invocations. This class handles the
 // retry for proxy invocations. It also ensures the child observer is
@@ -164,7 +166,7 @@ class ICE_API ProxyOutgoingAsyncBase : public OutgoingAsyncBase,
 
 protected:
 
-    ProxyOutgoingAsyncBase(const Ice::ObjectPrx&);
+    ProxyOutgoingAsyncBase(Ice::ObjectPrx);
     ~ProxyOutgoingAsyncBase();
 
     void invokeImpl(bool);
@@ -184,14 +186,16 @@ class ICE_API ProxyOutgoingAsyncBase : public OutgoingAsyncBase,
     bool _sent;
 };
 
+using ProxyOutgoingAsyncBasePtr = ::std::shared_ptr<ProxyOutgoingAsyncBase>;
+
 //
 // Class for handling Slice operation invocations
 //
 class ICE_API OutgoingAsync : public ProxyOutgoingAsyncBase
 {
 public:
 
-    OutgoingAsync(const Ice::ObjectPrx&, bool);
+    OutgoingAsync(Ice::ObjectPrx, bool);
 
     void prepare(const std::string&, Ice::OperationMode, const Ice::Context&);
 
@@ -220,7 +224,7 @@ class ICE_API OutgoingAsync : public ProxyOutgoingAsyncBase
     {
         _os.writeEmptyEncapsulation(_encoding);
     }
-    void writeParamEncaps(const ::Ice::Byte* encaps, ::std::int32_t size)
+    void writeParamEncaps(const ::Ice::Byte* encaps, std::int32_t size)
     {
         if(size == 0)
         {
@@ -239,16 +243,13 @@ class ICE_API OutgoingAsync : public ProxyOutgoingAsyncBase
     bool _synchronous;
 };
 
-}
-
-namespace IceInternal
-{
+using OutgoingAsyncPtr = ::std::shared_ptr<OutgoingAsync>;
 
 class ICE_API LambdaInvoke : public virtual OutgoingAsyncCompletionCallback
 {
 public:
 
-    LambdaInvoke(std::function<void(::std::exception_ptr)> exception, std::function<void(bool)> sent) :
+    LambdaInvoke(std::function<void(std::exception_ptr)> exception, std::function<void(bool)> sent) :
         _exception(std::move(exception)), _sent(std::move(sent))
     {
     }
@@ -263,25 +264,21 @@ class ICE_API LambdaInvoke : public virtual OutgoingAsyncCompletionCallback
     virtual void handleInvokeException(std::exception_ptr, OutgoingAsyncBase*) const override;
     virtual void handleInvokeResponse(bool, OutgoingAsyncBase*) const override;
 
-    std::function<void(::std::exception_ptr)> _exception;
+    std::function<void(std::exception_ptr)> _exception;
     std::function<void(bool)> _sent;
     std::function<void(bool)> _response;
 };
 
-template<typename Promise>
+template<typename R>
 class PromiseInvoke : public virtual OutgoingAsyncCompletionCallback
 {
 public:
 
-    auto
-    getFuture() -> decltype(std::declval<Promise>().get_future())
-    {
-        return _promise.get_future();
-    }
+    std::future<R> getFuture() { return _promise.get_future(); }
 
 protected:
 
-    Promise _promise;
+    std::promise<R> _promise;
     std::function<void(bool)> _response;
 
 private:
@@ -388,11 +385,11 @@ class LambdaOutgoing : public OutgoingAsyncT<R>, public LambdaInvoke
 {
 public:
 
-    LambdaOutgoing(const Ice::ObjectPrx& proxy,
+    LambdaOutgoing(Ice::ObjectPrx proxy,
                    std::function<void(R)> response,
-                   std::function<void(::std::exception_ptr)> ex,
+                   std::function<void(std::exception_ptr)> ex,
                    std::function<void(bool)> sent) :
-        OutgoingAsyncT<R>(proxy, false), LambdaInvoke(std::move(ex), std::move(sent))
+        OutgoingAsyncT<R>(std::move(proxy), false), LambdaInvoke(std::move(ex), std::move(sent))
     {
         _response = [this, response = std::move(response)](bool ok)
         {
@@ -424,11 +421,11 @@ class LambdaOutgoing<void> : public OutgoingAsyncT<void>, public LambdaInvoke
 {
 public:
 
-    LambdaOutgoing(const Ice::ObjectPrx& proxy,
+    LambdaOutgoing(Ice::ObjectPrx proxy,
                    std::function<void()> response,
-                   std::function<void(::std::exception_ptr)> ex,
+                   std::function<void(std::exception_ptr)> ex,
                    std::function<void(bool)> sent) :
-        OutgoingAsyncT<void>(proxy, false), LambdaInvoke(std::move(ex), std::move(sent))
+        OutgoingAsyncT<void>(std::move(proxy), false), LambdaInvoke(std::move(ex), std::move(sent))
     {
         _response = [this, response = std::move(response)](bool ok)
         {
@@ -460,11 +457,11 @@ class CustomLambdaOutgoing : public OutgoingAsync, public LambdaInvoke
 {
 public:
 
-    CustomLambdaOutgoing(const Ice::ObjectPrx& proxy,
+    CustomLambdaOutgoing(Ice::ObjectPrx proxy,
                          std::function<void(Ice::InputStream*)> read,
-                         std::function<void(::std::exception_ptr)> ex,
+                         std::function<void(std::exception_ptr)> ex,
                          std::function<void(bool)> sent) :
-        OutgoingAsync(proxy, false), LambdaInvoke(std::move(ex), std::move(sent))
+        OutgoingAsync(std::move(proxy), false), LambdaInvoke(std::move(ex), std::move(sent))
     {
         _response = [this, read = std::move(read)](bool ok)
         {
@@ -495,13 +492,13 @@ class CustomLambdaOutgoing : public OutgoingAsync, public LambdaInvoke
     }
 };
 
-template<typename P, typename R>
-class PromiseOutgoing : public OutgoingAsyncT<R>, public PromiseInvoke<P>
+template<typename R>
+class PromiseOutgoing : public OutgoingAsyncT<R>, public PromiseInvoke<R>
 {
 public:
 
-    PromiseOutgoing(const Ice::ObjectPrx& proxy, bool sync) :
-        OutgoingAsyncT<R>(proxy, sync)
+    PromiseOutgoing(Ice::ObjectPrx proxy, bool sync) :
+        OutgoingAsyncT<R>(std::move(proxy), sync)
     {
         this->_response = [this](bool ok)
         {
@@ -521,13 +518,13 @@ class PromiseOutgoing : public OutgoingAsyncT<R>, public PromiseInvoke<P>
     }
 };
 
-template<typename P>
-class PromiseOutgoing<P, void> : public OutgoingAsyncT<void>, public PromiseInvoke<P>
+template<>
+class PromiseOutgoing<void> : public OutgoingAsyncT<void>, public PromiseInvoke<void>
 {
 public:
 
-    PromiseOutgoing(const Ice::ObjectPrx& proxy, bool sync) :
-        OutgoingAsyncT<void>(proxy, sync)
+    PromiseOutgoing(Ice::ObjectPrx proxy, bool sync) :
+        OutgoingAsyncT<void>(std::move(proxy), sync)
     {
         this->_response = [&](bool ok)
         {
@@ -556,12 +553,28 @@ class PromiseOutgoing<P, void> : public OutgoingAsyncT<void>, public PromiseInvo
     {
         if(done)
         {
-            PromiseInvoke<P>::_promise.set_value();
+            PromiseInvoke<void>::_promise.set_value();
         }
         return false;
     }
 };
 
+template<typename R, typename Obj, typename Fn, typename... Args>
+inline std::future<R> makePromiseOutgoing(bool sync, Obj obj, Fn fn, Args&&... args)
+{
+    auto outAsync = std::make_shared<PromiseOutgoing<R>>(*obj, sync);
+    (obj->*fn)(outAsync, std::forward<Args>(args)...);
+    return outAsync->getFuture();
+}
+
+template<typename R, typename Re, typename E, typename S, typename Obj, typename Fn, typename... Args>
+inline std::function<void()> makeLambdaOutgoing(Re r, E e, S s, Obj obj, Fn fn, Args&&... args)
+{
+    auto outAsync = std::make_shared<LambdaOutgoing<R>>(*obj, std::move(r), std::move(e), std::move(s));
+    (obj->*fn)(outAsync, std::forward<Args>(args)...);
+    return [outAsync]() { outAsync->cancel(); };
+}
+
 }
 
 #endif