asiohiper.cpp 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486
  1. /***************************************************************************
  2. * _ _ ____ _
  3. * Project ___| | | | _ \| |
  4. * / __| | | | |_) | |
  5. * | (__| |_| | _ <| |___
  6. * \___|\___/|_| \_\_____|
  7. *
  8. * Copyright (C) 2012 - 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. * demonstrate the use of multi socket interface with boost::asio
  24. * </DESC>
  25. */
  26. /*
  27. * This program is in c++ and uses boost::asio instead of libevent/libev.
  28. * Requires boost::asio, boost::bind and boost::system
  29. *
  30. * This is an adaptation of libcurl's "hiperfifo.c" and "evhiperfifo.c"
  31. * sample programs. This example implements a subset of the functionality from
  32. * hiperfifo.c, for full functionality refer hiperfifo.c or evhiperfifo.c
  33. *
  34. * Written by Lijo Antony based on hiperfifo.c by Jeff Pohlmeyer
  35. *
  36. * When running, the program creates an easy handle for a URL and
  37. * uses the curl_multi API to fetch it.
  38. *
  39. * Note:
  40. * For the sake of simplicity, URL is hard coded to "www.google.com"
  41. *
  42. * This is purely a demo app, all retrieved data is simply discarded by the
  43. * write callback.
  44. */
  45. #include <curl/curl.h>
  46. #include <boost/asio.hpp>
  47. #include <boost/bind.hpp>
  48. #include <iostream>
  49. #define MSG_OUT stdout /* Send info to stdout, change to stderr if you want */
  50. /* boost::asio related objects
  51. * using global variables for simplicity
  52. */
  53. boost::asio::io_service io_service;
  54. boost::asio::deadline_timer timer(io_service);
  55. std::map<curl_socket_t, boost::asio::ip::tcp::socket *> socket_map;
  56. /* Global information, common to all connections */
  57. typedef struct _GlobalInfo
  58. {
  59. CURLM *multi;
  60. int still_running;
  61. } GlobalInfo;
  62. /* Information associated with a specific easy handle */
  63. typedef struct _ConnInfo
  64. {
  65. CURL *easy;
  66. char *url;
  67. GlobalInfo *global;
  68. char error[CURL_ERROR_SIZE];
  69. } ConnInfo;
  70. static void timer_cb(const boost::system::error_code & error, GlobalInfo *g);
  71. /* Update the event timer after curl_multi library calls */
  72. static int multi_timer_cb(CURLM *multi, long timeout_ms, GlobalInfo *g)
  73. {
  74. fprintf(MSG_OUT, "\nmulti_timer_cb: timeout_ms %ld", timeout_ms);
  75. /* cancel running timer */
  76. timer.cancel();
  77. if(timeout_ms > 0) {
  78. /* update timer */
  79. timer.expires_from_now(boost::posix_time::millisec(timeout_ms));
  80. timer.async_wait(boost::bind(&timer_cb, _1, g));
  81. }
  82. else if(timeout_ms == 0) {
  83. /* call timeout function immediately */
  84. boost::system::error_code error; /*success*/
  85. timer_cb(error, g);
  86. }
  87. return 0;
  88. }
  89. /* Die if we get a bad CURLMcode somewhere */
  90. static void mcode_or_die(const char *where, CURLMcode code)
  91. {
  92. if(CURLM_OK != code) {
  93. const char *s;
  94. switch(code) {
  95. case CURLM_CALL_MULTI_PERFORM:
  96. s = "CURLM_CALL_MULTI_PERFORM";
  97. break;
  98. case CURLM_BAD_HANDLE:
  99. s = "CURLM_BAD_HANDLE";
  100. break;
  101. case CURLM_BAD_EASY_HANDLE:
  102. s = "CURLM_BAD_EASY_HANDLE";
  103. break;
  104. case CURLM_OUT_OF_MEMORY:
  105. s = "CURLM_OUT_OF_MEMORY";
  106. break;
  107. case CURLM_INTERNAL_ERROR:
  108. s = "CURLM_INTERNAL_ERROR";
  109. break;
  110. case CURLM_UNKNOWN_OPTION:
  111. s = "CURLM_UNKNOWN_OPTION";
  112. break;
  113. case CURLM_LAST:
  114. s = "CURLM_LAST";
  115. break;
  116. default:
  117. s = "CURLM_unknown";
  118. break;
  119. case CURLM_BAD_SOCKET:
  120. s = "CURLM_BAD_SOCKET";
  121. fprintf(MSG_OUT, "\nERROR: %s returns %s", where, s);
  122. /* ignore this error */
  123. return;
  124. }
  125. fprintf(MSG_OUT, "\nERROR: %s returns %s", where, s);
  126. exit(code);
  127. }
  128. }
  129. /* Check for completed transfers, and remove their easy handles */
  130. static void check_multi_info(GlobalInfo *g)
  131. {
  132. char *eff_url;
  133. CURLMsg *msg;
  134. int msgs_left;
  135. ConnInfo *conn;
  136. CURL *easy;
  137. CURLcode res;
  138. fprintf(MSG_OUT, "\nREMAINING: %d", g->still_running);
  139. while((msg = curl_multi_info_read(g->multi, &msgs_left))) {
  140. if(msg->msg == CURLMSG_DONE) {
  141. easy = msg->easy_handle;
  142. res = msg->data.result;
  143. curl_easy_getinfo(easy, CURLINFO_PRIVATE, &conn);
  144. curl_easy_getinfo(easy, CURLINFO_EFFECTIVE_URL, &eff_url);
  145. fprintf(MSG_OUT, "\nDONE: %s => (%d) %s", eff_url, res, conn->error);
  146. curl_multi_remove_handle(g->multi, easy);
  147. free(conn->url);
  148. curl_easy_cleanup(easy);
  149. free(conn);
  150. }
  151. }
  152. }
  153. /* Called by asio when there is an action on a socket */
  154. static void event_cb(GlobalInfo *g, curl_socket_t s,
  155. int action, const boost::system::error_code & error,
  156. int *fdp)
  157. {
  158. fprintf(MSG_OUT, "\nevent_cb: action=%d", action);
  159. if(socket_map.find(s) == socket_map.end()) {
  160. fprintf(MSG_OUT, "\nevent_cb: socket already closed");
  161. return;
  162. }
  163. /* make sure the event matches what are wanted */
  164. if(*fdp == action || *fdp == CURL_POLL_INOUT) {
  165. CURLMcode rc;
  166. if(error)
  167. action = CURL_CSELECT_ERR;
  168. rc = curl_multi_socket_action(g->multi, s, action, &g->still_running);
  169. mcode_or_die("event_cb: curl_multi_socket_action", rc);
  170. check_multi_info(g);
  171. if(g->still_running <= 0) {
  172. fprintf(MSG_OUT, "\nlast transfer done, kill timeout");
  173. timer.cancel();
  174. }
  175. /* keep on watching.
  176. * the socket may have been closed and/or fdp may have been changed
  177. * in curl_multi_socket_action(), so check them both */
  178. if(!error && socket_map.find(s) != socket_map.end() &&
  179. (*fdp == action || *fdp == CURL_POLL_INOUT)) {
  180. boost::asio::ip::tcp::socket *tcp_socket = socket_map.find(s)->second;
  181. if(action == CURL_POLL_IN) {
  182. tcp_socket->async_read_some(boost::asio::null_buffers(),
  183. boost::bind(&event_cb, g, s,
  184. action, _1, fdp));
  185. }
  186. if(action == CURL_POLL_OUT) {
  187. tcp_socket->async_write_some(boost::asio::null_buffers(),
  188. boost::bind(&event_cb, g, s,
  189. action, _1, fdp));
  190. }
  191. }
  192. }
  193. }
  194. /* Called by asio when our timeout expires */
  195. static void timer_cb(const boost::system::error_code & error, GlobalInfo *g)
  196. {
  197. if(!error) {
  198. fprintf(MSG_OUT, "\ntimer_cb: ");
  199. CURLMcode rc;
  200. rc = curl_multi_socket_action(g->multi, CURL_SOCKET_TIMEOUT, 0,
  201. &g->still_running);
  202. mcode_or_die("timer_cb: curl_multi_socket_action", rc);
  203. check_multi_info(g);
  204. }
  205. }
  206. /* Clean up any data */
  207. static void remsock(int *f, GlobalInfo *g)
  208. {
  209. fprintf(MSG_OUT, "\nremsock: ");
  210. if(f) {
  211. free(f);
  212. }
  213. }
  214. static void setsock(int *fdp, curl_socket_t s, CURL *e, int act, int oldact,
  215. GlobalInfo *g)
  216. {
  217. fprintf(MSG_OUT, "\nsetsock: socket=%d, act=%d, fdp=%p", s, act, fdp);
  218. std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it =
  219. socket_map.find(s);
  220. if(it == socket_map.end()) {
  221. fprintf(MSG_OUT, "\nsocket %d is a c-ares socket, ignoring", s);
  222. return;
  223. }
  224. boost::asio::ip::tcp::socket * tcp_socket = it->second;
  225. *fdp = act;
  226. if(act == CURL_POLL_IN) {
  227. fprintf(MSG_OUT, "\nwatching for socket to become readable");
  228. if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
  229. tcp_socket->async_read_some(boost::asio::null_buffers(),
  230. boost::bind(&event_cb, g, s,
  231. CURL_POLL_IN, _1, fdp));
  232. }
  233. }
  234. else if(act == CURL_POLL_OUT) {
  235. fprintf(MSG_OUT, "\nwatching for socket to become writable");
  236. if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
  237. tcp_socket->async_write_some(boost::asio::null_buffers(),
  238. boost::bind(&event_cb, g, s,
  239. CURL_POLL_OUT, _1, fdp));
  240. }
  241. }
  242. else if(act == CURL_POLL_INOUT) {
  243. fprintf(MSG_OUT, "\nwatching for socket to become readable & writable");
  244. if(oldact != CURL_POLL_IN && oldact != CURL_POLL_INOUT) {
  245. tcp_socket->async_read_some(boost::asio::null_buffers(),
  246. boost::bind(&event_cb, g, s,
  247. CURL_POLL_IN, _1, fdp));
  248. }
  249. if(oldact != CURL_POLL_OUT && oldact != CURL_POLL_INOUT) {
  250. tcp_socket->async_write_some(boost::asio::null_buffers(),
  251. boost::bind(&event_cb, g, s,
  252. CURL_POLL_OUT, _1, fdp));
  253. }
  254. }
  255. }
  256. static void addsock(curl_socket_t s, CURL *easy, int action, GlobalInfo *g)
  257. {
  258. /* fdp is used to store current action */
  259. int *fdp = (int *) calloc(sizeof(int), 1);
  260. setsock(fdp, s, easy, action, 0, g);
  261. curl_multi_assign(g->multi, s, fdp);
  262. }
  263. /* CURLMOPT_SOCKETFUNCTION */
  264. static int sock_cb(CURL *e, curl_socket_t s, int what, void *cbp, void *sockp)
  265. {
  266. fprintf(MSG_OUT, "\nsock_cb: socket=%d, what=%d, sockp=%p", s, what, sockp);
  267. GlobalInfo *g = (GlobalInfo*) cbp;
  268. int *actionp = (int *) sockp;
  269. const char *whatstr[] = { "none", "IN", "OUT", "INOUT", "REMOVE"};
  270. fprintf(MSG_OUT,
  271. "\nsocket callback: s=%d e=%p what=%s ", s, e, whatstr[what]);
  272. if(what == CURL_POLL_REMOVE) {
  273. fprintf(MSG_OUT, "\n");
  274. remsock(actionp, g);
  275. }
  276. else {
  277. if(!actionp) {
  278. fprintf(MSG_OUT, "\nAdding data: %s", whatstr[what]);
  279. addsock(s, e, what, g);
  280. }
  281. else {
  282. fprintf(MSG_OUT,
  283. "\nChanging action from %s to %s",
  284. whatstr[*actionp], whatstr[what]);
  285. setsock(actionp, s, e, what, *actionp, g);
  286. }
  287. }
  288. return 0;
  289. }
  290. /* CURLOPT_WRITEFUNCTION */
  291. static size_t write_cb(void *ptr, size_t size, size_t nmemb, void *data)
  292. {
  293. size_t written = size * nmemb;
  294. char *pBuffer = (char *)malloc(written + 1);
  295. strncpy(pBuffer, (const char *)ptr, written);
  296. pBuffer[written] = '\0';
  297. fprintf(MSG_OUT, "%s", pBuffer);
  298. free(pBuffer);
  299. return written;
  300. }
  301. /* CURLOPT_PROGRESSFUNCTION */
  302. static int prog_cb(void *p, double dltotal, double dlnow, double ult,
  303. double uln)
  304. {
  305. ConnInfo *conn = (ConnInfo *)p;
  306. (void)ult;
  307. (void)uln;
  308. fprintf(MSG_OUT, "\nProgress: %s (%g/%g)", conn->url, dlnow, dltotal);
  309. fprintf(MSG_OUT, "\nProgress: %s (%g)", conn->url, ult);
  310. return 0;
  311. }
  312. /* CURLOPT_OPENSOCKETFUNCTION */
  313. static curl_socket_t opensocket(void *clientp, curlsocktype purpose,
  314. struct curl_sockaddr *address)
  315. {
  316. fprintf(MSG_OUT, "\nopensocket :");
  317. curl_socket_t sockfd = CURL_SOCKET_BAD;
  318. /* restrict to IPv4 */
  319. if(purpose == CURLSOCKTYPE_IPCXN && address->family == AF_INET) {
  320. /* create a tcp socket object */
  321. boost::asio::ip::tcp::socket *tcp_socket =
  322. new boost::asio::ip::tcp::socket(io_service);
  323. /* open it and get the native handle*/
  324. boost::system::error_code ec;
  325. tcp_socket->open(boost::asio::ip::tcp::v4(), ec);
  326. if(ec) {
  327. /* An error occurred */
  328. std::cout << std::endl << "Couldn't open socket [" << ec << "][" <<
  329. ec.message() << "]";
  330. fprintf(MSG_OUT, "\nERROR: Returning CURL_SOCKET_BAD to signal error");
  331. }
  332. else {
  333. sockfd = tcp_socket->native_handle();
  334. fprintf(MSG_OUT, "\nOpened socket %d", sockfd);
  335. /* save it for monitoring */
  336. socket_map.insert(std::pair<curl_socket_t,
  337. boost::asio::ip::tcp::socket *>(sockfd, tcp_socket));
  338. }
  339. }
  340. return sockfd;
  341. }
  342. /* CURLOPT_CLOSESOCKETFUNCTION */
  343. static int close_socket(void *clientp, curl_socket_t item)
  344. {
  345. fprintf(MSG_OUT, "\nclose_socket : %d", item);
  346. std::map<curl_socket_t, boost::asio::ip::tcp::socket *>::iterator it =
  347. socket_map.find(item);
  348. if(it != socket_map.end()) {
  349. delete it->second;
  350. socket_map.erase(it);
  351. }
  352. return 0;
  353. }
  354. /* Create a new easy handle, and add it to the global curl_multi */
  355. static void new_conn(char *url, GlobalInfo *g)
  356. {
  357. ConnInfo *conn;
  358. CURLMcode rc;
  359. conn = (ConnInfo *) calloc(1, sizeof(ConnInfo));
  360. conn->easy = curl_easy_init();
  361. if(!conn->easy) {
  362. fprintf(MSG_OUT, "\ncurl_easy_init() failed, exiting!");
  363. exit(2);
  364. }
  365. conn->global = g;
  366. conn->url = strdup(url);
  367. curl_easy_setopt(conn->easy, CURLOPT_URL, conn->url);
  368. curl_easy_setopt(conn->easy, CURLOPT_WRITEFUNCTION, write_cb);
  369. curl_easy_setopt(conn->easy, CURLOPT_WRITEDATA, &conn);
  370. curl_easy_setopt(conn->easy, CURLOPT_VERBOSE, 1L);
  371. curl_easy_setopt(conn->easy, CURLOPT_ERRORBUFFER, conn->error);
  372. curl_easy_setopt(conn->easy, CURLOPT_PRIVATE, conn);
  373. curl_easy_setopt(conn->easy, CURLOPT_NOPROGRESS, 1L);
  374. curl_easy_setopt(conn->easy, CURLOPT_PROGRESSFUNCTION, prog_cb);
  375. curl_easy_setopt(conn->easy, CURLOPT_PROGRESSDATA, conn);
  376. curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_TIME, 3L);
  377. curl_easy_setopt(conn->easy, CURLOPT_LOW_SPEED_LIMIT, 10L);
  378. /* call this function to get a socket */
  379. curl_easy_setopt(conn->easy, CURLOPT_OPENSOCKETFUNCTION, opensocket);
  380. /* call this function to close a socket */
  381. curl_easy_setopt(conn->easy, CURLOPT_CLOSESOCKETFUNCTION, close_socket);
  382. fprintf(MSG_OUT,
  383. "\nAdding easy %p to multi %p (%s)", conn->easy, g->multi, url);
  384. rc = curl_multi_add_handle(g->multi, conn->easy);
  385. mcode_or_die("new_conn: curl_multi_add_handle", rc);
  386. /* note that the add_handle() will set a time-out to trigger very soon so
  387. that the necessary socket_action() call will be called by this app */
  388. }
  389. int main(int argc, char **argv)
  390. {
  391. GlobalInfo g;
  392. (void)argc;
  393. (void)argv;
  394. memset(&g, 0, sizeof(GlobalInfo));
  395. g.multi = curl_multi_init();
  396. curl_multi_setopt(g.multi, CURLMOPT_SOCKETFUNCTION, sock_cb);
  397. curl_multi_setopt(g.multi, CURLMOPT_SOCKETDATA, &g);
  398. curl_multi_setopt(g.multi, CURLMOPT_TIMERFUNCTION, multi_timer_cb);
  399. curl_multi_setopt(g.multi, CURLMOPT_TIMERDATA, &g);
  400. new_conn((char *)"www.google.com", &g); /* add a URL */
  401. /* enter io_service run loop */
  402. io_service.run();
  403. curl_multi_cleanup(g.multi);
  404. fprintf(MSG_OUT, "\ndone.\n");
  405. return 0;
  406. }