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