ghiper.c 12 KB

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