data_encoding: read streams of Mbytes.t, tests, remove dead code (debug functions), expose a function that checks without reading the data

.gitignore

@@ -43,6 +43,7 @@
 /test/reports
 
 /test/utils/test-data-encoding
+/test/utils/test-stream-data-encoding
 /test/utils/test-merkle
 /test/utils/test-lwt-pipe
 /test/p2p/test-p2p-io-scheduler

src/minutils/data_encoding.ml

@@ -28,6 +28,7 @@ exception Unexpected_tag of int
 exception Duplicated_tag of int
 exception Invalid_tag of int * [ `Uint8 | `Uint16 ]
 exception Unexpected_enum of string * string list
+exception Invalid_size of int
 
 let apply fs v =
   let rec loop = function
@@ -1206,7 +1207,9 @@ let rec length : type x. x t -> x -> int = fun e ->
         let read = read_rec e in
         fun buf ofs len ->
           let ofs, sz = int32 buf ofs len in
-          read buf ofs (Int32.to_int sz)
+          let sz = Int32.to_int sz in
+          if sz < 0 then raise (Invalid_size sz);
+          read buf ofs sz
 
   let read t buf ofs len =
     try Some (read_rec t buf ofs len)
@@ -1230,4 +1233,437 @@ let rec length : type x. x t -> x -> int = fun e ->
     | Some n -> n
     | None -> invalid_arg "Data_encoding.Binary.fixed_length_exn"
 
+
+  (* Facilities to decode streams of binary data *)
+
+  type 'a status =
+    | Success of { res : 'a ; res_len : int ; remaining : MBytes.t list }
+    | Await of (MBytes.t -> 'a status)
+    | Error
+
+  module Stream_reader = struct
+
+    (* used as a zipper to code the function read_checker with the
+       ability to stop and wait for more data. In 'P_seq' case, data
+       length is parameterized by the current offset. Hence, it's a
+       function 'fun_data_len'. For the 'P_list' case, we store the
+       base offset (before starting reading the elements) and the
+       number of elements that have been read so far. *)
+    type path =
+      | P_top : path
+      | P_await : { path : path ; encoding : 'a t ; data_len : int } -> path
+      | P_seq : { path : path ; encoding : 'a t ;
+                  fun_data_len : int -> int } -> path
+      | P_list : { path:path ; encoding:'a t ; data_len : int ;
+                    base_ofs : int ; nb_elts_read : int } -> path
+
+    (* used to accumulate given mbytes when reading a list of blocks,
+       as well as the current offset and the number of unread bytes *)
+    type mbytes_stream = {
+      past : MBytes.t Queue.t ; (* data that have been entirely read *)
+      future : (MBytes.t * int) Queue.t ; (* data that are not (fully) read *)
+      mutable past_len : int ; (*length of concatenation of data in 'past'*)
+      mutable unread : int ;  (*number of cells that are unread in 'future'*)
+      ofs : int (*current absolute offset wrt to concatenation past @ future*)
+    }
+
+    (* exception raised when additional mbytes are needed to continue
+       decoding *)
+    exception Need_more_data
+
+    (* read a data that is stored in may Mbytes *)
+    let read_from_many_blocks reader buf ofs d_ofs =
+      let tmp = MBytes.create d_ofs in (*we will merge data in this mbyte*)
+      let r = ref d_ofs in (*to count the cells to be read*)
+      let rel_ofs = ref ofs in (*= ofs for first mbyte, 0 for others*)
+      while !r > 0 do
+        assert (not (Queue.is_empty buf.future)) ;
+        let b, len_b = Queue.peek buf.future in (*take the next mbyte*)
+        let len_chunk = len_b - !rel_ofs in (*the number of cells to read*)
+        if !r >= len_chunk then
+          begin (*copy b in 'past' if it is read entirely*)
+            ignore (Queue.pop buf.future) ;
+            Queue.push b buf.past ;
+            buf.past_len <- buf.past_len + len_b ;
+          end ;
+        (* copy (min !r len_chunk) data from b to tmp *)
+        MBytes.blit b !rel_ofs tmp (d_ofs - !r) (min !r len_chunk) ;
+        r := !r - len_chunk ; (* len_chunk data read during this round*)
+        rel_ofs := 0 ; (*next mbytes will be read starting from zero*)
+      done ;
+      reader tmp 0 d_ofs
+
+
+    (* generic function that reads data from an mbytes_stream. It is
+       parameterized by a function "reader" that effectively reads the
+       data *)
+    let generic_read_data delta_ofs reader buf =
+      let absolute_ofs  = buf.ofs in
+      if buf.unread < delta_ofs then (*not enough data*)
+        raise Need_more_data ;
+      if delta_ofs = 0 then (*we'll read nothing*)
+        buf, reader (MBytes.create 0) 0 0
+      else
+        let new_ofs = absolute_ofs + delta_ofs in
+        let ofs = absolute_ofs - buf.past_len in (*relative ofs wrt 'future'*)
+        buf.unread <- buf.unread-delta_ofs ; (*'delta_ofs' cells will be read*)
+        assert (not (Queue.is_empty buf.future)) ; (*we have some data to read*)
+        let b, len_b = Queue.peek buf.future in
+        let buf = { buf with ofs = new_ofs } in
+        if ofs + delta_ofs > len_b then
+          (*should read data from many mbytes*)
+          buf, read_from_many_blocks reader buf ofs delta_ofs
+        else
+          begin
+            if ofs + delta_ofs = len_b then
+              begin (*the rest of b will be entirely read. Put it in 'past'*)
+                ignore (Queue.pop buf.future) ;
+                Queue.push b buf.past ;
+                buf.past_len <- buf.past_len + len_b ;
+              end ;
+            buf, reader b ofs delta_ofs
+          end
+
+
+    (* functions that try to read data from a given mbytes_stream,
+       or raise Need_more_data *)
+
+    let int8 buf =
+      generic_read_data Size.int8 (fun x y _ -> MBytes.get_int8 x y) buf
+
+    let uint8 buf =
+      generic_read_data Size.uint8 (fun x y _ -> MBytes.get_uint8 x y) buf
+
+    let char buf =
+      let buf, v = int8 buf in
+      buf, Char.chr v
+
+    let bool buf =
+      let buf, v = int8 buf in
+      buf, v <> 0
+
+    let int16 buf =
+      generic_read_data Size.int16 (fun x y _ -> MBytes.get_int16 x y) buf
+
+    let uint16 buf =
+      generic_read_data Size.uint16 (fun x y _ -> MBytes.get_uint16 x y) buf
+
+    let int31 buf =
+      generic_read_data Size.int31
+        (fun x y _ -> Int32.to_int (MBytes.get_int32 x y)) buf
+
+    let int32 buf =
+      generic_read_data Size.int32 (fun x y _ -> MBytes.get_int32 x y) buf
+
+    let int64 buf =
+      generic_read_data Size.int64 (fun x y _ -> MBytes.get_int64 x y) buf
+
+    (** read a float64 (double) **)
+    let float buf =
+      (*Here, float means float64, which is read using MBytes.get_double !!*)
+      generic_read_data Size.float (fun x y _ -> MBytes.get_double x y) buf
+
+    let fixed_length_bytes length buf =
+      generic_read_data length MBytes.sub buf
+
+    let fixed_length_string length buf =
+      generic_read_data length MBytes.substring buf
+
+    let read_tag = function
+      | `Uint8 -> uint8
+      | `Uint16 -> uint16
+
+    (* auxiliary function: computing size of data in branches
+       Objs(`Variable) and Tups(`Variable) *)
+    let varseq_lengths e1 e2 ofs len = match classify e1, classify e2 with
+      | (`Dynamic | `Fixed _), `Variable -> len, (fun ofs' -> len - ofs' + ofs)
+      | `Variable, `Fixed n -> (len - n), (fun _ -> n)
+      | _ -> assert false (* Should be rejected by Kind.combine *)
+
+
+    (* adaptation of function read_rec to check binary data
+       incrementally.  The function takes (and returns) a 'path' (for
+       incrementality), and 'mbytes_stream' *)
+    let rec data_checker
+      : type a.
+        path -> a encoding -> mbytes_stream -> int ->
+        path * mbytes_stream =
+      fun path e buf len ->
+        (*length of data with `Variable kind should be given by the caller*)
+        assert (classify e != `Variable || len >= 0) ;
+        try match e.encoding with
+          | Null   -> next_path path buf
+          | Empty  -> next_path path buf
+          | Constant _ -> next_path path buf
+          | Ignore -> next_path path { buf with ofs = buf.ofs + len }
+          | Bool   -> next_path path (fst (bool   buf))
+          | Int8   -> next_path path (fst (int8   buf))
+          | Uint8  -> next_path path (fst (uint8  buf))
+          | Int16  -> next_path path (fst (int16  buf))
+          | Uint16 -> next_path path (fst (uint16 buf))
+          | Int31  -> next_path path (fst (int31  buf))
+          | Int32  -> next_path path (fst (int32  buf))
+          | Int64  -> next_path path (fst (int64  buf))
+          | Float  -> next_path path (fst (float  buf))
+          | Bytes (`Fixed n) ->
+              next_path path (fst (fixed_length_bytes n buf))
+
+          | String (`Fixed n) ->
+              next_path path (fst (fixed_length_string n buf))
+
+          | Bytes `Variable ->
+              next_path path (fst (fixed_length_bytes len buf))
+
+          | String `Variable ->
+              next_path path (fst (fixed_length_string len buf))
+
+          | String_enum (kind, _) -> (* ! approx! *)
+              data_checker path (make @@ (String kind)) buf len
+
+          | Array e ->
+              let p = P_list { path ; encoding = e ; base_ofs = buf.ofs ;
+                               data_len = len ; nb_elts_read = 0 } in
+              next_path p buf
+
+          | List e ->
+              let p = P_list { path ; encoding = e ; base_ofs = buf.ofs ;
+                               data_len = len ; nb_elts_read = 0 } in
+              next_path p buf
+
+          | Obj (Req (_, e)) -> data_checker path e buf len
+
+          | Obj (Opt (`Dynamic, _, e)) ->
+              let buf, v = int8 buf in
+              if v = 0 then next_path path buf
+              else data_checker path e buf (len - Size.int8)
+
+          | Obj (Opt (`Variable, _, e)) ->
+              if len = 0 then next_path path buf
+              else data_checker path e buf len
+
+          | Obj (Dft (_, e, _)) -> data_checker path e buf len
+
+          | Objs ((`Fixed _ | `Dynamic), e1, e2) ->
+              let f_len2 ofs' = len - (ofs' - buf.ofs) in
+              let path =
+                P_seq { path ; encoding = e2 ; fun_data_len = f_len2 } in
+              data_checker path e1 buf len
+
+          | Objs (`Variable, e1, e2) ->
+              let len1, f_len2 = varseq_lengths e1 e2 buf.ofs len in
+              let path =
+                P_seq { path ; encoding = e2 ; fun_data_len = f_len2 } in
+              data_checker path e1 buf len1
+
+          | Tup e -> data_checker path e buf len
+
+          | Tups ((`Fixed _ | `Dynamic), e1, e2) ->
+              let f_len2 ofs' = len - (ofs' - buf.ofs) in
+              let path =
+                P_seq { path ; encoding = e2 ; fun_data_len = f_len2 } in
+              data_checker path e1 buf len
+
+          | Tups (`Variable, e1, e2) ->
+              let len1, f_len2 = varseq_lengths e1 e2 buf.ofs len in
+              let path =
+                P_seq { path ; encoding = e2 ; fun_data_len = f_len2 } in
+              data_checker path e1 buf len1
+
+          | Conv { encoding = e } -> data_checker path e buf len
+
+          | Describe { encoding = e } -> data_checker path e buf len
+
+          | Def { encoding = e } -> data_checker path e buf len
+
+          | Splitted { encoding = e } -> data_checker path e buf len
+
+          | Mu (_, _, self) -> data_checker path (self e) buf len
+
+          | Union (_, sz, cases) ->
+              let buf, ctag = read_tag sz buf in
+              let opt =
+                List.fold_left
+                  (fun acc c -> match c with
+                     | (Case { encoding ; tag = Some tag })
+                       when tag == ctag ->
+                         assert (acc == None) ;
+                         Some (data_checker path encoding buf)
+                     | _ -> acc
+                  )None cases
+              in
+              begin match opt with
+                | None -> raise (Unexpected_tag ctag)
+                | Some func -> func (len - (tag_size sz))
+              end
+
+          | Dynamic_size e ->
+              let buf, sz = int32 buf in
+              let sz = Int32.to_int sz in
+              if sz < 0 then raise (Invalid_size sz) ;
+              data_checker path e buf sz
+
+        with Need_more_data ->
+          P_await { path ; encoding = e ; data_len = len }, buf
+
+    and next_path : path -> mbytes_stream -> path * mbytes_stream =
+      fun path buf ->
+        match path with
+        | P_top ->
+            P_top, buf (* success case *)
+
+        | P_seq { path ; encoding ; fun_data_len } ->
+            (* check the right branch of a sequence. fun_data_len ofs gives
+               the length of the data to read *)
+            data_checker path encoding buf (fun_data_len buf.ofs)
+
+        | P_await { path ; encoding ; data_len } ->
+            (* resume from an await *)
+            data_checker path encoding buf data_len
+
+        | P_list
+            ({ path ; encoding ; base_ofs ; data_len ; nb_elts_read } as r) ->
+            (* read/check an eventual element of a list *)
+            if data_len = buf.ofs - base_ofs then
+              (* we've read all the elements of the list *)
+              next_path path buf
+            else
+              begin
+                (*some more elements to read*)
+                assert (data_len > buf.ofs - base_ofs) ;
+                (*check: if we've already read some elements, then currrent ofs
+                  should be greater then initial ofs *)
+                assert (nb_elts_read <= 0 || buf.ofs - base_ofs > 0) ;
+                let path =
+                  P_list { r with nb_elts_read = nb_elts_read + 1} in
+                data_checker path encoding buf data_len
+              end
+
+    let data_checker = next_path
+
+    (* insert a given MBytes.t in a given mbytes_stream *)
+    let insert_mbytes mb_buf mb =
+      let len = MBytes.length mb in
+      if len > 0 then begin
+        Queue.push (mb, len) mb_buf.future ;
+        mb_buf.unread <- mb_buf.unread + len ;
+      end
+
+    (* aux function called when data_checker succeeds: splits a given
+       mbytes_stream into a 'read' and 'unread' queues. This may
+       modify the content of the given mbytes_stream *)
+    let split_mbytes_stream { past_len ; past ; future ; unread ; ofs } =
+      let rel_ofs = ofs - past_len in
+      assert (rel_ofs >= 0) ;
+      if rel_ofs = 0 then past, future (* already done *)
+      else begin
+        assert (not(Queue.is_empty future)) ; (*because data_checker succeeded*)
+        let b, len = Queue.pop future in
+        assert (rel_ofs < len) ; (*inv. maintained by read_from_many_blocks*)
+        let b1 = MBytes.sub b 0 rel_ofs in (* read part of b *)
+        let b2 = MBytes.sub b rel_ofs (len-rel_ofs) in (* unread part of b *)
+        Queue.push b1 past ;
+
+        (* push b2 at the beginning of 'future' using Queue.transfer*)
+        let tmp = Queue.create() in
+        Queue.push (b2, unread) tmp ;
+        Queue.transfer future tmp ; (*tmp === b2 ::: future in constant time*)
+        past, tmp
+      end
+
+    (* given a state, this function returns a new status:
+       - if data are successfully checked, accumulated mbytes are
+         passed to 'success_result' that computes the final
+         result. Unread mbytes are also returned
+       - if some more data are needed, a function that waits for some
+         additional mbytes is returned
+       - eventual errors are reported/returned *)
+    let rec bytes_stream_reader_rec (path, mb_buf) success_result =
+      let success =
+        match path with
+        | P_top -> true
+        | P_await _ -> false
+        | _ -> assert false
+      in
+      assert (mb_buf.ofs >= mb_buf.past_len) ;
+      if success then
+        let q_read, q_unread = split_mbytes_stream mb_buf in
+        match success_result q_read mb_buf.ofs with
+        | Some a ->
+            let remaining =
+              List.rev @@
+              Queue.fold
+                (fun acc (b, len) ->
+                   if len = 0 then acc else b:: acc) [] q_unread
+            in
+            Success { res = a ; res_len = mb_buf.ofs ; remaining }
+        | None -> Error
+        (* success_result may fail because data_checker is
+           approximative in some situations *)
+      else
+        Await
+          (fun mb ->
+             insert_mbytes mb_buf mb ;
+             try
+               let state = data_checker path mb_buf in
+               bytes_stream_reader_rec state success_result
+             with _ -> Error)
+
+    (* This function checks reading a stream of 'MBytes.t' wrt. a given
+       encoding:
+       - the given data encoding should have a 'Fixed' or a 'Dynamic'
+       size, otherwise an error is returned,
+       - the function returns an 'Error', a function w
+       ('Await w') that waits for more data (Mbytes.t), or
+       'Success'. The function is parameterized by 'success_result'
+       that computes the data to return in case of success.
+       An exception 'Invalid_argument "streaming data with variable
+       size"' is raised if the encoding has a variable size *)
+    let bytes_stream_reader :
+      MBytes.t list -> 'a t ->
+      (MBytes.t Queue.t -> int -> 'b option) -> 'b status
+      = fun l e success_result ->
+      match classify e with
+      | `Variable -> invalid_arg "streaming data with variable size"
+      | `Fixed _ | `Dynamic ->
+          let mb_buf = {
+            past   = Queue.create() ; past_len = 0 ;
+            future = Queue.create() ; unread = 0; ofs = 0 }
+          in
+          List.iter (insert_mbytes mb_buf) l ;
+          let path =
+            P_await { path = P_top ; encoding = e ; data_len = - 1 } in
+          try bytes_stream_reader_rec (data_checker path mb_buf) success_result
+          with _ -> Error
+
+  end
+
+  (* concats a queue of mbytes into one MByte *)
+  let concat_mbyte_chunks queue tot_len =
+    if Queue.length queue = 1 then Queue.pop queue (* no copy *)
+    else (* copy smaller mbytes into one big mbyte *)
+      let buf = MBytes.create tot_len in
+      let cpt = ref 0 in
+      let tot_len' = ref tot_len in
+      while not (Queue.is_empty queue) do
+        let mb = Queue.pop queue in
+        let len = MBytes.length mb in
+        tot_len' := !tot_len' - len ;
+        assert (!tot_len' >= 0) ;
+        MBytes.blit mb 0 buf !cpt len ;
+        cpt := !cpt + len ;
+      done ;
+      assert (!tot_len' = 0) ;
+      buf
+
+  (* Decode a stream of MBytes. see
+     Stream_reader.bytes_stream_traversal for more details *)
+  let read_stream_of_bytes ?(init=[]) encoding =
+    Stream_reader.bytes_stream_reader init encoding
+      (fun read_q ofs -> of_bytes encoding (concat_mbyte_chunks read_q ofs))
+
+  (* Check reading a stream of MBytes. see
+     Stream_reader.bytes_stream_traversal for more details *)
+  let check_stream_of_bytes ?(init=[]) encoding =
+    Stream_reader.bytes_stream_reader init encoding (fun _ _ -> Some ())
+
 end

src/minutils/data_encoding.mli

@@ -249,6 +249,30 @@ module Binary : sig
       MBytes.sub used internally *)
   val to_bytes_list : ?copy_blocks:bool -> int  -> 'a t -> 'a -> MBytes.t list
 
+  (** This type is used when decoding binary data incrementally.
+      - In case of 'Success', the decoded data, the size of used data
+       to decode the result, and the remaining data are returned
+      - In case of error, 'Error' is returned
+      - 'Await' status embeds a function that waits for additional data
+       to continue decoding, when given data are not sufficient *)
+  type 'a status =
+    | Success of { res : 'a ; res_len : int ; remaining : MBytes.t list }
+    | Await of (MBytes.t -> 'a status)
+    | Error
+
+  (** This function allows to decode (or to initialize decoding) a
+      stream of 'MByte.t'. The given data encoding should have a
+      'Fixed' or a 'Dynamic' size, otherwise an exception
+      'Invalid_argument "streaming data with variable size"' is
+      raised *)
+  val read_stream_of_bytes : ?init:MBytes.t list -> 'a t -> 'a status
+
+  (** Like read_stream_of_bytes, but only checks that the stream can
+      be read. Note that this is an approximation because failures
+      that may come from conversion functions present in encodings are
+      not checked *)
+  val check_stream_of_bytes : ?init:MBytes.t list -> 'a t -> unit status
+
   val fixed_length : 'a encoding -> int option
   val fixed_length_exn : 'a encoding -> int
 

test/utils/Makefile

@@ -4,6 +4,7 @@ SRCDIR=../../src
 TESTS := \
   merkle \
   data-encoding \
+  stream-data-encoding \
   # lwt-pipe
 
 include ../Makefile.shared
@@ -71,3 +72,21 @@ test-data-encoding: ${LIB} ${TEST_DATA_ENCODING_IMPLS:.ml=.cmx}
 
 clean::
 	rm -f test-data-encoding
+
+############################################################################
+## Streamed data_encoding
+
+.PHONY:run-test-stream-data-encoding
+run-test-stream-data-encoding:
+	@echo
+	./test-stream-data-encoding
+
+TEST_DATA_ENCODING_IMPLS := \
+  test_data_encoding.ml
+
+test-stream-data-encoding: ${LIB} ${TEST_DATA_ENCODING_IMPLS:.ml=.cmx}
+	@echo LINK $(notdir $@)
+	@${OCAMLOPT} -linkall -linkpkg ${OCAMLFLAGS} -o $@ $^
+
+clean::
+	rm -f test-stream-data-encoding

test/utils/test_stream_data_encoding.ml

@@ -0,0 +1,452 @@
+open Data_encoding
+open Context
+open Hash
+open Error_monad
+
+let (>>=) = Lwt.bind
+let (>|=) = Lwt.(>|=)
+let (//) = Filename.concat
+
+let is_invalid_arg = function
+  | Invalid_argument _ -> true
+  | _ -> false
+
+
+let is_await = function Binary.Await _ -> true | _ -> false
+let is_success = function Binary.Success _ -> true | _ -> false
+let is_error = function Binary.Error _ -> true | _ -> false
+
+
+let rec fold_left_pending f accu l =
+  match l with
+  | [] -> accu
+  | a::l -> fold_left_pending f (f accu a l) l
+
+let test_read_simple_bin_ko_invalid_data
+    ?msg ?(not_equal=Assert.not_equal) encoding value =
+  let len_data = MBytes.length (Binary.to_bytes encoding value) in
+  if classify encoding != `Variable && len_data > 0 then
+    for sz = 1 to len_data do
+      let l = (Binary.to_bytes_list sz encoding value) in
+      List.iter (fun b ->
+          for i = 0 to MBytes.length b - 1 do
+            (* alter data *)
+            MBytes.set_int8 b i ((MBytes.get_int8 b i)+1)
+          done
+        )l;
+      ignore(
+        fold_left_pending
+          (fun _done e _todo ->
+             let _done = e :: _done in
+             begin
+               let status =
+                 Binary.read_stream_of_bytes ~init:(List.rev _done) encoding in
+               let status =
+                 List.fold_left
+                   (fun status mbyte ->
+                      match status with
+                      | Binary.Await f -> f mbyte
+                      | _ -> status
+                   )status _todo
+               in
+               match status with
+               | Binary.Await _ -> ()
+               | Binary.Error _ -> ()
+               | Binary.Success {res; remaining} ->
+                   (* should not have "Success" *)
+                   Assert.equal ~msg:__LOC__ remaining [];
+                   not_equal value res
+             end;
+             _done
+          )[] l
+      )
+    done
+
+let unexpected loc =
+  loc ^ ": This case should not happen"
+
+let test_read_simple_bin_ko_await ?msg encoding value =
+  let len_data = MBytes.length (Binary.to_bytes encoding value) in
+  if classify encoding != `Variable && len_data > 0 then
+    for sz = 1 to len_data do
+      let l = Binary.to_bytes_list sz encoding value in
+      match List.rev l with
+      | [] -> Assert.fail_msg "%s" (unexpected __LOC__)
+      | e::r ->
+          let l = List.rev r in (* last mbyte removed !! *)
+          ignore(
+            fold_left_pending
+              (fun _done e _todo ->
+                 let _done = e :: _done in
+                 begin
+                   let status=
+                     Binary.read_stream_of_bytes
+                       ~init:(List.rev _done) encoding in
+                   let status =
+                     List.fold_left
+                       (fun status mbyte ->
+                          if not (is_await status) then
+                            Assert.fail_msg "%s" (unexpected __LOC__);
+                          match status with
+                          | Binary.Await f -> f mbyte
+                          | _ -> status
+                       )status _todo
+                   in
+                   match status with
+                   | Binary.Await _ -> ()
+                   | Binary.Error _ ->
+                       if not (classify encoding == `Variable) then
+                         Assert.fail_msg "%s" (unexpected __LOC__)
+                   | Binary.Success result ->
+                       Assert.fail_msg "%s" (unexpected __LOC__)
+                 end;
+                 _done
+              )[] l
+          )
+    done
+
+let test_read_simple_bin_ok ?msg ?(equal=Assert.equal) encoding value =
+  let len_data = max 1 (MBytes.length (Binary.to_bytes encoding value)) in
+  for sz = 1 to len_data do
+    ignore(
+      fold_left_pending
+        (fun _done e _todo ->
+           let _done = e :: _done in
+           begin
+             let status =
+               Binary.read_stream_of_bytes ~init:(List.rev _done) encoding in
+             let status =
+               List.fold_left
+                 (fun status mbyte ->
+                    if MBytes.length mbyte <> 0 && is_success status then
+                      Assert.fail_msg "%s" (unexpected __LOC__);
+                    match status with
+                    | Binary.Await f -> f mbyte
+                    | _ -> status
+                 )status _todo
+             in
+             match status with
+             | Binary.Success {res; remaining} ->
+                 Assert.equal ~msg:__LOC__ remaining [];
+                 equal ?msg value res
+             | Binary.Await _ -> Assert.fail_msg "%s" (unexpected __LOC__)
+             | Binary.Error _ ->
+                 if not (classify encoding == `Variable) then
+                   Assert.fail_msg "%s" (unexpected __LOC__)
+           end;
+           _done
+        )[] (Binary.to_bytes_list sz encoding value)
+    )
+  done
+
+let test_check_simple_bin_ko_invalid_data
+    ?msg ?(not_equal=Assert.not_equal) encoding value =
+  let len_data = MBytes.length (Binary.to_bytes encoding value) in
+  if classify encoding != `Variable && len_data > 0 then
+    for sz = 1 to len_data do
+      let l = (Binary.to_bytes_list sz encoding value) in
+      List.iter (fun b ->
+          for i = 0 to MBytes.length b - 1 do
+            (* alter data *)
+            MBytes.set_int8 b i ((MBytes.get_int8 b i)+1)
+          done
+        )l;
+      ignore(
+        fold_left_pending
+          (fun _done e _todo ->
+             let _done = e :: _done in
+             begin
+               let status =
+                 Binary.check_stream_of_bytes ~init:(List.rev _done) encoding in
+               let status =
+                 List.fold_left
+                   (fun status mbyte ->
+                      match status with
+                      | Binary.Await f -> f mbyte
+                      | _ -> status
+                   )status _todo
+               in
+               match status with
+               | Binary.Await _ -> ()
+               | Binary.Error _ -> ()
+               | Binary.Success {res; remaining} ->
+                   Assert.equal ~msg:__LOC__ remaining [];
+                   (* res is unit for check *)
+             end;
+             _done
+          )[] l
+      )
+    done
+
+let test_check_simple_bin_ko_await ?msg encoding value =
+  let len_data = MBytes.length (Binary.to_bytes encoding value) in
+  if classify encoding != `Variable && len_data > 0 then
+    for sz = 1 to len_data do
+      let l = Binary.to_bytes_list sz encoding value in
+      match List.rev l with
+      | [] -> Assert.fail_msg "%s" (unexpected __LOC__)
+      | e::r ->
+          let l = List.rev r in (* last mbyte removed !! *)
+          ignore(
+            fold_left_pending
+              (fun _done e _todo ->
+                 let _done = e :: _done in
+                 begin
+                   let status=
+                     Binary.check_stream_of_bytes
+                       ~init:(List.rev _done) encoding in
+                   let status =
+                     List.fold_left
+                       (fun status mbyte ->
+                          if not (is_await status) then
+                            Assert.fail_msg "%s" (unexpected __LOC__);
+                          match status with
+                          | Binary.Await f -> f mbyte
+                          | _ -> status
+                       )status _todo
+                   in
+                   match status with
+                   | Binary.Await _ -> ()
+                   | Binary.Error _ ->
+                       if not (classify encoding == `Variable) then
+                         Assert.fail_msg "%s" (unexpected __LOC__)
+                   | Binary.Success result ->
+                       Assert.fail_msg "%s" (unexpected __LOC__)
+                 end;
+                 _done
+              )[] l
+          )
+    done
+
+let test_check_simple_bin_ok ?msg ?(equal=Assert.equal) encoding value =
+  let len_data = max 1 (MBytes.length (Binary.to_bytes encoding value)) in
+  for sz = 1 to len_data do
+    ignore(
+      fold_left_pending
+        (fun _done e _todo ->
+           let _done = e :: _done in
+           begin
+             let status =
+               Binary.check_stream_of_bytes ~init:(List.rev _done) encoding in
+             let status =
+               List.fold_left
+                 (fun status mbyte ->
+                    if MBytes.length mbyte <> 0 && is_success status then
+                      Assert.fail_msg "%s" (unexpected __LOC__);
+                    match status with
+                    | Binary.Await f -> f mbyte
+                    | _ -> status
+                 )status _todo
+             in
+             match status with
+             | Binary.Success {res; remaining} ->
+                 Assert.equal ~msg:__LOC__ remaining [];
+                 (* res is unit for check *)
+             | Binary.Await _ -> Assert.fail_msg "%s" (unexpected __LOC__)
+             | Binary.Error _ ->
+                 if not (classify encoding == `Variable) then
+                   Assert.fail_msg "%s" (unexpected __LOC__)
+           end;
+           _done
+        )[] (Binary.to_bytes_list sz encoding value)
+    )
+  done
+
+let test_simple
+    ~msg ?(equal=Assert.equal) ?(not_equal=Assert.not_equal) enc value
+  =
+  test_check_simple_bin_ok ~msg:(msg ^ ": binary-ok") ~equal enc value;
+  test_check_simple_bin_ko_await ~msg:(msg ^ ": binary-ko_await") enc value;
+  test_check_simple_bin_ko_invalid_data
+    ~msg:(msg ^ ": binary-invalid_data") ~not_equal enc value;
+
+  test_read_simple_bin_ok ~msg:(msg ^ ": binary-ok") ~equal enc value;
+  test_read_simple_bin_ko_await ~msg:(msg ^ ": binary-ko_await") enc value;
+  test_read_simple_bin_ko_invalid_data
+    ~msg:(msg ^ ": binary-invalid_data") ~not_equal enc value
+
+
+
+
+let test_simple_int ~msg encoding i =
+  let range_min = - (1 lsl (i-1)) in
+  let range_max = (1 lsl (i-1)) - 1 in
+  test_simple ~msg encoding range_min ;
+  test_simple ~msg encoding range_max
+
+let test_simple_uint ~msg encoding i =
+  let range_min = 0 in
+  let range_max = (1 lsl i) - 1 in
+  test_simple ~msg encoding range_min ;
+  test_simple ~msg encoding range_max
+
+let test_simple_values _ =
+  test_simple ~msg:__LOC__ null ();
+  test_simple ~msg:__LOC__ empty ();
+  test_simple ~msg:__LOC__ (constant "toto") ();
+  test_simple_int ~msg:__LOC__ int8 8;
+  test_simple_uint ~msg:__LOC__ uint8 8;
+  test_simple_int ~msg:__LOC__ int16 16;
+  test_simple_uint ~msg:__LOC__ uint16 16;
+  test_simple_int ~msg:__LOC__ int31 31;
+  test_simple ~msg:__LOC__ int32 Int32.min_int;
+  test_simple ~msg:__LOC__ int32 Int32.max_int;
+  test_simple ~msg:__LOC__ int64 Int64.min_int;
+  test_simple ~msg:__LOC__ int64 Int64.max_int;
+  test_simple ~msg:__LOC__ bool true;
+  test_simple ~msg:__LOC__ bool false;
+  test_simple ~msg:__LOC__ string "tutu";
+  test_simple ~msg:__LOC__ bytes (MBytes.of_string "titi");
+  test_simple ~msg:__LOC__ float 42.;
+  test_simple ~msg:__LOC__ float max_float;
+  test_simple ~msg:__LOC__ float min_float;
+  test_simple ~msg:__LOC__ float (-. 0.);
+  test_simple ~msg:__LOC__ float (+. 0.);
+  test_simple ~msg:__LOC__ float infinity;
+  test_simple ~msg:__LOC__ float neg_infinity;
+  test_simple ~msg:__LOC__ float epsilon_float;
+  test_simple ~msg:__LOC__ ~equal:Assert.equal_float float nan;
+  test_simple ~msg:__LOC__ (option string) (Some "thing");
+  test_simple ~msg:__LOC__ (option string) None;
+  let enum_enc =
+    ["one", 1; "two", 2; "three", 3; "four", 4; "five", 6; "six", 6] in
+  test_simple ~msg:__LOC__ (string_enum enum_enc) 4;
+
+  Lwt.return_unit
+
+
+type t = A of int | B of string | C of int | D of string | E
+
+let prn_t = function
+  | A i -> Printf.sprintf "A %d" i
+  | B s -> Printf.sprintf "B %s" s
+  | C i -> Printf.sprintf "C %d" i
+  | D s -> Printf.sprintf "D %s" s
+  | E -> "E"
+
+let test_union _ =
+  let enc =
+    (union [
+        case ~tag:1
+          int8
+          (function A i -> Some i | _ -> None)
+          (fun i -> A i) ;
+        case ~tag:2
+          string
+          (function B s -> Some s | _ -> None)
+          (fun s -> B s) ;
+        case ~tag:3
+          int8
+          (function C i -> Some i | _ -> None)
+          (fun i -> C i) ;
+        case ~tag:4
+          (obj2
+             (req "kind" (constant "D"))
+             (req "data" (string)))
+          (function D s -> Some ((), s) | _ -> None)
+          (fun ((), s) -> D s) ;
+      ]) in
+  let jsonA = Json.construct enc (A 1) in
+  let jsonB = Json.construct enc (B "2") in
+  let jsonC = Json.construct enc (C 3) in
+  let jsonD = Json.construct enc (D"4") in
+  Assert.test_fail
+    ~msg:__LOC__ (fun () -> Json.construct enc E) is_invalid_arg ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (A 1) (Json.destruct enc jsonA) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (B "2") (Json.destruct enc jsonB) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (A 3) (Json.destruct enc jsonC) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (D "4") (Json.destruct enc jsonD) ;
+  let binA = Binary.to_bytes_list 1 enc (A 1) in
+  let binB = Binary.to_bytes_list 1 enc (B "2") in
+  let binC = Binary.to_bytes_list 1 enc (C 3) in
+  let binD = Binary.to_bytes_list 1 enc (D "4") in
+  Assert.test_fail ~msg:__LOC__ (fun () -> Binary.to_bytes enc E)
+    (function
+      | No_case_matched -> true
+      | _ -> false) ;
+  let get_result ~msg bin_l =
+    let status = Binary.read_stream_of_bytes enc in
+    let status =
+      List.fold_left
+        (fun status mbyte ->
+           match status with
+           | Binary.Await f -> f mbyte
+           | _ -> status
+        )status bin_l
+    in
+    match status with
+    | Binary.Error _ -> Assert.fail_msg "%s" msg
+    | Binary.Await _ -> Assert.fail_msg "%s" msg
+    | Binary.Success {res} -> res
+  in
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (A 1) (get_result ~msg:__LOC__ binA) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (B "2") (get_result ~msg:__LOC__ binB) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (C 3) (get_result ~msg:__LOC__ binC) ;
+  Assert.equal ~prn:prn_t ~msg:__LOC__ (D "4") (get_result ~msg:__LOC__ binD) ;
+  Lwt.return_unit
+
+type s = { field : int }
+
+let test_splitted _ =
+  let s_enc =
+    def "s" @@
+    describe
+      ~title:"testsuite encoding test"
+      ~description: "A human readable description" @@
+    conv
+      (fun s -> string_of_int s.field)
+      (fun s -> { field = int_of_string s })
+      string in
+  let enc =
+    (splitted
+       ~binary:string
+       ~json:
+         (union [
+             case ~tag:1
+               string
+               (fun _ -> None)
+               (fun s -> s) ;
+             case ~tag:2
+               s_enc
+               (fun s -> Some { field = int_of_string s })
+              (fun s -> string_of_int s.field) ;
+           ])) in
+  let get_result ~msg bin_l =
+    let status = Binary.read_stream_of_bytes enc in
+    let status =
+      List.fold_left
+        (fun status mbyte ->
+           match status with
+           | Binary.Await f -> f mbyte
+           | _ -> status
+        )status bin_l
+    in
+    match status with
+    | Binary.Error _ -> Assert.fail_msg "%s" msg
+    | Binary.Await _ -> Assert.fail_msg "%s" msg
+    | Binary.Success {res} -> res
+  in
+  let jsonA = Json.construct enc "41" in
+  let jsonB = Json.construct s_enc {field = 42} in
+  let binA = Binary.to_bytes_list 1 enc "43" in
+  let binB = Binary.to_bytes_list 1 s_enc {field = 44} in
+  Assert.equal ~msg:__LOC__ "41" (Json.destruct enc jsonA);
+  Assert.equal ~msg:__LOC__ "42" (Json.destruct enc jsonB);
+  Assert.equal ~msg:__LOC__ "43" (get_result ~msg:__LOC__ binA);
+  Assert.equal ~msg:__LOC__ "44" (get_result ~msg:__LOC__ binB);
+  Lwt.return_unit
+
+   
+let wrap_test f base_dir =
+  f base_dir >>= fun result ->
+  return result
+
+let tests = [
+  "simple", test_simple_values ;
+  "union", test_union ;
+  "splitted", test_splitted ;
+]
+
+let () =
+  Test.run "stream_data_encoding."
+    (List.map (fun (s, f) -> s, wrap_test f) tests)