include/pulsar/Consumer.h - pulsar-client-cpp - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */
 #ifndef CONSUMER_HPP_
 #define CONSUMER_HPP_

 #include <pulsar/BrokerConsumerStats.h>
 #include <pulsar/ConsumerConfiguration.h>
 #include <pulsar/TypedMessage.h>
 #include <pulsar/defines.h>

 #include <iostream>

 namespace pulsar {
 class PulsarWrapper;
 class ConsumerImplBase;
 class PulsarFriend;
 typedef std::shared_ptr<ConsumerImplBase> ConsumerImplBasePtr;
 /**
  *
  */
 class PULSAR_PUBLIC Consumer {
    public:
     /**
      * Construct an uninitialized consumer object
      */
     Consumer();
     virtual ~Consumer() = default;

     /**
      * @return the topic this consumer is subscribed to
      */
     const std::string& getTopic() const;

     /**
      * @return the consumer name
      */
     const std::string& getSubscriptionName() const;

     /**
      * Unsubscribe the current consumer from the topic.
      *
      * This method will block until the operation is completed. Once the consumer is
      * unsubscribed, no more messages will be received and subsequent new messages
      * will not be retained for this consumer.
      *
      * This consumer object cannot be reused.
      *
      * @see asyncUnsubscribe
      * @return Result::ResultOk if the unsubscribe operation completed successfully
      * @return Result::ResultError if the unsubscribe operation failed
      */
     Result unsubscribe();

     /**
      * Asynchronously unsubscribe the current consumer from the topic.
      *
      * This method will block until the operation is completed. Once the consumer is
      * unsubscribed, no more messages will be received and subsequent new messages
      * will not be retained for this consumer.
      *
      * This consumer object cannot be reused.
      *
      * @param callback the callback to get notified when the operation is complete
      */
     void unsubscribeAsync(ResultCallback callback);

     /**
      * Receive a single message.
      *
      * If a message is not immediately available, this method will block until a new
      * message is available.
      *
      * @param msg a non-const reference where the received message will be copied
      * @return ResultOk when a message is received
      * @return ResultInvalidConfiguration if a message listener had been set in the configuration
      */
     Result receive(Message& msg);

     template <typename T>
     Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder) {
         Message rawMsg;
         auto result = receive(rawMsg);
         msg = TypedMessage<T>{rawMsg, decoder};
         return result;
     }

     /**
      *
      * @param msg a non-const reference where the received message will be copied
      * @param timeoutMs the receive timeout in milliseconds
      * @return ResultOk if a message was received
      * @return ResultTimeout if the receive timeout was triggered
      * @return ResultInvalidConfiguration if a message listener had been set in the configuration
      */
     Result receive(Message& msg, int timeoutMs);

     template <typename T>
     Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder) {
         Message rawMsg;
         auto result = receive(rawMsg, timeoutMs);
         msg = TypedMessage<T>{rawMsg, decoder};
         return result;
     }

     /**
      * Receive a single message
      * <p>
      * Retrieves a message when it will be available and completes callback with received message.
      * </p>
      * <p>
      * receiveAsync() should be called subsequently once callback gets completed with received message.
      * Else it creates <i> backlog of receive requests </i> in the application.
      * </p>
      * @param ReceiveCallback will be completed when message is available
      */
     void receiveAsync(ReceiveCallback callback);

     template <typename T>
     void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
                       typename TypedMessage<T>::Decoder decoder) {
         receiveAsync([callback, decoder](Result result, const Message& msg) {
             callback(result, TypedMessage<T>{msg, decoder});
         });
     }

     /**
      * Batch receiving messages.
      *
      * <p>This calls blocks until has enough messages or wait timeout, more details to see {@link
      * BatchReceivePolicy}.
      *
      * @param msgs a non-const reference where the received messages will be copied
      * @return ResultOk when a message is received
      * @return ResultInvalidConfiguration if a message listener had been set in the configuration
      */
     Result batchReceive(Messages& msgs);

     /**
      * Async Batch receiving messages.
      * <p>
      * Retrieves a message when it will be available and completes callback with received message.
      * </p>
      * <p>
      * batchReceiveAsync() should be called subsequently once callback gets completed with received message.
      * Else it creates <i> backlog of receive requests </i> in the application.
      * </p>
      * @param BatchReceiveCallback will be completed when messages are available.
      */
     void batchReceiveAsync(BatchReceiveCallback callback);

     /**
      * Acknowledge the reception of a single message.
      *
      * This method will block until an acknowledgement is sent to the broker. After
      * that, the message will not be re-delivered to this consumer.
      *
      * @see asyncAcknowledge
      * @param message the message to acknowledge
      * @return ResultOk if the message was successfully acknowledged
      * @return ResultError if there was a failure
      */
     Result acknowledge(const Message& message);

     /**
      * Acknowledge the reception of a single message.
      *
      * This method is blocked until an acknowledgement is sent to the broker. After that, the message is not
      * re-delivered to the consumer.
      *
      * @see asyncAcknowledge
      * @param messageId the MessageId to acknowledge
      * @return ResultOk if the messageId is successfully acknowledged
      */
     Result acknowledge(const MessageId& messageId);

     /**
      * Acknowledge the consumption of a list of message.
      * @param messageIdList
      */
     Result acknowledge(const MessageIdList& messageIdList);

     /**
      * Asynchronously acknowledge the reception of a single message.
      *
      * This method will initiate the operation and return immediately. The provided callback
      * will be triggered when the operation is complete.
      *
      * @param message the message to acknowledge
      * @param callback callback that will be triggered when the message has been acknowledged
      */
     void acknowledgeAsync(const Message& message, ResultCallback callback);

     /**
      * Asynchronously acknowledge the reception of a single message.
      *
      * This method initiates the operation and returns the result immediately. The provided callback
      * is triggered when the operation is completed.
      *
      * @param messageId the messageId to acknowledge
      * @param callback the callback that is triggered when the message has been acknowledged or not
      */
     void acknowledgeAsync(const MessageId& messageId, ResultCallback callback);

     /**
      * Asynchronously acknowledge the consumption of a list of message.
      * @param messageIdList
      * @param callback the callback that is triggered when the message has been acknowledged or not
      * @return
      */
     void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback);

     /**
      * Acknowledge the reception of all the messages in the stream up to (and including)
      * the provided message.
      *
      * This method will block until an acknowledgement is sent to the broker. After
      * that, the messages will not be re-delivered to this consumer.
      *
      * Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
      *
      * It's equivalent to calling asyncAcknowledgeCumulative(const Message&, ResultCallback) and
      * waiting for the callback to be triggered.
      *
      * @param message the last message in the stream to acknowledge
      * @return ResultOk if the message was successfully acknowledged. All previously delivered messages for
      * this topic are also acknowledged.
      * @return ResultError if there was a failure
      */
     Result acknowledgeCumulative(const Message& message);

     /**
      * Acknowledge the reception of all the messages in the stream up to (and including)
      * the provided message.
      *
      * This method is blocked until an acknowledgement is sent to the broker. After
      * that, the message is not re-delivered to this consumer.
      *
      * Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
      *
      * It is equivalent to calling the asyncAcknowledgeCumulative(const Message&, ResultCallback) method and
      * waiting for the callback to be triggered.
      *
      * @param messageId the last messageId in the stream to acknowledge
      * @return ResultOk if the message is successfully acknowledged. All previously delivered messages for
      * this topic are also acknowledged.
      */
     Result acknowledgeCumulative(const MessageId& messageId);

     /**
      * Asynchronously acknowledge the reception of all the messages in the stream up to (and
      * including) the provided message.
      *
      * This method will initiate the operation and return immediately. The provided callback
      * will be triggered when the operation is complete.
      *
      * @param message the message to acknowledge
      * @param callback callback that will be triggered when the message has been acknowledged
      */
     void acknowledgeCumulativeAsync(const Message& message, ResultCallback callback);

     /**
      * Asynchronously acknowledge the reception of all the messages in the stream up to (and
      * including) the provided message.
      *
      * This method initiates the operation and returns the result immediately. The provided callback
      * is triggered when the operation is completed.
      *
      * @param messageId the messageId to acknowledge
      * @param callback the callback that is triggered when the message has been acknowledged or not
      */
     void acknowledgeCumulativeAsync(const MessageId& messageId, ResultCallback callback);

     /**
      * Acknowledge the failure to process a single message.
      * <p>
      * When a message is "negatively acked" it will be marked for redelivery after
      * some fixed delay. The delay is configurable when constructing the consumer
      * with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}.
      * <p>
      * This call is not blocking.
      *
      * <p>
      * Example of usage:
      * <pre><code>
      * while (true) {
      *     Message msg;
      *     consumer.receive(msg);
      *
      *     try {
      *          // Process message...
      *
      *          consumer.acknowledge(msg);
      *     } catch (Throwable t) {
      *          log.warn("Failed to process message");
      *          consumer.negativeAcknowledge(msg);
      *     }
      * }
      * </code></pre>
      *
      * @param message
      *            The {@code Message} to be acknowledged
      */
     void negativeAcknowledge(const Message& message);

     /**
      * Acknowledge the failure to process a single message.
      * <p>
      * When a message is "negatively acked" it will be marked for redelivery after
      * some fixed delay. The delay is configurable when constructing the consumer
      * with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}.
      * <p>
      * This call is not blocking.
      *
      * <p>
      * Example of usage:
      * <pre><code>
      * while (true) {
      *     Message msg;
      *     consumer.receive(msg);
      *
      *     try {
      *          // Process message...
      *
      *          consumer.acknowledge(msg);
      *     } catch (Throwable t) {
      *          log.warn("Failed to process message");
      *          consumer.negativeAcknowledge(msg);
      *     }
      * }
      * </code></pre>
      *
      * @param messageId
      *            The {@code MessageId} to be acknowledged
      */
     void negativeAcknowledge(const MessageId& messageId);

     /**
      * Close the consumer and stop the broker to push more messages
      */
     Result close();

     /**
      * Asynchronously close the consumer and stop the broker to push more messages
      *
      */
     void closeAsync(ResultCallback callback);

     /**
      * Pause receiving messages via the messageListener, till resumeMessageListener() is called.
      */
     Result pauseMessageListener();

     /**
      * Resume receiving the messages via the messageListener.
      * Asynchronously receive all the messages enqueued from time pauseMessageListener() was called.
      */
     Result resumeMessageListener();

     /**
      * Redelivers all the unacknowledged messages. In Failover mode, the request is ignored if the consumer is
      * not
      * active for the given topic. In Shared mode, the consumers messages to be redelivered are distributed
      * across all
      * the connected consumers. This is a non blocking call and doesn't throw an exception. In case the
      * connection
      * breaks, the messages are redelivered after reconnect.
      */
     void redeliverUnacknowledgedMessages();

     /**
      * Gets Consumer Stats from broker.
      * The stats are cached for 30 seconds, if a call is made before the stats returned by the previous call
      * expires
      * then cached data will be returned. BrokerConsumerStats::isValid() function can be used to check if the
      * stats are
      * still valid.
      *
      * @param brokerConsumerStats - if the function returns ResultOk, this object will contain consumer stats
      *
      * @note This is a blocking call with timeout of thirty seconds.
      */
     Result getBrokerConsumerStats(BrokerConsumerStats& brokerConsumerStats);

     /**
      * Asynchronous call to gets Consumer Stats from broker.
      * The stats are cached for 30 seconds, if a call is made before the stats returned by the previous call
      * expires
      * then cached data will be returned. BrokerConsumerStats::isValid() function can be used to check if the
      * stats are
      * still valid.
      *
      * @param callback - callback function to get the brokerConsumerStats,
      *                   if result is ResultOk then the brokerConsumerStats will be populated
      */
     void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback);

     /**
      * Reset the subscription associated with this consumer to a specific message id.
      * The message id can either be a specific message or represent the first or last messages in the topic.
      *
      * Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the
      * seek() on the individual partitions.
      *
      * @param messageId
      *            the message id where to reposition the subscription
      */
     Result seek(const MessageId& messageId);

     /**
      * Reset the subscription associated with this consumer to a specific message publish time.
      *
      * @param timestamp
      *            the message publish time where to reposition the subscription
      */
     Result seek(uint64_t timestamp);

     /**
      * Asynchronously reset the subscription associated with this consumer to a specific message id.
      * The message id can either be a specific message or represent the first or last messages in the topic.
      *
      * Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the
      * seek() on the individual partitions.
      *
      * @param messageId
      *            the message id where to reposition the subscription
      */
     virtual void seekAsync(const MessageId& messageId, ResultCallback callback);

     /**
      * Asynchronously reset the subscription associated with this consumer to a specific message publish time.
      *
      * @param timestamp
      *            the message publish time where to reposition the subscription
      */
     virtual void seekAsync(uint64_t timestamp, ResultCallback callback);

     /**
      * @return Whether the consumer is currently connected to the broker
      */
     bool isConnected() const;

     /**
      * Asynchronously get an ID of the last available message or a message ID with -1 as an entryId if the
      * topic is empty.
      */
     void getLastMessageIdAsync(GetLastMessageIdCallback callback);

     /**
      * Get an ID of the last available message or a message ID with -1 as an entryId if the topic is empty.
      */
     Result getLastMessageId(MessageId& messageId);

    private:
     ConsumerImplBasePtr impl_;
     explicit Consumer(ConsumerImplBasePtr);

     friend class PulsarFriend;
     friend class PulsarWrapper;
     friend class MultiTopicsConsumerImpl;
     friend class ConsumerImpl;
     friend class ClientImpl;
     friend class ConsumerTest;
 };
 }  // namespace pulsar

 #endif /* CONSUMER_HPP_ */
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/
	#ifndef CONSUMER_HPP_
	#define CONSUMER_HPP_

	#include <pulsar/BrokerConsumerStats.h>
	#include <pulsar/ConsumerConfiguration.h>
	#include <pulsar/TypedMessage.h>
	#include <pulsar/defines.h>

	#include <iostream>

	namespace pulsar {
	class PulsarWrapper;
	class ConsumerImplBase;
	class PulsarFriend;
	typedef std::shared_ptr<ConsumerImplBase> ConsumerImplBasePtr;
	/**
	*
	*/
	class PULSAR_PUBLIC Consumer {
	public:
	/**
	* Construct an uninitialized consumer object
	*/
	Consumer();
	virtual ~Consumer() = default;

	/**
	* @return the topic this consumer is subscribed to
	*/
	const std::string& getTopic() const;

	/**
	* @return the consumer name
	*/
	const std::string& getSubscriptionName() const;

	/**
	* Unsubscribe the current consumer from the topic.
	*
	* This method will block until the operation is completed. Once the consumer is
	* unsubscribed, no more messages will be received and subsequent new messages
	* will not be retained for this consumer.
	*
	* This consumer object cannot be reused.
	*
	* @see asyncUnsubscribe
	* @return Result::ResultOk if the unsubscribe operation completed successfully
	* @return Result::ResultError if the unsubscribe operation failed
	*/
	Result unsubscribe();

	/**
	* Asynchronously unsubscribe the current consumer from the topic.
	*
	* This method will block until the operation is completed. Once the consumer is
	* unsubscribed, no more messages will be received and subsequent new messages
	* will not be retained for this consumer.
	*
	* This consumer object cannot be reused.
	*
	* @param callback the callback to get notified when the operation is complete
	*/
	void unsubscribeAsync(ResultCallback callback);

	/**
	* Receive a single message.
	*
	* If a message is not immediately available, this method will block until a new
	* message is available.
	*
	* @param msg a non-const reference where the received message will be copied
	* @return ResultOk when a message is received
	* @return ResultInvalidConfiguration if a message listener had been set in the configuration
	*/
	Result receive(Message& msg);

	template <typename T>
	Result receive(TypedMessage<T>& msg, typename TypedMessage<T>::Decoder decoder) {
	Message rawMsg;
	auto result = receive(rawMsg);
	msg = TypedMessage<T>{rawMsg, decoder};
	return result;
	}

	/**
	*
	* @param msg a non-const reference where the received message will be copied
	* @param timeoutMs the receive timeout in milliseconds
	* @return ResultOk if a message was received
	* @return ResultTimeout if the receive timeout was triggered
	* @return ResultInvalidConfiguration if a message listener had been set in the configuration
	*/
	Result receive(Message& msg, int timeoutMs);

	template <typename T>
	Result receive(TypedMessage<T>& msg, int timeoutMs, typename TypedMessage<T>::Decoder decoder) {
	Message rawMsg;
	auto result = receive(rawMsg, timeoutMs);
	msg = TypedMessage<T>{rawMsg, decoder};
	return result;
	}

	/**
	* Receive a single message
	* <p>
	* Retrieves a message when it will be available and completes callback with received message.
	* </p>
	* <p>
	* receiveAsync() should be called subsequently once callback gets completed with received message.
	* Else it creates <i> backlog of receive requests </i> in the application.
	* </p>
	* @param ReceiveCallback will be completed when message is available
	*/
	void receiveAsync(ReceiveCallback callback);

	template <typename T>
	void receiveAsync(std::function<void(Result result, const TypedMessage<T>&)> callback,
	typename TypedMessage<T>::Decoder decoder) {
	receiveAsync([callback, decoder](Result result, const Message& msg) {
	callback(result, TypedMessage<T>{msg, decoder});
	});
	}

	/**
	* Batch receiving messages.
	*
	* <p>This calls blocks until has enough messages or wait timeout, more details to see {@link
	* BatchReceivePolicy}.
	*
	* @param msgs a non-const reference where the received messages will be copied
	* @return ResultOk when a message is received
	* @return ResultInvalidConfiguration if a message listener had been set in the configuration
	*/
	Result batchReceive(Messages& msgs);

	/**
	* Async Batch receiving messages.
	* <p>
	* Retrieves a message when it will be available and completes callback with received message.
	* </p>
	* <p>
	* batchReceiveAsync() should be called subsequently once callback gets completed with received message.
	* Else it creates <i> backlog of receive requests </i> in the application.
	* </p>
	* @param BatchReceiveCallback will be completed when messages are available.
	*/
	void batchReceiveAsync(BatchReceiveCallback callback);

	/**
	* Acknowledge the reception of a single message.
	*
	* This method will block until an acknowledgement is sent to the broker. After
	* that, the message will not be re-delivered to this consumer.
	*
	* @see asyncAcknowledge
	* @param message the message to acknowledge
	* @return ResultOk if the message was successfully acknowledged
	* @return ResultError if there was a failure
	*/
	Result acknowledge(const Message& message);

	/**
	* Acknowledge the reception of a single message.
	*
	* This method is blocked until an acknowledgement is sent to the broker. After that, the message is not
	* re-delivered to the consumer.
	*
	* @see asyncAcknowledge
	* @param messageId the MessageId to acknowledge
	* @return ResultOk if the messageId is successfully acknowledged
	*/
	Result acknowledge(const MessageId& messageId);

	/**
	* Acknowledge the consumption of a list of message.
	* @param messageIdList
	*/
	Result acknowledge(const MessageIdList& messageIdList);

	/**
	* Asynchronously acknowledge the reception of a single message.
	*
	* This method will initiate the operation and return immediately. The provided callback
	* will be triggered when the operation is complete.
	*
	* @param message the message to acknowledge
	* @param callback callback that will be triggered when the message has been acknowledged
	*/
	void acknowledgeAsync(const Message& message, ResultCallback callback);

	/**
	* Asynchronously acknowledge the reception of a single message.
	*
	* This method initiates the operation and returns the result immediately. The provided callback
	* is triggered when the operation is completed.
	*
	* @param messageId the messageId to acknowledge
	* @param callback the callback that is triggered when the message has been acknowledged or not
	*/
	void acknowledgeAsync(const MessageId& messageId, ResultCallback callback);

	/**
	* Asynchronously acknowledge the consumption of a list of message.
	* @param messageIdList
	* @param callback the callback that is triggered when the message has been acknowledged or not
	* @return
	*/
	void acknowledgeAsync(const MessageIdList& messageIdList, ResultCallback callback);

	/**
	* Acknowledge the reception of all the messages in the stream up to (and including)
	* the provided message.
	*
	* This method will block until an acknowledgement is sent to the broker. After
	* that, the messages will not be re-delivered to this consumer.
	*
	* Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
	*
	* It's equivalent to calling asyncAcknowledgeCumulative(const Message&, ResultCallback) and
	* waiting for the callback to be triggered.
	*
	* @param message the last message in the stream to acknowledge
	* @return ResultOk if the message was successfully acknowledged. All previously delivered messages for
	* this topic are also acknowledged.
	* @return ResultError if there was a failure
	*/
	Result acknowledgeCumulative(const Message& message);

	/**
	* Acknowledge the reception of all the messages in the stream up to (and including)
	* the provided message.
	*
	* This method is blocked until an acknowledgement is sent to the broker. After
	* that, the message is not re-delivered to this consumer.
	*
	* Cumulative acknowledge cannot be used when the consumer type is set to ConsumerShared.
	*
	* It is equivalent to calling the asyncAcknowledgeCumulative(const Message&, ResultCallback) method and
	* waiting for the callback to be triggered.
	*
	* @param messageId the last messageId in the stream to acknowledge
	* @return ResultOk if the message is successfully acknowledged. All previously delivered messages for
	* this topic are also acknowledged.
	*/
	Result acknowledgeCumulative(const MessageId& messageId);

	/**
	* Asynchronously acknowledge the reception of all the messages in the stream up to (and
	* including) the provided message.
	*
	* This method will initiate the operation and return immediately. The provided callback
	* will be triggered when the operation is complete.
	*
	* @param message the message to acknowledge
	* @param callback callback that will be triggered when the message has been acknowledged
	*/
	void acknowledgeCumulativeAsync(const Message& message, ResultCallback callback);

	/**
	* Asynchronously acknowledge the reception of all the messages in the stream up to (and
	* including) the provided message.
	*
	* This method initiates the operation and returns the result immediately. The provided callback
	* is triggered when the operation is completed.
	*
	* @param messageId the messageId to acknowledge
	* @param callback the callback that is triggered when the message has been acknowledged or not
	*/
	void acknowledgeCumulativeAsync(const MessageId& messageId, ResultCallback callback);

	/**
	* Acknowledge the failure to process a single message.
	* <p>
	* When a message is "negatively acked" it will be marked for redelivery after
	* some fixed delay. The delay is configurable when constructing the consumer
	* with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}.
	* <p>
	* This call is not blocking.
	*
	* <p>
	* Example of usage:
	* <pre><code>
	* while (true) {
	* Message msg;
	* consumer.receive(msg);
	*
	* try {
	* // Process message...
	*
	* consumer.acknowledge(msg);
	* } catch (Throwable t) {
	* log.warn("Failed to process message");
	* consumer.negativeAcknowledge(msg);
	* }
	* }
	* </code></pre>
	*
	* @param message
	* The {@code Message} to be acknowledged
	*/
	void negativeAcknowledge(const Message& message);

	/**
	* Acknowledge the failure to process a single message.
	* <p>
	* When a message is "negatively acked" it will be marked for redelivery after
	* some fixed delay. The delay is configurable when constructing the consumer
	* with {@link ConsumerConfiguration#setNegativeAckRedeliveryDelayMs}.
	* <p>
	* This call is not blocking.
	*
	* <p>
	* Example of usage:
	* <pre><code>
	* while (true) {
	* Message msg;
	* consumer.receive(msg);
	*
	* try {
	* // Process message...
	*
	* consumer.acknowledge(msg);
	* } catch (Throwable t) {
	* log.warn("Failed to process message");
	* consumer.negativeAcknowledge(msg);
	* }
	* }
	* </code></pre>
	*
	* @param messageId
	* The {@code MessageId} to be acknowledged
	*/
	void negativeAcknowledge(const MessageId& messageId);

	/**
	* Close the consumer and stop the broker to push more messages
	*/
	Result close();

	/**
	* Asynchronously close the consumer and stop the broker to push more messages
	*
	*/
	void closeAsync(ResultCallback callback);

	/**
	* Pause receiving messages via the messageListener, till resumeMessageListener() is called.
	*/
	Result pauseMessageListener();

	/**
	* Resume receiving the messages via the messageListener.
	* Asynchronously receive all the messages enqueued from time pauseMessageListener() was called.
	*/
	Result resumeMessageListener();

	/**
	* Redelivers all the unacknowledged messages. In Failover mode, the request is ignored if the consumer is
	* not
	* active for the given topic. In Shared mode, the consumers messages to be redelivered are distributed
	* across all
	* the connected consumers. This is a non blocking call and doesn't throw an exception. In case the
	* connection
	* breaks, the messages are redelivered after reconnect.
	*/
	void redeliverUnacknowledgedMessages();

	/**
	* Gets Consumer Stats from broker.
	* The stats are cached for 30 seconds, if a call is made before the stats returned by the previous call
	* expires
	* then cached data will be returned. BrokerConsumerStats::isValid() function can be used to check if the
	* stats are
	* still valid.
	*
	* @param brokerConsumerStats - if the function returns ResultOk, this object will contain consumer stats
	*
	* @note This is a blocking call with timeout of thirty seconds.
	*/
	Result getBrokerConsumerStats(BrokerConsumerStats& brokerConsumerStats);

	/**
	* Asynchronous call to gets Consumer Stats from broker.
	* The stats are cached for 30 seconds, if a call is made before the stats returned by the previous call
	* expires
	* then cached data will be returned. BrokerConsumerStats::isValid() function can be used to check if the
	* stats are
	* still valid.
	*
	* @param callback - callback function to get the brokerConsumerStats,
	* if result is ResultOk then the brokerConsumerStats will be populated
	*/
	void getBrokerConsumerStatsAsync(BrokerConsumerStatsCallback callback);

	/**
	* Reset the subscription associated with this consumer to a specific message id.
	* The message id can either be a specific message or represent the first or last messages in the topic.
	*
	* Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the
	* seek() on the individual partitions.
	*
	* @param messageId
	* the message id where to reposition the subscription
	*/
	Result seek(const MessageId& messageId);

	/**
	* Reset the subscription associated with this consumer to a specific message publish time.
	*
	* @param timestamp
	* the message publish time where to reposition the subscription
	*/
	Result seek(uint64_t timestamp);

	/**
	* Asynchronously reset the subscription associated with this consumer to a specific message id.
	* The message id can either be a specific message or represent the first or last messages in the topic.
	*
	* Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the
	* seek() on the individual partitions.
	*
	* @param messageId
	* the message id where to reposition the subscription
	*/
	virtual void seekAsync(const MessageId& messageId, ResultCallback callback);

	/**
	* Asynchronously reset the subscription associated with this consumer to a specific message publish time.
	*
	* @param timestamp
	* the message publish time where to reposition the subscription
	*/
	virtual void seekAsync(uint64_t timestamp, ResultCallback callback);

	/**
	* @return Whether the consumer is currently connected to the broker
	*/
	bool isConnected() const;

	/**
	* Asynchronously get an ID of the last available message or a message ID with -1 as an entryId if the
	* topic is empty.
	*/
	void getLastMessageIdAsync(GetLastMessageIdCallback callback);

	/**
	* Get an ID of the last available message or a message ID with -1 as an entryId if the topic is empty.
	*/
	Result getLastMessageId(MessageId& messageId);

	private:
	ConsumerImplBasePtr impl_;
	explicit Consumer(ConsumerImplBasePtr);

	friend class PulsarFriend;
	friend class PulsarWrapper;
	friend class MultiTopicsConsumerImpl;
	friend class ConsumerImpl;
	friend class ClientImpl;
	friend class ConsumerTest;
	};
	} // namespace pulsar

	#endif /* CONSUMER_HPP_ */