docs/examples/evhiperfifo.c

   1 /*****************************************************************************
   2  *                                  _   _ ____  _
   3  *  Project                     ___| | | |  _ \| |
   4  *                             / __| | | | |_) | |
   5  *                            | (__| |_| |  _ <| |___
   6  *                             \___|\___/|_| \_\_____|
   7  *
   8  *
   9  * Example application source code using the multi socket interface to
  10  * download many files at once.
  11  *
  12  * This example features the same basic functionality as hiperfifo.c does,
  13  * but this uses libev instead of libevent.
  14  *
  15  * Written by Jeff Pohlmeyer, converted to use libev by Markus Koetter
  16
  17 Requires libev and a (POSIX?) system that has mkfifo().
  18
  19 This is an adaptation of libcurl's "hipev.c" and libevent's "event-test.c"
  20 sample programs.
  21
  22 When running, the program creates the named pipe "hiper.fifo"
  23
  24 Whenever there is input into the fifo, the program reads the input as a list
  25 of URL's and creates some new easy handles to fetch each URL via the
  26 curl_multi "hiper" API.
  27
  28
  29 Thus, you can try a single URL:
  30   % echo http://www.yahoo.com > hiper.fifo
  31
  32 Or a whole bunch of them:
  33   % cat my-url-list > hiper.fifo
  34
  35 The fifo buffer is handled almost instantly, so you can even add more URL's
  36 while the previous requests are still being downloaded.
  37
  38 Note:
  39   For the sake of simplicity, URL length is limited to 1023 char's !
  40
  41 This is purely a demo app, all retrieved data is simply discarded by the write
  42 callback.
  43
  44 */
  45
  46 #include <stdio.h>
  47 #include <string.h>
  48 #include <stdlib.h>
  49 #include <sys/time.h>
  50 #include <time.h>
  51 #include <unistd.h>
  52 #include <sys/poll.h>
  53 #include <curl/curl.h>
  54 #include <ev.h>
  55 #include <fcntl.h>
  56 #include <sys/stat.h>
  57 #include <errno.h>
  58
  59 #define DPRINT(x...) printf(x)
  60
  61 #define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */
  62
  63
  64 /* Global information, common to all connections */
  65 typedef struct _GlobalInfo
  66 {
  67   struct ev_loop *loop;
  68   struct ev_io fifo_event;
  69   struct ev_timer timer_event;
  70   CURLM *multi;
  71   int still_running;
  72   FILE* input;
  73 } GlobalInfo;
  74
  75
  76 /* Information associated with a specific easy handle */
  77 typedef struct _ConnInfo
  78 {
  79   CURL *easy;
  80   char *url;
  81   GlobalInfo *global;
  82   char error[CURL_ERROR_SIZE];
  83 } ConnInfo;
  84
  85
  86 /* Information associated with a specific socket */
  87 typedef struct _SockInfo
  88 {
  89   curl_socket_t sockfd;
  90   CURL *easy;
  91   int action;
  92   long timeout;
  93   struct ev_io ev;
  94   int evset;
  95   GlobalInfo *global;
  96 } SockInfo;
  97
  98 static void timer_cb(EV_P_ struct ev_timer *w, int revents);
  99
 100 /* Update the event timer after curl_multi library calls */
 101 static int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo *g)
 102 {
 103   DPRINT("%s %li\n", __PRETTY_FUNCTION__,  timeout_ms);
 104   ev_timer_stop(g->loop, &g->timer_event);
 105   if (timeout_ms > 0)
 106   {
 107     double  t = timeout_ms / 1000;
 108     ev_timer_init(&g->timer_event, timer_cb, t, 0.);
 109     ev_timer_start(g->loop, &g->timer_event);
 110   }else
 111     timer_cb(g->loop, &g->timer_event, 0);
 112   return 0;
 113 }
 114
 115 /* Die if we get a bad CURLMcode somewhere */
 116 static void mcode_or_die(const char *where, CURLMcode code)
 117 {
 118   if ( CURLM_OK != code )
 119   {
 120     const char *s;
 121     switch ( code )
 122     {
 123     case CURLM_CALL_MULTI_PERFORM: s="CURLM_CALL_MULTI_PERFORM"; break;
 124     case CURLM_BAD_HANDLE:         s="CURLM_BAD_HANDLE";         break;
 125     case CURLM_BAD_EASY_HANDLE:    s="CURLM_BAD_EASY_HANDLE";    break;
 126     case CURLM_OUT_OF_MEMORY:      s="CURLM_OUT_OF_MEMORY";      break;
 127     case CURLM_INTERNAL_ERROR:     s="CURLM_INTERNAL_ERROR";     break;
 128     case CURLM_UNKNOWN_OPTION:     s="CURLM_UNKNOWN_OPTION";     break;
 129     case CURLM_LAST:               s="CURLM_LAST";               break;
 130     default: s="CURLM_unknown";
 131       break;
 132     case     CURLM_BAD_SOCKET:         s="CURLM_BAD_SOCKET";
 133       fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
 134       /* ignore this error */
 135       return;
 136     }
 137     fprintf(MSG_OUT, "ERROR: %s returns %s\n", where, s);
 138     exit(code);
 139   }
 140 }
 141
 142
 143
 144 /* Check for completed transfers, and remove their easy handles */
 145 static void check_multi_info(GlobalInfo *g)
 146 {
 147   char *eff_url;
 148   CURLMsg *msg;
 149   int msgs_left;
 150   ConnInfo *conn;
 151   CURL *easy;
 152   CURLcode res;
 153
 154   fprintf(MSG_OUT, "REMAINING: %d\n", g->still_running);
 155   while ((msg = curl_multi_info_read(g->multi, &msgs_left))) {
 156     if (msg->msg == CURLMSG_DONE) {
 157       easy = msg->easy_handle;
 158       res = msg->data.result;
 159       curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
 160       curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url);
 161       fprintf(MSG_OUT, "DONE: %s => (%d) %s\n", eff_url, res, conn->error);
 162       curl_multi_remove_handle(g->multi, easy);
 163       free(conn->url);
 164       curl_easy_cleanup(easy);
 165       free(conn);
 166     }
 167   }
 168 }
 169
 170
 171
 172 /* Called by libevent when we get action on a multi socket */
 173 static void event_cb(EV_P_ struct ev_io *w, int revents)
 174 {
 175   DPRINT("%s  w %p revents %i\n", __PRETTY_FUNCTION__, w, revents);
 176   GlobalInfo *g = (GlobalInfo*) w->data;
 177   CURLMcode rc;
 178
 179   int action = (revents&EV_READ?CURL_POLL_IN:0)|
 180     (revents&EV_WRITE?CURL_POLL_OUT:0);
 181   rc = curl_multi_socket_action(g->multi, w->fd, action, &g->still_running);
 182   mcode_or_die("event_cb: curl_multi_socket_action", rc);
 183   check_multi_info(g);
 184   if ( g->still_running <= 0 )
 185   {
 186     fprintf(MSG_OUT, "last transfer done, kill timeout\n");
 187     ev_timer_stop(g->loop, &g->timer_event);
 188   }
 189 }
 190
 191 /* Called by libevent when our timeout expires */
 192 static void timer_cb(EV_P_ struct ev_timer *w, int revents)
 193 {
 194   DPRINT("%s  w %p revents %i\n", __PRETTY_FUNCTION__, w, revents);
 195
 196   GlobalInfo *g = (GlobalInfo *)w->data;
 197   CURLMcode rc;
 198
 199   rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0, &g->still_running);
 200   mcode_or_die("timer_cb: curl_multi_socket_action", rc);
 201   check_multi_info(g);
 202 }
 203
 204 /* Clean up the SockInfo structure */
 205 static void remsock(SockInfo *f, GlobalInfo *g)
 206 {
 207   printf("%s  \n", __PRETTY_FUNCTION__);
 208   if ( f )
 209   {
 210     if ( f->evset )
 211       ev_io_stop(g->loop, &f->ev);
 212     free(f);
 213   }
 214 }
 215
 216
 217
 218 /* Assign information to a SockInfo structure */
 219 static void setsock(SockInfo*f, curl_socket_t s, CURL*e, int act, GlobalInfo*g)
 220 {
 221   printf("%s  \n", __PRETTY_FUNCTION__);
 222
 223   int kind = (act&CURL_POLL_IN?EV_READ:0)|(act&CURL_POLL_OUT?EV_WRITE:0);
 224
 225   f->sockfd = s;
 226   f->action = act;
 227   f->easy = e;
 228   if ( f->evset )
 229     ev_io_stop(g->loop, &f->ev);
 230   ev_io_init(&f->ev, event_cb, f->sockfd, kind);
 231   f->ev.data = g;
 232   f->evset=1;
 233   ev_io_start(g->loop, &f->ev);
 234 }
 235
 236
 237
 238 /* Initialize a new SockInfo structure */
 239 static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo *g)
 240 {
 241   SockInfo *fdp = calloc(sizeof(SockInfo), 1);
 242
 243   fdp->global = g;
 244   setsock(fdp, s, easy, action, g);
 245   curl_multi_assign(g->multi, s, fdp);
 246 }
 247
 248 /* CURLMOPT_SOCKETFUNCTION */
 249 static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp)
 250 {
 251   DPRINT("%s e %p s %i what %i cbp %p sockp %p\n",
 252          __PRETTY_FUNCTION__, e, s, what, cbp, sockp);
 253
 254   GlobalInfo *g = (GlobalInfo*) cbp;
 255   SockInfo *fdp = (SockInfo*) sockp;
 256   const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE"};
 257
 258   fprintf(MSG_OUT,
 259           "socket callback: s=%d e=%p what=%s ", s, e, whatstr[what]);
 260   if ( what == CURL_POLL_REMOVE )
 261   {
 262     fprintf(MSG_OUT, "\n");
 263     remsock(fdp, g);
 264   } else
 265   {
 266     if ( !fdp )
 267     {
 268       fprintf(MSG_OUT, "Adding data: %s\n", whatstr[what]);
 269       addsock(s, e, what, g);
 270     } else
 271     {
 272       fprintf(MSG_OUT,
 273               "Changing action from %s to %s\n",
 274               whatstr[fdp->action], whatstr[what]);
 275       setsock(fdp, s, e, what, g);
 276     }
 277   }
 278   return 0;
 279 }
 280
 281
 282 /* CURLOPT_WRITEFUNCTION */
 283 static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data)
 284 {
 285   size_t realsize = size * nmemb;
 286   ConnInfo *conn = (ConnInfo*) data;
 287   (void)ptr;
 288   (void)conn;
 289   return realsize;
 290 }
 291
 292
 293 /* CURLOPT_PROGRESSFUNCTION */
 294 static int prog_cb (void *p, double dltotal, double dlnow, double ult,
 295                     double uln)
 296 {
 297   ConnInfo *conn = (ConnInfo *)p;
 298   (void)ult;
 299   (void)uln;
 300
 301   fprintf(MSG_OUT, "Progress: %s (%g/%g)\n", conn->url, dlnow, dltotal);
 302   return 0;
 303 }
 304
 305
 306 /* Create a new easy handle, and add it to the global curl_multi */
 307 static void new_conn(char *url, GlobalInfo *g )
 308 {
 309   ConnInfo *conn;
 310   CURLMcode rc;
 311
 312   conn = calloc(1, sizeof(ConnInfo));
 313   memset(conn, 0, sizeof(ConnInfo));
 314   conn->error[0]='\0';
 315
 316   conn->easy = curl_easy_init();
 317   if ( !conn->easy )
 318   {
 319     fprintf(MSG_OUT, "curl_easy_init() failed, exiting!\n");
 320     exit(2);
 321   }
 322   conn->global = g;
 323   conn->url = strdup(url);
 324   curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url);
 325   curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb);
 326   curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, &conn);
 327   curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L);
 328   curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error);
 329   curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn);
 330   curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 0L);
 331   curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb);
 332   curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn);
 333   curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_TIME, 3L);
 334   curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_LIMIT, 10L);
 335
 336   fprintf(MSG_OUT,
 337           "Adding easy %p to multi %p (%s)\n", conn->easy, g->multi, url);
 338   rc = curl_multi_add_handle(g->multi, conn->easy);
 339   mcode_or_die("new_conn: curl_multi_add_handle", rc);
 340
 341   /* note that the add_handle() will set a time-out to trigger very soon so
 342      that the necessary socket_action() call will be called by this app */
 343 }
 344
 345 /* This gets called whenever data is received from the fifo */
 346 static void fifo_cb(EV_P_ struct ev_io *w, int revents)
 347 {
 348   char s[1024];
 349   long int rv=0;
 350   int n=0;
 351   GlobalInfo *g = (GlobalInfo *)w->data;
 352
 353   do
 354   {
 355     s[0]='\0';
 356     rv=fscanf(g->input, "%1023s%n", s, &n);
 357     s[n]='\0';
 358     if ( n && s[0] )
 359     {
 360       new_conn(s,g);  /* if we read a URL, go get it! */
 361     } else break;
 362   } while ( rv != EOF );
 363 }
 364
 365 /* Create a named pipe and tell libevent to monitor it */
 366 static int init_fifo (GlobalInfo *g)
 367 {
 368   struct stat st;
 369   static const char *fifo = "hiper.fifo";
 370   int sockfd;
 371
 372   fprintf(MSG_OUT, "Creating named pipe \"%s\"\n", fifo);
 373   if ( lstat (fifo, &st) == 0 )
 374   {
 375     if ( (st.st_mode & S_IFMT) == S_IFREG )
 376     {
 377       errno = EEXIST;
 378       perror("lstat");
 379       exit (1);
 380     }
 381   }
 382   unlink(fifo);
 383   if ( mkfifo (fifo, 0600) == -1 )
 384   {
 385     perror("mkfifo");
 386     exit (1);
 387   }
 388   sockfd = open(fifo, O_RDWR | O_NONBLOCK, 0);
 389   if ( sockfd == -1 )
 390   {
 391     perror("open");
 392     exit (1);
 393   }
 394   g->input = fdopen(sockfd, "r");
 395
 396   fprintf(MSG_OUT, "Now, pipe some URL's into > %s\n", fifo);
 397   ev_io_init(&g->fifo_event, fifo_cb, sockfd, EV_READ);
 398   ev_io_start(g->loop, &g->fifo_event);
 399   return(0);
 400 }
 401
 402 int main(int argc, char **argv)
 403 {
 404   GlobalInfo g;
 405   CURLMcode rc;
 406   (void)argc;
 407   (void)argv;
 408
 409   memset(&g, 0, sizeof(GlobalInfo));
 410   g.loop = ev_default_loop(0);
 411
 412   init_fifo(&g);
 413   g.multi = curl_multi_init();
 414
 415   ev_timer_init(&g.timer_event, timer_cb, 0., 0.);
 416   g.timer_event.data = &g;
 417   g.fifo_event.data = &g;
 418   curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
 419   curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g);
 420   curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
 421   curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g);
 422
 423   /* we don't call any curl_multi_socket*() function yet as we have no handles
 424      added! */
 425
 426   ev_loop(g.loop, 0);
 427   curl_multi_cleanup(g.multi);
 428   return 0;
 429 }