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