Awaiter.h

Go to the documentation of this file.

#pragma once

#include <swarmio/Mailbox.h>
#include <swarmio/Exception.h>
#include <mutex>
#include <chrono>
#include <atomic>
#include <condition_variable>
#include <exception>

namespace swarmio::services 
{
    template <class T>
    class Awaiter : public Mailbox 
    {
        private:

            uint64_t _requestIdentifier;

            T _response;

            bool _valid;

            std::exception_ptr _exception = nullptr;

            std::mutex _mutex;

            std::condition_variable _conditionVariable;

        protected:

            virtual T ExtractResponse(const Node* node, const data::Message* message) = 0;

            virtual bool IsFinished()
            {
                return _valid;
            }

        public:

            Awaiter(Endpoint* endpoint, uint64_t requestIdentifier)
                : Mailbox(endpoint), _requestIdentifier(requestIdentifier), _valid(false) { }

            Awaiter(const T& value)
                : Mailbox(), _requestIdentifier(0), _valid(true), _response(value) { }

            Awaiter(Awaiter&& other)
                : Mailbox(std::move(other))
            {
                // Lock
                std::unique_lock<std::mutex> guard(other._mutex);

                // Pass event handling to this instance
                other.FinishMovingTo(this);

                // Copy over values
                _valid = other._valid;
                _response = other._response;
                _requestIdentifier = other._requestIdentifier;
            }

            T GetResponse()
            {
                std::unique_lock<std::mutex> guard(_mutex);
                if (_valid)
                {
                    if (_exception)
                    {
                        std::rethrow_exception(_exception);
                    }
                    else
                    {
                        return _response;
                    }
                }
                else
                {
                    throw Exception("Response is not yet available");
                }
            }

            virtual bool ReceiveMessage(const Node* sender, const data::Message* message) override
            {
                // Check if we still need the response and then check the reply_to field
                if (!IsFinished() &&
                    message->header().reply_to() == _requestIdentifier)
                {
                    // Acquire lock
                    std::unique_lock<std::mutex> guard(_mutex);

                    // Extract the response
                    try
                    {
                        _response = ExtractResponse(sender, message);
                        _exception = nullptr;
                    }
                    catch (...)
                    {
                        _exception = std::current_exception();
                    }

                    // Mark as valid and notify all clients
                    _valid = true;
                    guard.unlock();
                    _conditionVariable.notify_all();

                    // Handled
                    return true;
                }
                else
                {
                    // Not handled
                    return false;
                }
            }

            bool WaitForResponse(const std::chrono::milliseconds& timeout)
            {
                // Acquire lock
                std::unique_lock<std::mutex> guard(_mutex);

                // Wait
                return _conditionVariable.wait_for(guard, timeout, [this]{ return _valid; });
            }

            bool HasException()
            {
                std::unique_lock<std::mutex> guard(_mutex);
                if (_valid)
                {
                    return (bool)_exception;
                }
                else
                {
                    throw Exception("Response is not yet available");
                }
            }

            uint64_t GetIdentifier() const
            {
                return _requestIdentifier;
            }
    };
}