core/src/connect_handler.c

   1 #include "connect_handler.h"
   2 #include "request.h"
   3 #include "core.h"
   4
   5 #include <memory.h>
   6 #include <malloc.h> /* calloc */
   7 #include <errno.h>  /* errno */
   8 #include <ev.h>
   9
  10 struct connect_request {
  11         request_t request;
  12         union {
  13                 ev_io file_watcher;
  14                 ev_timer time_watcher;
  15         };
  16         int fd;
  17         int domain, type, protocol;
  18         struct sockaddr *addr;
  19         socklen_t addr_len;
  20         char timeout_iterations;
  21 };
  22
  23 #define DEFAULT_TIMEOUT_ITERATIONS 10
  24 #define DEFAULT_TIMEOUT_INTERVAL (0.25)
  25
  26 connect_handler_t *new_connect_handler(core_t *c) {
  27         connect_handler_t *h = calloc(1, sizeof(connect_handler_t));
  28         if(!h) goto fail;
  29         h->core = c;
  30         return h;
  31 fail:
  32         free_connect_handler(h);
  33         return NULL;
  34 }
  35
  36 void free_connect_handler(connect_handler_t *h) {
  37         if(!h) return;
  38         free(h);
  39 }
  40
  41 connect_request_t *new_connect_request(connect_handler_t *h, const struct sockaddr *addr, socklen_t addr_len) {
  42         connect_request_t *cr = calloc(1, sizeof(connect_request_t));
  43         if(!cr) goto fail;
  44         cr->addr = malloc(addr_len);
  45         if(!cr->addr) goto fail;
  46         memcpy(cr->addr, addr, addr_len);
  47         cr->addr_len = addr_len;
  48         cr->timeout_iterations = DEFAULT_TIMEOUT_ITERATIONS;
  49         return cr;
  50 fail:
  51         free_connect_request(cr);
  52         return NULL;
  53 }
  54
  55 void free_connect_request(connect_request_t *cr) {
  56         if(!cr) return;
  57         if(cr->addr) free(cr->addr);
  58         free(cr);
  59 }
  60
  61 void connect_request_set_socket(connect_request_t *cr, int fd) {
  62         cr->fd = fd;
  63 }
  64
  65 int connect_request_get_socket(connect_request_t *cr) {
  66         return cr->fd;
  67 }
  68
  69 void connect_request_set_create_params(connect_request_t *cr, int domain, int type, int protocol) {
  70         cr->fd = -1;
  71         cr->domain = domain;
  72         cr->type = type;
  73         cr->protocol = protocol;
  74 }
  75
  76 static void connect_cb(int revents, void* arg) {
  77         connect_request_t *cr = arg;
  78         int fd = cr->fd;
  79         if(revents & EV_TIMEOUT) {
  80                 /* TODO: timeout */
  81                 return;
  82         }
  83         if(!(revents & EV_WRITE)) /* TODO: error ? */
  84                 return;
  85         /* Check for connection error */
  86         int err;
  87         socklen_t errlen = sizeof(err);
  88         getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen);
  89         if(err == 0) {
  90                 cr->request.on_complete((request_t*)cr);
  91                 return;
  92         }
  93         errno = err;
  94         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
  95 }
  96
  97 static void setup_connect(struct ev_loop *loop, connect_request_t *cr) {
  98 tryagain:
  99         if(0 != connect(cr->fd, cr->addr, cr->addr_len)) {
 100                 if(errno == EINTR) goto tryagain;
 101                 if(errno == EINPROGRESS) {
 102                         ev_once(loop, cr->fd, EV_WRITE, -1., connect_cb, cr);
 103                 } else {
 104                         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
 105                 }
 106         } else {
 107                 cr->request.on_complete((request_t*)cr);
 108         }
 109 }
 110
 111 static void setup_socket_cb(struct ev_loop *loop, struct ev_timer *w, int revents);
 112
 113 static void setup_socket(struct ev_loop *loop, connect_request_t *cr, int firstcall) {
 114         cr->fd = socket(cr->domain, cr->type, cr->protocol);
 115         if(cr->fd == -1) {
 116                 if(errno == ENFILE) { /* no files remaining - should retry */
 117                         if(!firstcall) return;
 118                         ev_timer_init(&cr->time_watcher, setup_socket_cb, 0., DEFAULT_TIMEOUT_INTERVAL);
 119                         cr->time_watcher.data = cr;
 120                         ev_timer_start(loop, &cr->time_watcher);
 121                 } else { /* another error occurred... */
 122                         if(!firstcall)
 123                                 ev_timer_stop(loop, &cr->time_watcher);
 124                         cr->request.on_error((request_t*)cr, IOR_ERROR_ERRNO);
 125                         return;
 126                 }
 127         }
 128         /* begin connection process.. */
 129         if(!firstcall)
 130                 ev_timer_stop(loop, &cr->time_watcher);
 131         setup_connect(loop, cr);
 132 }
 133
 134 static void setup_socket_cb(struct ev_loop *loop, struct ev_timer *w, int revents) {
 135         connect_request_t *cr = w->data;
 136         cr->timeout_iterations --;
 137         if(cr->timeout_iterations <= 0) {
 138                 ev_timer_stop(loop, w);
 139         }
 140         setup_socket(loop, cr, 0);
 141 }
 142
 143 void connect_handler_queue(connect_handler_t *h, connect_request_t *cr) {
 144         if(cr->fd == -1) {
 145                 setup_socket(h->core->loop, cr, 1);
 146                 return;
 147         }
 148         setup_connect(h->core->loop, cr);
 149 }
 150