Support for multiple channels.

 - amqp_login() no longer does amqp_channel_open() for you
 - amqp_login() takes a channel_max argument
 - amqp_login() actually calls amqp_tune_connection() now
 - amqp_channel_close() and amqp_basic_publish() now take a channel number

5 files changed, 22 insertions, 10 deletions

diff --git a/examples/amqp_consumer.c b/examples/amqp_consumer.c
index 2ce6707..3dd3262 100644
--- a/examples/amqp_consumer.c
+++ b/examples/amqp_consumer.c
@@ -104,8 +104,10 @@ int main(int argc, char const * const *argv) {
 
   die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
   amqp_set_sockfd(conn, sockfd);
-  die_on_amqp_error(amqp_login(conn, "/", 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
+  amqp_channel_open(conn, 1);
+  die_on_amqp_error(amqp_rpc_reply, "Opening channel");
 
   {
     amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, AMQP_EMPTY_BYTES, 0, 0, 0, 1,
@@ -126,7 +128,7 @@ int main(int argc, char const * const *argv) {
 
   run(conn);
 
-  die_on_amqp_error(amqp_channel_close(conn, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   amqp_destroy_connection(conn);
   die_on_error(close(sockfd), "Closing socket");
diff --git a/examples/amqp_exchange_declare.c b/examples/amqp_exchange_declare.c
index f163db1..0082d62 100644
--- a/examples/amqp_exchange_declare.c
+++ b/examples/amqp_exchange_declare.c
@@ -33,14 +33,16 @@ int main(int argc, char const * const *argv) {
 
   die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
   amqp_set_sockfd(conn, sockfd);
-  die_on_amqp_error(amqp_login(conn, "/", 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
+  amqp_channel_open(conn, 1);
+  die_on_amqp_error(amqp_rpc_reply, "Opening channel");
 
   amqp_exchange_declare(conn, 1, amqp_cstring_bytes(exchange), amqp_cstring_bytes(exchangetype),
 			0, 0, 0, AMQP_EMPTY_TABLE);
   die_on_amqp_error(amqp_rpc_reply, "Declaring exchange");
 
-  die_on_amqp_error(amqp_channel_close(conn, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   amqp_destroy_connection(conn);
   die_on_error(close(sockfd), "Closing socket");
diff --git a/examples/amqp_listen.c b/examples/amqp_listen.c
index 975d49f..dc0f00c 100644
--- a/examples/amqp_listen.c
+++ b/examples/amqp_listen.c
@@ -40,8 +40,10 @@ int main(int argc, char const * const *argv) {
 
   die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
   amqp_set_sockfd(conn, sockfd);
-  die_on_amqp_error(amqp_login(conn, "/", 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
+  amqp_channel_open(conn, 1);
+  die_on_amqp_error(amqp_rpc_reply, "Opening channel");
 
   {
     amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, AMQP_EMPTY_BYTES, 0, 0, 0, 1,
@@ -133,7 +135,7 @@ int main(int argc, char const * const *argv) {
     }
   }
 
-  die_on_amqp_error(amqp_channel_close(conn, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   amqp_destroy_connection(conn);
   die_on_error(close(sockfd), "Closing socket");
diff --git a/examples/amqp_producer.c b/examples/amqp_producer.c
index 89e56bf..106997d 100644
--- a/examples/amqp_producer.c
+++ b/examples/amqp_producer.c
@@ -33,6 +33,7 @@ static void send_batch(amqp_connection_state_t conn,
   for (i = 0; i < message_count; i++) {
     long long now = now_microseconds();
     die_on_error(amqp_basic_publish(conn,
+				    1,
 				    amqp_cstring_bytes("amq.direct"),
 				    amqp_cstring_bytes(queue_name),
 				    0,
@@ -91,12 +92,14 @@ int main(int argc, char const * const *argv) {
 
   die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
   amqp_set_sockfd(conn, sockfd);
-  die_on_amqp_error(amqp_login(conn, "/", 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
+  amqp_channel_open(conn, 1);
+  die_on_amqp_error(amqp_rpc_reply, "Opening channel");
 
   send_batch(conn, "test queue", rate_limit, message_count);
 
-  die_on_amqp_error(amqp_channel_close(conn, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   amqp_destroy_connection(conn);
   die_on_error(close(sockfd), "Closing socket");
diff --git a/examples/amqp_sendstring.c b/examples/amqp_sendstring.c
index c914f86..a26cd3a 100644
--- a/examples/amqp_sendstring.c
+++ b/examples/amqp_sendstring.c
@@ -35,14 +35,17 @@ int main(int argc, char const * const *argv) {
 
   die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
   amqp_set_sockfd(conn, sockfd);
-  die_on_amqp_error(amqp_login(conn, "/", 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
+  amqp_channel_open(conn, 1);
+  die_on_amqp_error(amqp_rpc_reply, "Opening channel");
 
   {
     amqp_basic_properties_t props;
     props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG;
     props.content_type = amqp_cstring_bytes("text/plain");
     die_on_error(amqp_basic_publish(conn,
+				    1,
 				    amqp_cstring_bytes(exchange),
 				    amqp_cstring_bytes(routingkey),
 				    0,
@@ -52,7 +55,7 @@ int main(int argc, char const * const *argv) {
 		 "Publishing");
   }
 
-  die_on_amqp_error(amqp_channel_close(conn, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   amqp_destroy_connection(conn);
   die_on_error(close(sockfd), "Closing socket");