Add support for heartbeats in amqp_basic_publish

Check heartbeats when doing basic.publish. Do this by doing a recv on
the socket.

1 files changed, 17 insertions, 0 deletions

diff --git a/librabbitmq/amqp_api.c b/librabbitmq/amqp_api.c
index bdd7d80..525d92c 100644
--- a/librabbitmq/amqp_api.c
+++ b/librabbitmq/amqp_api.c
@@ -39,6 +39,7 @@
 #endif
 
 #include "amqp_private.h"
+#include "amqp_timer.h"
 #include <assert.h>
 #include <stdarg.h>
 #include <stdint.h>
@@ -186,6 +187,22 @@ int amqp_basic_publish(amqp_connection_state_t state,
   m.immediate = immediate;
   m.ticket = 0;
 
+  if (state->heartbeat > 0) {
+    uint64_t current_timestamp = amqp_get_monotonic_timestamp();
+    if (0 == current_timestamp) {
+      return AMQP_STATUS_TIMER_FAILURE;
+    }
+
+    if (current_timestamp > state->next_recv_heartbeat) {
+      res = amqp_try_recv(state, current_timestamp);
+      if (AMQP_STATUS_TIMEOUT == res) {
+        return AMQP_STATUS_HEARTBEAT_TIMEOUT;
+      } else if (AMQP_STATUS_OK != res) {
+        return res;
+      }
+    }
+  }
+
   res = amqp_send_method(state, channel, AMQP_BASIC_PUBLISH_METHOD, &m);
   if (res < 0) {
     return res;