cpp-software-patterns/RxTransport/MessageFuture.h at master · intact-software-systems/cpp-software-patterns · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#pragma once

#include"RxTransport/CommonDefines.h"
#include"RxTransport/MessagePolicy.h"

namespace BaseLib { namespace Concurrent {

/**
 * Future to wait for message delivery.
 *
 * Enables a client to wait for delivery before continuing
 */
template <typename T>
class MessageFuture
    : public Templates::Notifyable<MessageFuture<T>, WaitCondition, Mutex>
      , public Templates::ResultMethod1<std::pair<Status::FutureStatus, ChangeForReaderStatusKind>, int64>
      , public Templates::IsReceivedMethod
{
public:
    MessageFuture()
        : message_()
        , future_()
    { }

    MessageFuture(Message<T> message, std::shared_future<ChangeForReaderStatusKind> future)
        : message_(message)
        , future_(future)
    { }

    virtual ~MessageFuture()
    { }

    virtual bool WaitFor(int64 msecs = LONG_MAX) const
    {
        std::chrono::duration<int64, std::milli> duration(msecs);

        return future_.wait_for(duration) == std::future_status::ready;
    }

    std::pair<Status::FutureStatus, ChangeForReaderStatusKind> Result(int64 msecs = LONG_MAX) const
    {
        std::chrono::duration<int64, std::milli> duration(msecs);

        std::future_status status = future_.wait_for(duration);

        ChangeForReaderStatusKind kind = future_.get();

        switch(status)
        {
            case std::future_status::ready:
            {
                return std::pair<Status::FutureStatus, ChangeForReaderStatusKind>(Status::FutureStatus::Ready(), kind);
            }
            case std::future_status::timeout:
            {
                return std::pair<Status::FutureStatus, ChangeForReaderStatusKind>(Status::FutureStatus::Timeout(), kind);
            }
            case std::future_status::deferred:
            {
                return std::pair<Status::FutureStatus, ChangeForReaderStatusKind>(Status::FutureStatus::Deferred(), kind);
            }
        }
        return std::pair<Status::FutureStatus, ChangeForReaderStatusKind>(Status::FutureStatus::Deferred(), kind);
    }

    virtual bool IsReceived() const
    {
        static std::chrono::duration<int64, std::milli> duration(1);

        return future_.wait_for(duration) == std::future_status::ready;
    }

    virtual bool IsContent() const
    {
        return message_.IsContent();
    }

    const Message<T>& Data() const
    {
        return message_;
    }

    static MessageFuture<T> Empty()
    {
        return Concurrent::MessageFuture<T>();
    }

private:
    Message<T>                                            message_;
    mutable std::shared_future<ChangeForReaderStatusKind> future_;
};

}}