AWSClientAsyncCRTP.h

 
#pragma once
 
#include <aws/core/client/AWSAsyncOperationTemplate.h>
#include <aws/core/utils/logging/ErrorMacros.h>
#include <aws/core/utils/component-registry/ComponentRegistry.h>
 
namespace Aws
{
namespace Http { class HttpRequest; }
namespace Utils {
  namespace Stream { class HttpWriteDataStreamBuf; }
  namespace Threading { class Executor; }
}
namespace Client
{
    class AsyncCallerContext;
    template <typename OutcomeT, typename ClientT, typename AWSEndpointT, typename RequestT, typename HandlerT>
    class BidirectionalEventStreamingTask;
 
    template<typename T>
    struct AWS_CORE_LOCAL IsEventStreamOperation : IsEventStreamOperation<decltype(&T::operator())> {};
 
    template<typename ReturnT, typename ClassT, typename RequestT>
    struct AWS_CORE_LOCAL IsEventStreamOperation<ReturnT(ClassT::*)(RequestT) const>
    {
        static const bool value = !std::is_const<typename std::remove_reference<RequestT>::type>::value;
    };
 
    template<typename ReturnT, typename ClassT>
    struct AWS_CORE_LOCAL IsEventStreamOperation<ReturnT(ClassT::*)() const>
    {
        static const bool value = false;
    };
 
 
    template <typename AwsServiceClientT>
    class ClientWithAsyncTemplateMethods
    {
    public:
        ClientWithAsyncTemplateMethods()
         : m_isInitialized(true),
           m_operationsProcessed(0)
        {
            AwsServiceClientT* pThis = static_cast<AwsServiceClientT*>(this);
            Aws::Utils::ComponentRegistry::RegisterComponent(AwsServiceClientT::GetServiceName(),
                                                             pThis,
                                                             &AwsServiceClientT::ShutdownSdkClient);
 
        }
 
        ClientWithAsyncTemplateMethods(const ClientWithAsyncTemplateMethods& other)
         : m_isInitialized(other.m_isInitialized.load()),
           m_operationsProcessed(0)
        {
            AwsServiceClientT* pThis = static_cast<AwsServiceClientT*>(this);
            Aws::Utils::ComponentRegistry::RegisterComponent(AwsServiceClientT::GetServiceName(),
                                                             pThis,
                                                             &AwsServiceClientT::ShutdownSdkClient);
        }
 
        ClientWithAsyncTemplateMethods& operator=(const ClientWithAsyncTemplateMethods& other)
        {
            if (&other != this)
            {
                ShutdownSdkClient(static_cast<AwsServiceClientT*>(this));
                m_operationsProcessed = 0;
                m_isInitialized = other.m_isInitialized.load();
            }
 
            return *this;
        }
 
        virtual ~ClientWithAsyncTemplateMethods()
        {
            AwsServiceClientT* pClient = static_cast<AwsServiceClientT*>(this);
            Aws::Utils::ComponentRegistry::DeRegisterComponent(pClient);
        }
 
        static void ShutdownSdkClient(void* pThis, int64_t timeoutMs = -1)
        {
            AwsServiceClientT* pClient = reinterpret_cast<AwsServiceClientT*>(pThis);
            AWS_CHECK_PTR(AwsServiceClientT::GetServiceName(), pClient);
            if(!pClient->m_isInitialized)
            {
                return;
            }
 
            std::unique_lock<std::mutex> lock(pClient->m_shutdownMutex);
 
            pClient->m_isInitialized = false;
            if (pClient->GetHttpClient().use_count() == 1)
            {
                pClient->DisableRequestProcessing();
            }
 
            if (timeoutMs == -1)
            {
                timeoutMs = pClient->m_clientConfiguration.requestTimeoutMs;
            }
            pClient->m_shutdownSignal.wait_for(lock,
                                      std::chrono::milliseconds(timeoutMs),
                                      [&](){ return pClient->m_operationsProcessed.load() == 0; });
 
            if (pClient->m_operationsProcessed.load())
            {
                AWS_LOGSTREAM_FATAL(AwsServiceClientT::GetAllocationTag(), "Service client "
                    << AwsServiceClientT::GetServiceName() << " is shutting down while async tasks are present.");
            }
 
            pClient->m_clientConfiguration.executor.reset();
            pClient->m_clientConfiguration.retryStrategy.reset();
            pClient->m_endpointProvider.reset();
        }
 
        template<typename RequestT, typename HandlerT, typename OperationFuncT, typename std::enable_if<!IsEventStreamOperation<OperationFuncT>::value, int>::type = 0>
        void SubmitAsync(OperationFuncT operationFunc,
                         const RequestT& request,
                         const HandlerT& handler,
                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr) const
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            Aws::Client::MakeAsyncOperation(operationFunc, clientThis, request, handler, context, clientThis->m_clientConfiguration.executor.get());
        }
 
        template<typename RequestT, typename HandlerT, typename OperationFuncT, typename std::enable_if<IsEventStreamOperation<OperationFuncT>::value, int>::type = 0>
        void SubmitAsync(OperationFuncT operationFunc,
                         RequestT& request, // note non-const ref
                         const HandlerT& handler,
                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr) const
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            Aws::Client::MakeAsyncStreamingOperation(operationFunc, clientThis, request, handler, context, clientThis->m_clientConfiguration.executor.get());
        }
 
        template<typename HandlerT, typename OperationFuncT>
        void SubmitAsync(OperationFuncT operationFunc,
                         const HandlerT& handler,
                         const std::shared_ptr<const Aws::Client::AsyncCallerContext>& context = nullptr) const
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            Aws::Client::MakeAsyncOperation(operationFunc, clientThis, handler, context, clientThis->m_clientConfiguration.executor.get());
        }
 
        template<typename RequestT, typename OperationFuncT, typename std::enable_if<!IsEventStreamOperation<OperationFuncT>::value, int>::type = 0>
        auto SubmitCallable(OperationFuncT operationFunc,
                            const RequestT& request) const
            -> std::future<decltype((static_cast<const AwsServiceClientT*>(nullptr)->*operationFunc)(request))>
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            return Aws::Client::MakeCallableOperation(AwsServiceClientT::GetAllocationTag(), operationFunc, clientThis, request, clientThis->m_clientConfiguration.executor.get());
        }
 
        template<typename RequestT, typename OperationFuncT, typename std::enable_if<IsEventStreamOperation<OperationFuncT>::value, int>::type = 0>
        auto SubmitCallable(OperationFuncT operationFunc, /*note non-const ref*/ RequestT& request) const
            -> std::future<decltype((static_cast<const AwsServiceClientT*>(nullptr)->*operationFunc)(request))>
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            return Aws::Client::MakeCallableStreamingOperation(AwsServiceClientT::GetAllocationTag(), operationFunc, clientThis, request, clientThis->m_clientConfiguration.executor.get());
        }
 
        template<typename OperationFuncT>
        auto SubmitCallable(OperationFuncT operationFunc) const
            -> std::future<decltype((static_cast<const AwsServiceClientT*>(nullptr)->*operationFunc)())>
        {
            const AwsServiceClientT* clientThis = static_cast<const AwsServiceClientT*>(this);
            return Aws::Client::MakeCallableOperation(AwsServiceClientT::GetAllocationTag(), operationFunc, clientThis, clientThis->m_clientConfiguration.executor.get());
        }
    protected:
        template <typename OutcomeT, typename ClientT, typename AWSEndpointT, typename RequestT, typename HandlerT>
        friend class BidirectionalEventStreamingTask; // allow BidirectionalEventStreamingTask to access m_isInitialized
 
        template <typename ClientT, typename OutcomeT, typename RequestT, typename EncoderStreamT,
                  typename HandlerT, typename StreamReadyHandlerT>
        friend void SubmitBidirectionalStreamingRequest(
            const ClientT*, RequestT&, std::shared_ptr<RequestT>, std::shared_ptr<EncoderStreamT>,
            std::shared_ptr<Aws::Utils::Stream::HttpWriteDataStreamBuf>, const std::shared_ptr<Aws::Http::HttpRequest>&,
            Aws::Utils::Threading::Executor*,
            const StreamReadyHandlerT&, const HandlerT&, const std::shared_ptr<const Aws::Client::AsyncCallerContext>&);
 
        std::atomic<bool> m_isInitialized;
        mutable std::atomic<size_t> m_operationsProcessed;
        mutable std::condition_variable m_shutdownSignal;
        mutable std::mutex m_shutdownMutex;
 
        // TODO: track scheduled tasks
        // std::atomic<size_t> m_operationsScheduled;
    };
} // namespace Client
} // namespace Aws