init

librabbitmq/amqp_tcp_socket.c

@@ -0,0 +1,219 @@
+// Copyright 2007 - 2021, Alan Antonuk and the rabbitmq-c contributors.
+// SPDX-License-Identifier: mit
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include "amqp_private.h"
+#include "rabbitmq-c/tcp_socket.h"
+
+#include <errno.h>
+#if ((defined(_WIN32)) || (defined(__MINGW32__)) || (defined(__MINGW64__)))
+#ifndef WIN32_LEAN_AND_MEAN
+#define WIN32_LEAN_AND_MEAN
+#endif
+#include <winsock2.h>
+#else
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <sys/socket.h>
+#endif
+#include <stdio.h>
+#include <stdlib.h>
+
+struct amqp_tcp_socket_t {
+  const struct amqp_socket_class_t *klass;
+  int sockfd;
+  int internal_error;
+  int state;
+};
+
+static ssize_t amqp_tcp_socket_send(void *base, const void *buf, size_t len,
+                                    int flags) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+  ssize_t res;
+  int flagz = 0;
+
+  if (-1 == self->sockfd) {
+    return AMQP_STATUS_SOCKET_CLOSED;
+  }
+
+#ifdef MSG_NOSIGNAL
+  flagz |= MSG_NOSIGNAL;
+#endif
+
+#if defined(MSG_MORE)
+  if (flags & AMQP_SF_MORE) {
+    flagz |= MSG_MORE;
+  }
+/* Cygwin defines TCP_NOPUSH, but trying to use it will return not
+ * implemented. Disable it here. */
+#elif defined(TCP_NOPUSH) && !defined(__CYGWIN__)
+  if (flags & AMQP_SF_MORE && !(self->state & AMQP_SF_MORE)) {
+    int one = 1;
+    res = setsockopt(self->sockfd, IPPROTO_TCP, TCP_NOPUSH, &one, sizeof(one));
+    if (0 != res) {
+      self->internal_error = res;
+      return AMQP_STATUS_SOCKET_ERROR;
+    }
+    self->state |= AMQP_SF_MORE;
+  } else if (!(flags & AMQP_SF_MORE) && self->state & AMQP_SF_MORE) {
+    int zero = 0;
+    res =
+        setsockopt(self->sockfd, IPPROTO_TCP, TCP_NOPUSH, &zero, sizeof(&zero));
+    if (0 != res) {
+      self->internal_error = res;
+      res = AMQP_STATUS_SOCKET_ERROR;
+    } else {
+      self->state &= ~AMQP_SF_MORE;
+    }
+  }
+#endif
+
+start:
+#ifdef _WIN32
+  res = send(self->sockfd, buf, (int)len, flagz);
+#else
+  res = send(self->sockfd, buf, len, flagz);
+#endif
+
+  if (res < 0) {
+    self->internal_error = amqp_os_socket_error();
+    switch (self->internal_error) {
+      case EINTR:
+        goto start;
+#ifdef _WIN32
+      case WSAEWOULDBLOCK:
+#else
+      case EWOULDBLOCK:
+#endif
+#if defined(EAGAIN) && EAGAIN != EWOULDBLOCK
+      case EAGAIN:
+#endif
+        res = AMQP_PRIVATE_STATUS_SOCKET_NEEDWRITE;
+        break;
+      default:
+        res = AMQP_STATUS_SOCKET_ERROR;
+    }
+  } else {
+    self->internal_error = 0;
+  }
+
+  return res;
+}
+
+static ssize_t amqp_tcp_socket_recv(void *base, void *buf, size_t len,
+                                    int flags) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+  ssize_t ret;
+  if (-1 == self->sockfd) {
+    return AMQP_STATUS_SOCKET_CLOSED;
+  }
+
+start:
+#ifdef _WIN32
+  ret = recv(self->sockfd, buf, (int)len, flags);
+#else
+  ret = recv(self->sockfd, buf, len, flags);
+#endif
+
+  if (0 > ret) {
+    self->internal_error = amqp_os_socket_error();
+    switch (self->internal_error) {
+      case EINTR:
+        goto start;
+#ifdef _WIN32
+      case WSAEWOULDBLOCK:
+#else
+      case EWOULDBLOCK:
+#endif
+#if defined(EAGAIN) && EAGAIN != EWOULDBLOCK
+      case EAGAIN:
+#endif
+        ret = AMQP_PRIVATE_STATUS_SOCKET_NEEDREAD;
+        break;
+      default:
+        ret = AMQP_STATUS_SOCKET_ERROR;
+    }
+  } else if (0 == ret) {
+    ret = AMQP_STATUS_CONNECTION_CLOSED;
+  }
+
+  return ret;
+}
+
+static int amqp_tcp_socket_open(void *base, const char *host, int port,
+                                const struct timeval *timeout) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+  if (-1 != self->sockfd) {
+    return AMQP_STATUS_SOCKET_INUSE;
+  }
+  self->sockfd = amqp_open_socket_noblock(host, port, timeout);
+  if (0 > self->sockfd) {
+    int err = self->sockfd;
+    self->sockfd = -1;
+    return err;
+  }
+  return AMQP_STATUS_OK;
+}
+
+static int amqp_tcp_socket_close(void *base,
+                                 AMQP_UNUSED amqp_socket_close_enum force) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+  if (-1 == self->sockfd) {
+    return AMQP_STATUS_SOCKET_CLOSED;
+  }
+
+  if (amqp_os_socket_close(self->sockfd)) {
+    return AMQP_STATUS_SOCKET_ERROR;
+  }
+  self->sockfd = -1;
+
+  return AMQP_STATUS_OK;
+}
+
+static int amqp_tcp_socket_get_sockfd(void *base) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+  return self->sockfd;
+}
+
+static void amqp_tcp_socket_delete(void *base) {
+  struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+
+  if (self) {
+    amqp_tcp_socket_close(self, AMQP_SC_NONE);
+    free(self);
+  }
+}
+
+static const struct amqp_socket_class_t amqp_tcp_socket_class = {
+    amqp_tcp_socket_send,       /* send */
+    amqp_tcp_socket_recv,       /* recv */
+    amqp_tcp_socket_open,       /* open */
+    amqp_tcp_socket_close,      /* close */
+    amqp_tcp_socket_get_sockfd, /* get_sockfd */
+    amqp_tcp_socket_delete      /* delete */
+};
+
+amqp_socket_t *amqp_tcp_socket_new(amqp_connection_state_t state) {
+  struct amqp_tcp_socket_t *self = calloc(1, sizeof(*self));
+  if (!self) {
+    return NULL;
+  }
+  self->klass = &amqp_tcp_socket_class;
+  self->sockfd = -1;
+
+  amqp_set_socket(state, (amqp_socket_t *)self);
+
+  return (amqp_socket_t *)self;
+}
+
+void amqp_tcp_socket_set_sockfd(amqp_socket_t *base, int sockfd) {
+  struct amqp_tcp_socket_t *self;
+  if (base->klass != &amqp_tcp_socket_class) {
+    amqp_abort("<%p> is not of type amqp_tcp_socket_t", base);
+  }
+  self = (struct amqp_tcp_socket_t *)base;
+  self->sockfd = sockfd;
+}