Extend API to permit heartbeat configuration.

11 files changed, 26 insertions, 13 deletions

diff --git a/examples/amqp_bind.c b/examples/amqp_bind.c
index 1d3055b..afbfed9 100644
--- a/examples/amqp_bind.c
+++ b/examples/amqp_bind.c
@@ -35,7 +35,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_consumer.c b/examples/amqp_consumer.c
index 3dd3262..230a94e 100644
--- a/examples/amqp_consumer.c
+++ b/examples/amqp_consumer.c
@@ -104,7 +104,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_exchange_declare.c b/examples/amqp_exchange_declare.c
index 0082d62..02884e8 100644
--- a/examples/amqp_exchange_declare.c
+++ b/examples/amqp_exchange_declare.c
@@ -33,7 +33,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_listen.c b/examples/amqp_listen.c
index dc0f00c..d39e4b4 100644
--- a/examples/amqp_listen.c
+++ b/examples/amqp_listen.c
@@ -40,7 +40,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_producer.c b/examples/amqp_producer.c
index 106997d..a60cb3e 100644
--- a/examples/amqp_producer.c
+++ b/examples/amqp_producer.c
@@ -92,7 +92,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_sendstring.c b/examples/amqp_sendstring.c
index 305811b..a321f52 100644
--- a/examples/amqp_sendstring.c
+++ b/examples/amqp_sendstring.c
@@ -35,7 +35,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/examples/amqp_unbind.c b/examples/amqp_unbind.c
index 0677203..9a9a35c 100644
--- a/examples/amqp_unbind.c
+++ b/examples/amqp_unbind.c
@@ -35,7 +35,7 @@ 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, "/", 0, 131072, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"),
 		    "Logging in");
   amqp_channel_open(conn, 1);
   die_on_amqp_error(amqp_rpc_reply, "Opening channel");
diff --git a/librabbitmq/amqp.h b/librabbitmq/amqp.h
index 563c429..5101364 100644
--- a/librabbitmq/amqp.h
+++ b/librabbitmq/amqp.h
@@ -142,7 +142,8 @@ extern void amqp_set_sockfd(amqp_connection_state_t state,
 			    int sockfd);
 extern int amqp_tune_connection(amqp_connection_state_t state,
 				int channel_max,
-				int frame_max);
+				int frame_max,
+				int heartbeat);
 int amqp_get_channel_max(amqp_connection_state_t state);
 extern void amqp_destroy_connection(amqp_connection_state_t state);
 
@@ -206,6 +207,7 @@ extern amqp_rpc_reply_t amqp_login(amqp_connection_state_t state,
 				   char const *vhost,
 				   int channel_max,
 				   int frame_max,
+				   int heartbeat,
 				   amqp_sasl_method_enum sasl_method, ...);
 
 extern amqp_rpc_reply_t amqp_rpc_reply;
diff --git a/librabbitmq/amqp_connection.c b/librabbitmq/amqp_connection.c
index 9653694..d01af66 100644
--- a/librabbitmq/amqp_connection.c
+++ b/librabbitmq/amqp_connection.c
@@ -42,7 +42,7 @@ amqp_connection_state_t amqp_new_connection(void) {
 
   state->inbound_buffer.bytes = NULL;
   state->outbound_buffer.bytes = NULL;
-  if (amqp_tune_connection(state, 0, INITIAL_FRAME_POOL_PAGE_SIZE) != 0) {
+  if (amqp_tune_connection(state, 0, INITIAL_FRAME_POOL_PAGE_SIZE, 0) != 0) {
     empty_amqp_pool(&state->frame_pool);
     empty_amqp_pool(&state->decoding_pool);
     free(state);
@@ -81,7 +81,8 @@ void amqp_set_sockfd(amqp_connection_state_t state,
 
 int amqp_tune_connection(amqp_connection_state_t state,
 			 int channel_max,
-			 int frame_max)
+			 int frame_max,
+			 int heartbeat)
 {
   void *newbuf;
 
@@ -89,6 +90,7 @@ int amqp_tune_connection(amqp_connection_state_t state,
 
   state->channel_max = channel_max;
   state->frame_max = frame_max;
+  state->heartbeat = heartbeat;
 
   empty_amqp_pool(&state->frame_pool);
   init_amqp_pool(&state->frame_pool, frame_max);
diff --git a/librabbitmq/amqp_private.h b/librabbitmq/amqp_private.h
index 0f9987a..98a7a65 100644
--- a/librabbitmq/amqp_private.h
+++ b/librabbitmq/amqp_private.h
@@ -56,6 +56,7 @@ struct amqp_connection_state_t_ {
 
   int channel_max;
   int frame_max;
+  int heartbeat;
   amqp_bytes_t inbound_buffer;
 
   size_t inbound_offset;
diff --git a/librabbitmq/amqp_socket.c b/librabbitmq/amqp_socket.c
index 04e8c32..d29dcd6 100644
--- a/librabbitmq/amqp_socket.c
+++ b/librabbitmq/amqp_socket.c
@@ -277,12 +277,14 @@ amqp_rpc_reply_t amqp_simple_rpc(amqp_connection_state_t state,
 static int amqp_login_inner(amqp_connection_state_t state,
 			    int channel_max,
 			    int frame_max,
+			    int heartbeat,
 			    amqp_sasl_method_enum sasl_method,
 			    va_list vl)
 {
   amqp_method_t method;
   uint32_t server_frame_max;
   uint16_t server_channel_max;
+  uint16_t server_heartbeat;
 
   amqp_send_header(state);
 
@@ -318,6 +320,7 @@ static int amqp_login_inner(amqp_connection_state_t state,
     amqp_connection_tune_t *s = (amqp_connection_tune_t *) method.decoded;
     server_channel_max = s->channel_max;
     server_frame_max = s->frame_max;
+    server_heartbeat = s->heartbeat;
   }
 
   if (server_channel_max != 0 && server_channel_max < channel_max) {
@@ -328,14 +331,18 @@ static int amqp_login_inner(amqp_connection_state_t state,
     frame_max = server_frame_max;
   }
 
-  AMQP_CHECK_RESULT(amqp_tune_connection(state, channel_max, frame_max));
+  if (server_heartbeat != 0 && server_heartbeat < heartbeat) {
+    heartbeat = server_heartbeat;
+  }
+
+  AMQP_CHECK_RESULT(amqp_tune_connection(state, channel_max, frame_max, heartbeat));
 
   {
     amqp_connection_tune_ok_t s =
       (amqp_connection_tune_ok_t) {
         .channel_max = channel_max,
 	.frame_max = frame_max,
-	.heartbeat = 0
+	.heartbeat = heartbeat
       };
     AMQP_CHECK_RESULT(amqp_send_method(state, 0, AMQP_CONNECTION_TUNE_OK_METHOD, &s));
   }
@@ -349,6 +356,7 @@ amqp_rpc_reply_t amqp_login(amqp_connection_state_t state,
 			    char const *vhost,
 			    int channel_max,
 			    int frame_max,
+			    int heartbeat,
 			    amqp_sasl_method_enum sasl_method,
 			    ...)
 {
@@ -357,7 +365,7 @@ amqp_rpc_reply_t amqp_login(amqp_connection_state_t state,
 
   va_start(vl, sasl_method);
 
-  amqp_login_inner(state, channel_max, frame_max, sasl_method, vl);
+  amqp_login_inner(state, channel_max, frame_max, heartbeat, sasl_method, vl);
 
   {
     amqp_connection_open_t s =