3.0.0

.github/workflows/test.yml

@@ -13,16 +13,36 @@ jobs:
     - uses: avsm/setup-ocaml@master
       with:
         ocaml-version: ${{ matrix.ocaml-version }}
+    - name: Install pkg-config
+      if: runner.os == 'macOS'
+      run: brew install pkg-config
     - name: Deps
+      if: runner.os != 'Windows'
       run: |
         opam pin add -n conduit.dev .
         opam pin add -n conduit-lwt.dev .
         opam pin add -n conduit-lwt-unix.dev .
         opam pin add -n conduit-async.dev .
         opam pin add -n conduit-mirage.dev .
-        opam depext -y conduit conduit-lwt conduit-lwt-unix conduit-async conduit-mirage
-        opam install --deps-only .
+        opam pin add -n conduit-tls.dev .
+        opam depext  -y conduit conduit-tls conduit-lwt conduit-lwt-unix conduit-async conduit-mirage
+        opam install --deps-only -t conduit conduit-tls conduit-lwt conduit-lwt-unix conduit-async conduit-mirage
+    - name: Deps (Windows)
+      if: runner.os == 'Windows'
+      run: |
+        opam pin add -n conduit.dev .
+        opam pin add -n conduit-lwt.dev .
+        opam pin add -n conduit-lwt-unix.dev .
+        opam pin add -n conduit-mirage.dev .
+        opam pin add -n conduit-tls.dev .
+        opam depext  -y conduit conduit-tls conduit-lwt conduit-lwt-unix conduit-mirage
+        opam install --deps-only -t conduit conduit-tls conduit-lwt conduit-lwt-unix conduit-mirage
+    - name: Build (Windows)
+      if: runner.os == 'Windows'
+      run: opam exec -- dune build -p conduit,conduit-tls,conduit-lwt,conduit-lwt-unix,conduit-mirage
     - name: Build
+      if: runner.os != 'Windows'
       run: opam exec -- dune build
     - name: Test
-      run: opam exec -- dune runtest
+      if: runner.os != 'Windows'
+      run: opam exec -- dune runtest --no-buffer --verbose -j 1

.ocamlformat

@@ -0,0 +1,9 @@
+version = 0.14.2
+break-infix = fit-or-vertical
+parse-docstrings = true
+indicate-multiline-delimiters=no
+nested-match=align
+sequence-style=separator
+break-before-in=auto
+if-then-else=keyword-first
+parse-docstrings=false

async/conduit_async.ml

@@ -1,23 +1,80 @@
-(*
- * Copyright (c) 2012-2017 Anil Madhavapeddy <[email protected]>
- *
- * Permission to use, copy, modify, and distribute this software for any
- * purpose with or without fee is hereby granted, provided that the above
- * copyright notice and this permission notice appear in all copies.
- *
- * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
- * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
- * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
- * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
- * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
- * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
- * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
- *
- *)
-
-module V1 = V1
-module V2 = V2
-module V3 = V3
-
-[@@@deprecated "Use Conduit_async.V1"]
-include V1.Conduit_async
+module Async_scheduler = struct
+  type +'a t = 'a Async.Deferred.t
+
+  let bind x f = Async.Deferred.bind x ~f
+
+  let return x = Async.Deferred.return x
+end
+
+include Conduit.Make (Async_scheduler) (Cstruct) (Cstruct)
+
+let failwith fmt = Format.kasprintf failwith fmt
+
+let ( >>? ) x f = Async.Deferred.Result.bind x ~f
+
+let serve_with_handler :
+    type cfg master flow.
+    handler:(flow -> unit Async.Deferred.t) ->
+    service:(cfg, master, flow) Service.service ->
+    cfg ->
+    unit Async.Condition.t * unit Async.Deferred.t =
+ fun ~handler ~service cfg ->
+  let open Async in
+  let stop = Async.Condition.create () in
+  let module Svc = (val Service.impl service) in
+  let main =
+    Service.serve cfg ~service >>= function
+    | Error err -> failwith "%a" Service.pp_error err
+    | Ok master -> (
+        let rec loop () =
+          let close = Async.Condition.wait stop >>| fun () -> Ok `Stop in
+          let accept =
+            Svc.accept master >>? fun flow ->
+            Async.(Deferred.ok (return (`Flow flow))) in
+
+          Async.Deferred.any [ close; accept ] >>= function
+          | Ok (`Flow flow) ->
+              Async.don't_wait_for (handler flow) ;
+              Async.Scheduler.yield () >>= fun () -> (loop [@tailcall]) ()
+          | Ok `Stop -> Svc.close master
+          | Error err0 -> (
+              Svc.close master >>= function
+              | Ok () -> Async.return (Error err0)
+              | Error _err1 -> Async.return (Error err0)) in
+        loop () >>= function
+        | Ok () -> Async.return ()
+        | Error err -> failwith "%a" Svc.pp_error err) in
+  (stop, main)
+
+let reader_and_writer_of_flow flow =
+  let open Async in
+  let recv flow writer =
+    let tmp = Cstruct.create 0x1000 in
+    let rec loop () =
+      Client.recv flow tmp >>= function
+      | Ok (`Input len) ->
+          Pipe.write writer (Cstruct.to_string (Cstruct.sub tmp 0 len)) >>= loop
+      | Ok `End_of_input ->
+          Pipe.close writer ;
+          Async.return ()
+      | Error err -> failwith "%a" Client.pp_error err in
+    loop () in
+  let send flow reader =
+    let rec loop () =
+      Pipe.read reader >>= function
+      | `Eof -> Async.return ()
+      | `Ok v ->
+          let rec go tmp =
+            if Cstruct.len tmp = 0
+            then Async.return ()
+            else
+              Client.send flow tmp >>= function
+              | Ok shift -> go (Cstruct.shift tmp shift)
+              | Error err -> failwith "%a" Client.pp_error err in
+          go (Cstruct.of_string v) >>= loop in
+    loop () in
+  let preader = Pipe.create_reader ~close_on_exception:true (recv flow) in
+  let pwriter = Pipe.create_writer (send flow) in
+  Reader.of_pipe (Core.Info.of_string "reader") preader >>= fun reader ->
+  Writer.of_pipe (Core.Info.of_string "writer") pwriter >>= fun (writer, _) ->
+  Async.return (reader, writer)

async/conduit_async.mli

@@ -0,0 +1,19 @@
+(** Conduit with Async. *)
+
+module Async_scheduler :
+  Conduit.Sigs.SCHEDULER with type +'a t = 'a Async.Deferred.t
+
+include
+  Conduit.S
+    with type input = Cstruct.t
+     and type output = Cstruct.t
+     and type +'a s = 'a Async.Deferred.t
+
+val serve_with_handler :
+  handler:('flow -> unit Async.Deferred.t) ->
+  service:('cfg, 'master, 'flow) Service.service ->
+  'cfg ->
+  unit Async.Condition.t * unit Async.Deferred.t
+
+val reader_and_writer_of_flow :
+  Client.flow -> (Async.Reader.t * Async.Writer.t) Async.Deferred.t