Function amqp_poll

Synopsis

#include <librabbitmq/amqp_socket.h>

int amqp_poll(int fd, int event, amqp_time_t deadline)

Description

No description yet.

Source

Lines 167-263 in librabbitmq/amqp_socket.c. Line 168 in librabbitmq/amqp_socket.h.

int amqp_poll(int fd, int event, amqp_time_t deadline) {
#ifdef HAVE_POLL
  struct pollfd pfd;
  int res;
  int timeout_ms;

  /* Function should only ever be called with one of these two */
  assert(event == AMQP_SF_POLLIN || event == AMQP_SF_POLLOUT);

start_poll:
  pfd.fd = fd;
  switch (event) {
    case AMQP_SF_POLLIN:
      pfd.events = POLLIN;
      break;
    case AMQP_SF_POLLOUT:
      pfd.events = POLLOUT;
      break;
  }

  timeout_ms = amqp_time_ms_until(deadline);
  if (-1 > timeout_ms) {
    return timeout_ms;
  }

  res = poll(&pfd, 1, timeout_ms);

  if (0 < res) {
    /* TODO: optimize this a bit by returning the AMQP_STATUS_SOCKET_ERROR or
     * equivalent when pdf.revent is POLLHUP or POLLERR, so an extra syscall
     * doesn't need to be made. */
    return AMQP_STATUS_OK;
  } else if (0 == res) {
    return AMQP_STATUS_TIMEOUT;
  } else {
    switch (amqp_os_socket_error()) {
      case EINTR:
        goto start_poll;
      default:
        return AMQP_STATUS_SOCKET_ERROR;
    }
  }
  return AMQP_STATUS_OK;
#elif defined(HAVE_SELECT)
  fd_set fds;
  fd_set exceptfds;
  fd_set *exceptfdsp;
  int res;
  struct timeval tv;
  struct timeval *tvp;

  assert((0 != (event & AMQP_SF_POLLIN)) || (0 != (event & AMQP_SF_POLLOUT)));
#ifndef _WIN32
  /* On Win32 connect() failure is indicated through the exceptfds, it does not
   * make any sense to allow POLLERR on any other platform or condition */
  assert(0 == (event & AMQP_SF_POLLERR));
#endif

start_select:
  FD_ZERO(&fds);
  FD_SET(fd, &fds);

  if (event & AMQP_SF_POLLERR) {
    FD_ZERO(&exceptfds);
    FD_SET(fd, &exceptfds);
    exceptfdsp = &exceptfds;
  } else {
    exceptfdsp = NULL;
  }

  res = amqp_time_tv_until(deadline, &tv, &tvp);
  if (res != AMQP_STATUS_OK) {
    return res;
  }

  if (event & AMQP_SF_POLLIN) {
    res = select(fd + 1, &fds, NULL, exceptfdsp, tvp);
  } else if (event & AMQP_SF_POLLOUT) {
    res = select(fd + 1, NULL, &fds, exceptfdsp, tvp);
  }

  if (0 < res) {
    return AMQP_STATUS_OK;
  } else if (0 == res) {
    return AMQP_STATUS_TIMEOUT;
  } else {
    switch (amqp_os_socket_error()) {
      case EINTR:
        goto start_select;
      default:
        return AMQP_STATUS_SOCKET_ERROR;
    }
  }
#else
#error "poll() or select() is needed to compile rabbitmq-c"
#endif
}





Add Discussion as Guest

Log in to DocsForge