curl_lwt.ml

   1 (** Lwt support for Curl *)
   2
   3 module M = Curl.Multi
   4
   5 let log fmt = Printf.ksprintf prerr_endline fmt
   6
   7 type multi = {
   8   mt : Curl.Multi.mt;
   9   all_events : (Unix.file_descr, Lwt_engine.event list) Hashtbl.t;
  10   wakeners : (Curl.t, Curl.curlCode Lwt.u) Hashtbl.t;
  11 }
  12
  13 let create () =
  14   let mt = M.create () in
  15   let timer_event = ref Lwt_engine.fake_event in
  16   let all_events = Hashtbl.create 32 in
  17   let wakeners = Hashtbl.create 32 in
  18   let finished s =
  19     let rec loop n =
  20       match M.remove_finished mt with
  21       | None -> ()
  22       | Some (h,code) ->
  23         begin try
  24           let w = Hashtbl.find wakeners h in
  25           Hashtbl.remove wakeners h;
  26           Lwt.wakeup w code
  27         with Not_found ->
  28           prerr_endline "curl_lwt: orphan handle, how come?"
  29         end;
  30         loop (n+1)
  31     in
  32     loop 0
  33   in
  34   let on_readable fd _ =
  35     let (_:int) = M.action mt fd M.EV_IN in
  36     finished "on_readable";
  37   in
  38   let on_writable fd _ =
  39     let (_:int) = M.action mt fd M.EV_OUT in
  40     finished "on_writable";
  41   in
  42   let on_timer _ =
  43     Lwt_engine.stop_event !timer_event;
  44     M.action_timeout mt;
  45     finished "on_timer"
  46   in
  47   M.set_timer_function mt begin fun timeout ->
  48     Lwt_engine.stop_event !timer_event; (* duplicate stop_event is ok *)
  49     timer_event := Lwt_engine.on_timer (float_of_int timeout /. 1000.) false on_timer
  50   end;
  51   M.set_socket_function mt begin fun fd what ->
  52     begin
  53     try
  54       List.iter Lwt_engine.stop_event (Hashtbl.find all_events fd);
  55       Hashtbl.remove all_events fd;
  56     with
  57       Not_found -> () (* first event for the socket - no association *)
  58     end;
  59     let events = match what with
  60     | M.POLL_REMOVE | M.POLL_NONE -> []
  61     | M.POLL_IN -> [Lwt_engine.on_readable fd (on_readable fd)]
  62     | M.POLL_OUT -> [Lwt_engine.on_writable fd (on_writable fd)]
  63     | M.POLL_INOUT -> [Lwt_engine.on_readable fd (on_readable fd); Lwt_engine.on_writable fd (on_writable fd)]
  64     in
  65     match events with
  66     | [] -> ()
  67     | _ -> Hashtbl.add all_events fd events;
  68   end;
  69   { mt; all_events; wakeners; }
  70
  71 (* lwt may not run in parallel so one global is OK'ish *)
  72 let global = lazy (create ())
  73
  74 let setopt opt =
  75   let t = Lazy.force global in
  76   M.setopt t.mt opt
  77
  78 let perform h =
  79   let t = Lazy.force global in
  80   let (waiter,wakener) = Lwt.wait () in
  81   let waiter = Lwt.protected waiter in
  82   Lwt.on_cancel waiter (fun () ->
  83     Curl.Multi.remove t.mt h;
  84     Hashtbl.remove t.wakeners h;
  85   );
  86   Hashtbl.add t.wakeners h wakener;
  87   M.add t.mt h;
  88   waiter