#120 add second channel, and use it for queue declarations

Author: petersilva

sr_consume.c

@@ -86,6 +86,7 @@ int sr_consume_setup(struct sr_context *sr_c)
 	amqp_boolean_t passive = 0;
 	amqp_boolean_t exclusive = 0;
 	amqp_boolean_t auto_delete = 0;
+	amqp_queue_declare_ok_t *r;
 	struct sr_binding_s *t;
 	static amqp_basic_properties_t props;
 	static amqp_table_t table;
@@ -121,12 +122,16 @@ int sr_consume_setup(struct sr_context *sr_c)
 
 	//amqp_queue_declare_ok_t *r = 
 	if (sr_c->cfg->queueDeclare) {
-		amqp_queue_declare(sr_c->cfg->broker->conn,
-				   1,
+		r = amqp_queue_declare(sr_c->cfg->broker->conn,
+				   2,
 				   amqp_cstring_bytes(sr_c->cfg->queuename),
 				   passive, sr_c->cfg->durable, exclusive, auto_delete, table);
 		/* FIXME how to parse r for error? */
-
+                
+		if (r) {
+	               sr_log_msg(LOG_INFO, "queue declared: %p messages in queue: %d\n", 
+		                sr_c->cfg->queuename, r->message_count );
+                }
 		reply = amqp_get_rpc_reply(sr_c->cfg->broker->conn);
 		if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
 			sr_amqp_reply_print(reply, "queue declare failed");

sr_context.c

@@ -166,7 +166,13 @@ struct sr_broker_s *sr_broker_connect(struct sr_broker_s *broker)
 
 		open_status = amqp_channel_open(broker->conn, 1);
 		if (open_status == NULL) {
-			sr_log_msg(LOG_ERROR, "failed AMQP amqp_channel_open\n");
+			sr_log_msg(LOG_ERROR, "failed AMQP amqp_channel_open 1\n");
+			goto have_channel;
+		}
+
+		open_status = amqp_channel_open(broker->conn, 2);
+		if (open_status == NULL) {
+			sr_log_msg(LOG_ERROR, "failed AMQP amqp_channel_open 2\n");
 			goto have_channel;
 		}
 
@@ -192,6 +198,7 @@ struct sr_broker_s *sr_broker_connect(struct sr_broker_s *broker)
 
  have_channel:
 		reply = amqp_channel_close(broker->conn, 1, AMQP_REPLY_SUCCESS);
+		reply = amqp_channel_close(broker->conn, 2, AMQP_REPLY_SUCCESS);
 
  have_socket:
 		reply = amqp_connection_close(broker->conn, AMQP_REPLY_SUCCESS);
@@ -357,13 +364,18 @@ void sr_broker_close(struct sr_broker_s *broker)
 		//sr_log_msg(LOG_DEBUG, "amqp broker close: no connection present.\n");
 		return;
 	}
-	reply = amqp_channel_close(broker->conn, 1, AMQP_REPLY_SUCCESS);
+	reply = amqp_channel_close(broker->conn, 2, AMQP_REPLY_SUCCESS);
 	if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
-		sr_log_msg(LOG_ERROR, "amqp channel close failed.\n");
+		sr_log_msg(LOG_ERROR, "amqp channel close 2 failed.\n");
 	} else {
-		reply = amqp_connection_close(broker->conn, AMQP_REPLY_SUCCESS);
+		reply = amqp_channel_close(broker->conn, 1, AMQP_REPLY_SUCCESS);
 		if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
-			sr_log_msg(LOG_ERROR, "amqp connection close failed.\n");
+			sr_log_msg(LOG_ERROR, "amqp channel close 1 failed.\n");
+		} else {
+			reply = amqp_connection_close(broker->conn, AMQP_REPLY_SUCCESS);
+			if (reply.reply_type != AMQP_RESPONSE_NORMAL) {
+				sr_log_msg(LOG_ERROR, "amqp connection close failed.\n");
+			}
 		}
 	}
 

sr_context.h

@@ -57,7 +57,6 @@ struct sr_context {
 	const char *file;
 	const char *post_baseUrl;
 	amqp_socket_t *socket;
-	amqp_connection_state_t conn;
 	int port;
 	struct sr_config_s *cfg;
 	struct sr_metrics_s metrics;