A bunch of minor updates, fixes etc

CMakeLists.txt

@@ -5,6 +5,8 @@ project(amqpcpp)
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall")
 set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 
+option(ENABLE_SSL_SUPPORT "Enable SSL support" ON)
+
 add_library(
     amqpcpp
     SHARED
@@ -31,8 +33,22 @@ add_library(
 )
 target_include_directories(amqpcpp-static PUBLIC include/)
 
+if (NOT ENABLE_SSL_SUPPORT)
+    add_definitions(-DAMQP_NO_SSL)
+endif()
+
+if(MSVC)
+    target_compile_options(amqpcpp PUBLIC "/Zc:__cplusplus")
+    target_compile_options(amqpcpp-static PUBLIC "/Zc:__cplusplus")
+endif()
+
 add_subdirectory(rabbitmq-c)
 target_include_directories(rabbitmq PUBLIC rabbitmq-c/librabbitmq/)
 
-target_link_libraries(amqpcpp rabbitmq ssl crypto)
-target_link_libraries(amqpcpp-static rabbitmq ssl crypto)
+if (ENABLE_SSL_SUPPORT)
+    target_link_libraries(amqpcpp rabbitmq ssl crypto)
+    target_link_libraries(amqpcpp-static rabbitmq ssl crypto)
+else()
+    target_link_libraries(amqpcpp rabbitmq)
+    target_link_libraries(amqpcpp-static rabbitmq)
+endif()

examples/example_consume.cpp

@@ -10,7 +10,7 @@ int onCancel(AMQPMessage * message ) {
 }
 
 int  onMessage( AMQPMessage * message  ) {
-	uint32_t j = 0;
+	size_t j = 0;
 	char * data = message->getMessage(&j);
 	if (data)
 		  cout << data << endl;

examples/example_get.cpp

@@ -21,7 +21,7 @@ int main () {
 
 			cout << "count: "<<  m->getMessageCount() << endl;											 
 			if (m->getMessageCount() > -1) {
-			uint32_t j = 0;
+			size_t j = 0;
 			cout << "message\n"<< m->getMessage(&j) << "\nmessage key: "<<  m->getRoutingKey() << endl;
 			cout << "exchange: "<<  m->getExchange() << endl;											
 			cout << "Content-type: "<< m->getHeader("Content-type") << endl;	

include/AMQPcpp.h

@@ -41,18 +41,19 @@
 
 #include <stdint.h>
 
-#include "amqp.h"
-#include "amqp_framing.h"
-#include "amqp_tcp_socket.h"
-#include "amqp_ssl_socket.h"
+#include <rabbitmq-c/amqp.h>
+#include <rabbitmq-c/framing.h>
+#include <rabbitmq-c/tcp_socket.h>
+#include <rabbitmq-c/ssl_socket.h>
 
 #include <iostream>
 #include <vector>
 #include <map>
 #include <memory>
 #include <exception>
+#include <variant>
 
-#if __cplusplus > 199711L // C++11 or greater
+#if __cplusplus > 199711L || (defined(_MSC_VER) && _MSC_VER >= 1800) // C++11 or greater
 #include <functional>
 #endif
 //export AMQP;
@@ -94,10 +95,10 @@ class AMQPException : public std::exception {
 class AMQPMessage {
 
 	char * data;
-	uint32_t len;
+	size_t len;
 	std::string exchange;
 	std::string routing_key;
-	uint32_t delivery_tag;
+	uint64_t delivery_tag;
 	int message_count;
 	std::string consumer_tag;
 	AMQPQueue * queue;
@@ -107,14 +108,14 @@ class AMQPMessage {
 		AMQPMessage(AMQPQueue * queue);
 		~AMQPMessage();
 
-		void setMessage(const char * data,uint32_t length);
-		char * getMessage(uint32_t* length);
+		void setMessage(const char * data, size_t length);
+		char * getMessage(size_t* length);
 
 		void addHeader(std::string name, amqp_bytes_t * value);
 		void addHeader(std::string name, uint64_t * value);
 		void addHeader(std::string name, uint8_t * value);
 		void addHeader(amqp_bytes_t * name, amqp_bytes_t * value);
-		std::string getHeader(std::string name);
+		std::string getHeader(std::string name) const;
 
 		void setConsumerTag( amqp_bytes_t consumer_tag);
 		void setConsumerTag( std::string consumer_tag);
@@ -131,8 +132,8 @@ class AMQPMessage {
 		void setRoutingKey(std::string routing_key);
 		std::string getRoutingKey();
 
-		uint32_t getDeliveryTag();
-		void setDeliveryTag(uint32_t delivery_tag);
+		uint64_t getDeliveryTag();
+		void setDeliveryTag(uint64_t delivery_tag);
 
 		AMQPQueue * getQueue();
 };
@@ -144,7 +145,7 @@ class AMQPBase {
 		short parms;
 		amqp_connection_state_t * cnn;
 		int channelNum;
-		AMQPMessage * pmessage;
+		std::unique_ptr<AMQPMessage> pmessage;
 
 		short opened;
 
@@ -166,21 +167,27 @@ class AMQPBase {
 
 class AMQPQueue : public AMQPBase  {
 	protected:
-#if __cplusplus > 199711L // C++11 or greater
+#if __cplusplus > 199711L || (defined(_MSC_VER) && _MSC_VER >= 1800) // C++11 or greater
                 std::map< AMQPEvents_e, std::function<int(AMQPMessage*)> > events;
 #else
 		std::map< AMQPEvents_e, int(*)( AMQPMessage * ) > events;
 #endif
-		amqp_bytes_t consumer_tag;
-		uint32_t delivery_tag;
+		std::string consumer_tag;
+		uint64_t delivery_tag;
 		uint32_t count;
 	public:
 		AMQPQueue(amqp_connection_state_t * cnn, int channelNum);
 		AMQPQueue(amqp_connection_state_t * cnn, int channelNum, std::string name);
 
+		struct KeyValuePair
+		{
+			std::string key;
+			std::variant<std::string, int32_t> value;
+		};
+
 		void Declare();
 		void Declare(std::string name);
-		void Declare(std::string name, short parms);
+		void Declare(std::string name, short parms, const std::vector<KeyValuePair>& arguments = std::vector<KeyValuePair>());
 
 		void Delete();
 		void Delete(std::string name);
@@ -202,28 +209,31 @@ class AMQPQueue : public AMQPBase  {
 		void Cancel(std::string consumer_tag);
 
 		void Ack();
-		void Ack(uint32_t delivery_tag);
+		void Ack(uint64_t delivery_tag);
+
+		void Reject(bool requeue);
+		void Reject(uint64_t delivery_tag, bool requeue);
 
 		AMQPMessage * getMessage() {
-			return pmessage;
+			return pmessage.get();
 		}
 
 		uint32_t getCount() {
 			return count;
 		}
 
 		void setConsumerTag(std::string consumer_tag);
-		amqp_bytes_t getConsumerTag();
+		std::string getConsumerTag();
 
 		void addEvent( AMQPEvents_e eventType, int (*event)(AMQPMessage*) );
-#if __cplusplus > 199711L // C++11 or greater
+#if __cplusplus > 199711L || (defined(_MSC_VER) && _MSC_VER >= 1800) // C++11 or greater
                 void addEvent( AMQPEvents_e eventType, std::function<int(AMQPMessage*)>& event );
 #endif
 		virtual ~AMQPQueue();
 
 		void Qos(uint32_t prefetch_size, uint16_t prefetch_count, amqp_boolean_t global );
 	private:
-		void sendDeclareCommand();
+		void sendDeclareCommand(const std::vector<KeyValuePair>& arguments = std::vector<KeyValuePair>());
 		void sendDeleteCommand();
 		void sendPurgeCommand();
 		void sendBindCommand(const char * exchange, const char * key);
@@ -232,6 +242,7 @@ class AMQPQueue : public AMQPBase  {
 		void sendConsumeCommand();
 		void sendCancelCommand();
 		void sendAckCommand();
+		void sendRejectCommand(bool requeue);
 		void setHeaders(amqp_basic_properties_t * p);
 };
 
@@ -259,7 +270,7 @@ class AMQPExchange : public AMQPBase {
 		void Bind(std::string queueName, std::string key);
 
 		void Publish(std::string message, std::string key);
-		void Publish(char * data, uint32_t length, std::string key);
+		void Publish(char * data, size_t length, std::string key);
 
 		void setHeader(std::string name, int value);
 		void setHeader(std::string name, std::string value);
@@ -285,7 +296,7 @@ class AMQP {
 		AMQP();
 		AMQP(std::string cnnStr, bool use_ssl_=false,
 				std::string cacert_path_="", std::string client_cert_path_="", std::string client_key_path_="",
-				bool verify_peer_=false, bool verify_hostname_=false);
+				bool verify_peer_=false, bool verify_hostname_=false, int heartbeat = 60);
 		~AMQP();
 
 		AMQPExchange * createExchange();
@@ -298,15 +309,19 @@ class AMQP {
 
 		void closeChannel();
 
+        amqp_connection_state_t getConnectionState() {
+            return cnn;
+        }
+
 	private:
 		void init(enum AMQPProto_e proto);
 		void initDefault(enum AMQPProto_e proto);
-		void connect();
+		void connect(int heartbeat = 60);
 		void parseCnnString(std::string cnnString );
 		void parseHostPort(std::string hostPortString );
 		void parseUserStr(std::string userString );
 		void sockConnect();
-		void login();
+		void login(int heartbeat);
 
 
 

src/AMQP.cpp

@@ -16,11 +16,11 @@ AMQP::AMQP() {
 	AMQP::init(proto);
 	AMQP::initDefault(proto);
 	AMQP::connect();
-};
+}
 
 AMQP::AMQP(string cnnStr, bool use_ssl_,
 		string cacert_path_, string client_cert_path_, string client_key_path_,
-		bool verify_peer_, bool verify_hostname_) {
+		bool verify_peer_, bool verify_hostname_, int heartbeat) {
 	use_ssl = use_ssl_;
 	proto = SET_AMQP_PROTO_BY_SSL_USAGE(use_ssl);
 	cacert_path = cacert_path_;
@@ -31,8 +31,8 @@ AMQP::AMQP(string cnnStr, bool use_ssl_,
 
 	AMQP::init(proto);
 	AMQP::parseCnnString(cnnStr);
-	AMQP::connect();
-};
+	AMQP::connect(heartbeat);
+}
 
 AMQP::~AMQP() {
 	if (channels.size()) {
@@ -44,7 +44,7 @@ AMQP::~AMQP() {
 
 	amqp_connection_close(cnn, AMQP_REPLY_SUCCESS);
 	amqp_destroy_connection(cnn);
-};
+}
 
 void AMQP::init(enum AMQPProto_e proto) {
 	switch(proto) {
@@ -158,9 +158,9 @@ void AMQP::parseHostPort(string hostPortString ) {
 	}
 }
 
-void AMQP::connect() {
+void AMQP::connect(int heartbeat) {
 	AMQP::sockConnect();
-	AMQP::login();
+	AMQP::login(heartbeat);
 }
 
 void AMQP::printConnect() {
@@ -179,6 +179,9 @@ void AMQP::sockConnect() {
 
 	switch(proto) {
 		case AMQPS_proto: {
+#ifdef AMQP_NO_SSL
+			throw std::runtime_error("AMQPcpp built with no SSL support");
+#else // AMQP_NO_SSL
 			sockfd = amqp_ssl_socket_new(cnn);
 
 			status = amqp_ssl_socket_set_cacert(sockfd, cacert_path.c_str());
@@ -197,6 +200,7 @@ void AMQP::sockConnect() {
 			amqp_ssl_socket_set_verify_peer(sockfd, verify_peer ? 1 : 0);
 			amqp_ssl_socket_set_verify_hostname(sockfd, verify_hostname ? 1 : 0);
 #endif
+#endif // AMQP_NO_SSL
 		}
 		break;
 
@@ -215,8 +219,8 @@ void AMQP::sockConnect() {
 	}
 }
 
-void AMQP::login() {
-	amqp_rpc_reply_t res = amqp_login(cnn, vhost.c_str(), 0, FRAME_MAX, 0, AMQP_SASL_METHOD_PLAIN, user.c_str(), password.c_str());
+void AMQP::login(int heartbeat) {
+	amqp_rpc_reply_t res = amqp_login(cnn, vhost.c_str(), 0, FRAME_MAX, heartbeat, AMQP_SASL_METHOD_PLAIN, user.c_str(), password.c_str());
 	if ( res.reply_type != AMQP_RESPONSE_NORMAL) {
 		const AMQPException exception(&res);
 		amqp_destroy_connection(cnn);

src/AMQPException.cpp

@@ -21,7 +21,15 @@ AMQPException::AMQPException(string action, int error_code)
 
 AMQPException::AMQPException( amqp_rpc_reply_t * res) {
 	if( res->reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION) {
-		this->message = res->library_error ? strerror(res->library_error) : "end-of-stream";
+        if (res->library_error) {
+            switch (res->library_error) {
+            case AMQP_STATUS_SOCKET_ERROR: this->message = "AMQP socket error"; break;
+            default: this->message = "AMQP error " + std::to_string(res->library_error) + " (see amqp.h to figure out what this means)"; break;
+            }
+        }
+        else {
+            this->message = "end-of-stream";
+        }
 	}
 
 	if( res->reply_type == AMQP_RESPONSE_SERVER_EXCEPTION) {