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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
/**
* DeferredConfirm.h
*
* Deferred callback for RabbitMQ-specific publisher confirms mechanism.
*
* @author Marcin Gibula <m.gibula@gmail.com>
* @copyright 2018 Copernica BV
*/
/**
* Include guard
*/
#pragma once
/**
* Set up namespace
*/
namespace AMQP {
/**
* We extend from the default deferred and add extra functionality
*/
class DeferredConfirm : public Deferred
{
private:
/**
* Callback to execute when server confirms that message is processed
* @var AckCallback
*/
AckCallback _ackCallback;
/**
* Callback to execute when server sends negative acknowledgement
* @var NackCallback
*/
NackCallback _nackCallback;
/**
* Process an ACK frame
*
* @param frame The frame to process
*/
void process(BasicAckFrame &frame);
/**
* Process an ACK frame
*
* @param frame The frame to process
*/
void process(BasicNackFrame &frame);
/**
* The channel implementation may call our
* private members and construct us
*/
friend class ChannelImpl;
friend class BasicAckFrame;
friend class BasicNackFrame;
public:
/**
* Protected constructor that can only be called
* from within the channel implementation
*
* Note: this constructor _should_ be protected, but because make_shared
* will then not work, we have decided to make it public after all,
* because the work-around would result in not-so-easy-to-read code.
*
* @param boolean are we already failed?
*/
DeferredConfirm(bool failed = false) : Deferred(failed) {}
public:
/**
* Register the function that is called when channel is put in publisher
* confirmed mode
* @param callback
*/
inline DeferredConfirm &onSuccess(const SuccessCallback& callback) { return onSuccess(SuccessCallback(callback)); }
DeferredConfirm &onSuccess(SuccessCallback&& callback)
{
// call base
Deferred::onSuccess(std::move(callback));
// allow chaining
return *this;
}
/**
* Callback that is called when the broker confirmed message publication
* @param callback the callback to execute
*/
inline DeferredConfirm &onAck(const AckCallback& callback) { return onAck(AckCallback(callback)); }
DeferredConfirm &onAck(AckCallback&& callback)
{
// store callback
_ackCallback = std::move(callback);
// allow chaining
return *this;
}
/**
* Callback that is called when the broker denied message publication
* @param callback the callback to execute
*/
inline DeferredConfirm &onNack(const NackCallback& callback) { return onNack(NackCallback(callback)); }
DeferredConfirm &onNack(NackCallback&& callback)
{
// store callback
_nackCallback = std::move(callback);
// allow chaining
return *this;
}
};
/**
* End namespace
*/
}
|
