1/*
2 * Copyright (c) 2018, Matias Fontanini
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are
7 * met:
8 *
9 * * Redistributions of source code must retain the above copyright
10 * notice, this list of conditions and the following disclaimer.
11 * * Redistributions in binary form must reproduce the above
12 * copyright notice, this list of conditions and the following disclaimer
13 * in the documentation and/or other materials provided with the
14 * distribution.
15 *
16 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27 *
28 */
29
30#ifndef CPPKAFKA_EVENT_H
31#define CPPKAFKA_EVENT_H
32
33#include <memory>
34#include <string>
35#include <vector>
36#include "error.h"
37#include "message.h"
38#include "topic_partition.h"
39#include "topic_partition_list.h"
40
41namespace cppkafka {
42
43class Event {
44public:
45 /**
46 * Construct an Event from a rdkafka event handle and take ownership of it
47 *
48 * /param handle The handle to construct this event from
49 */
50 Event(rd_kafka_event_t* handle);
51
52 /**
53 * Returns the name of this event
54 */
55 std::string get_name() const;
56
57 /**
58 * Returns the type of this event
59 */
60 rd_kafka_event_type_t get_type() const;
61
62 /**
63 * \brief Gets the next message contained in this event.
64 *
65 * This call is only valid if the event type is one of:
66 * * RD_KAFKA_EVENT_FETCH
67 * * RD_KAFKA_EVENT_DR
68 *
69 * \note The returned message's lifetime *is tied to this Event*. That is, if the event
70 * is free'd so will the contents of the message.
71 */
72 Message get_next_message() const;
73
74 /**
75 * \brief Gets all messages in this event (if any)
76 *
77 * This call is only valid if the event type is one of:
78 * * RD_KAFKA_EVENT_FETCH
79 * * RD_KAFKA_EVENT_DR
80 *
81 * \note The returned messages' lifetime *is tied to this Event*. That is, if the event
82 * is free'd so will the contents of the messages.
83 *
84 * \return A vector containing 0 or more messages
85 */
86 std::vector<Message> get_messages();
87
88 /**
89 * \brief Gets all messages in this event (if any)
90 *
91 * This call is only valid if the event type is one of:
92 * * RD_KAFKA_EVENT_FETCH
93 * * RD_KAFKA_EVENT_DR
94 *
95 * \param allocator The allocator to use on the output vector
96 *
97 * \note The returned messages' lifetime *is tied to this Event*. That is, if the event
98 * is free'd so will the contents of the messages.
99 *
100 * \return A vector containing 0 or more messages
101 */
102 template <typename Allocator>
103 std::vector<Message, Allocator> get_messages(const Allocator allocator);
104
105 /**
106 * \brief Gets the number of messages contained in this event
107 *
108 * This call is only valid if the event type is one of:
109 * * RD_KAFKA_EVENT_FETCH
110 * * RD_KAFKA_EVENT_DR
111 */
112 size_t get_message_count() const;
113
114 /**
115 * \brief Returns the error in this event
116 */
117 Error get_error() const;
118
119 /**
120 * Gets the opaque pointer in this event
121 */
122 void* get_opaque() const;
123
124#if RD_KAFKA_VERSION >= RD_KAFKA_EVENT_STATS_SUPPORT_VERSION
125 /**
126 * \brief Gets the stats in this event
127 *
128 * This call is only valid if the event type is RD_KAFKA_EVENT_STATS
129 */
130 std::string get_stats() const {
131 return rd_kafka_event_stats(handle_.get());
132 }
133#endif
134
135 /**
136 * \brief Gets the topic/partition for this event
137 *
138 * This call is only valid if the event type is RD_KAFKA_EVENT_ERROR
139 */
140 TopicPartition get_topic_partition() const;
141
142 /**
143 * \brief Gets the list of topic/partitions in this event
144 *
145 * This call is only valid if the event type is one of:
146 * * RD_KAFKA_EVENT_REBALANCE
147 * * RD_KAFKA_EVENT_OFFSET_COMMIT
148 */
149 TopicPartitionList get_topic_partition_list() const;
150
151 /**
152 * Check whether this event is valid
153 *
154 * /return true iff this event has a valid (non-null) handle inside
155 */
156 operator bool() const;
157private:
158 using HandlePtr = std::unique_ptr<rd_kafka_event_t, decltype(&rd_kafka_event_destroy)>;
159
160 HandlePtr handle_;
161};
162
163template <typename Allocator>
164std::vector<Message, Allocator> Event::get_messages(const Allocator allocator) {
165 const size_t total_messages = get_message_count();
166 std::vector<const rd_kafka_message_t*> raw_messages(total_messages);
167 const auto messages_read = rd_kafka_event_message_array(handle_.get(),
168 raw_messages.data(),
169 total_messages);
170 std::vector<Message, Allocator> output(allocator);
171 output.reserve(messages_read);
172 for (auto message : raw_messages) {
173 output.emplace_back(Message::make_non_owning(const_cast<rd_kafka_message_t*>(message)));
174 }
175 return output;
176}
177
178} // cppkafka
179
180#endif // CPPKAFKA_EVENT_H
181