<commit_before>open Batteries
open Common

type handled_token = int

(* First a unit -> receive_handler function is used to get a function
   that handles each input.

   The handling function returns both a continuation for handling following
   lines and another function, that is used to handle the "ok" *)
type receive_handler = Cont of (string -> (receive_handler * receive_finish))
and receive_finish = unit -> unit

type receiver_state = {
  mutable received_ack		: int;
  mutable line_callbacks	: (unit -> (receive_handler * receive_finish)) BatDeque.t;
}

let register_of_char = function
  | 'X' -> `X
  | 'Y' -> `Y
  | 'Z' -> `Z
  | 'E' -> `E
  | 'F' -> `F
  | x -> failwith ("register_of_char: unknown register " ^ String.make 1 x)

type t = { 
  fd		: Unix.file_descr;
  mutable line	: int;
  receiver	: receiver_state Protect.t;
}

type 'a request = unit -> ((t -> unit) * (('a -> unit) -> unit -> (receive_handler * receive_finish)))

let rec ignore_loop : receive_handler =
  Cont (fun _str -> (ignore_loop, fun () -> Printf.printf "Ignoring finish after loop\n%!"))

let reader (fd, state) =
  let buf = String.create 1024 in
  let lb = LineBuffer.create () in
  let rec get_next_handler () : (receive_handler * receive_finish) option =
    let handler =
      Protect.access state (
	fun st ->
	  match BatDeque.front st.line_callbacks with
	    | None -> (fun () -> None)
	    | Some ((handler : unit -> (receive_handler * receive_finish)), remaining) ->
		st.line_callbacks <- remaining;
		fun () -> Some (handler ())
      ) ()
    in
      handler
  in
  let rec loop (handler : (receive_handler * receive_finish) option) =
    let n = Unix.read fd buf 0 (String.length buf) in
      if n > 0 
      then (
	let strs = LineBuffer.append_substring lb buf 0 n in
	let process_json handler str =
	  let open Json in
	  let json = from_string str in
	  match Some json +> "r" with
	  | None -> (* don't handle these at the moment *) handler
	  | Some r ->
	    let handler =
	      match handler with
	      | None -> get_next_handler ()
	      | Some handler -> Some handler
	    in
	    ( match handler with
	    | None -> None
	    | Some (Cont f, _finish) -> 
	      let (_handler, finish) = f (to_string r) in
	      Protect.access state (
		fun st ->
		  st.received_ack <- st.received_ack + 1;
	      );
	      finish ();
	      None
	    )
	in
	let handler =
	  List.fold_left
	    (fun (handler : (receive_handler * receive_finish) option) str ->
	       Printf.printf "CNC<-%s\n%!" str;
	       let handler : (receive_handler * receive_finish) option =
		 match str with
		 | str when Pcre.pmatch ~pat:"^{" str ->
		   process_json handler str
		 | str when Pcre.pmatch ~pat:"ok" str ->
		   (match handler with
		   | None -> BatOption.may (fun (_, finish) -> finish ()) (get_next_handler ())
		   | Some (_, finish) -> finish ());
		   Protect.access state (
		     fun st ->
		       st.received_ack <- st.received_ack + 1;
		   );
		   None
		 | "start" ->
		   Protect.access state (
		     fun st -> <commit_msg>Cnc.reader: skip unparseable json, show an error<commit_after>open Batteries
open Common

type handled_token = int

(* First a unit -> receive_handler function is used to get a function
   that handles each input.

   The handling function returns both a continuation for handling following
   lines and another function, that is used to handle the "ok" *)
type receive_handler = Cont of (string -> (receive_handler * receive_finish))
and receive_finish = unit -> unit

type receiver_state = {
  mutable received_ack		: int;
  mutable line_callbacks	: (unit -> (receive_handler * receive_finish)) BatDeque.t;
}

let register_of_char = function
  | 'X' -> `X
  | 'Y' -> `Y
  | 'Z' -> `Z
  | 'E' -> `E
  | 'F' -> `F
  | x -> failwith ("register_of_char: unknown register " ^ String.make 1 x)

type t = { 
  fd		: Unix.file_descr;
  mutable line	: int;
  receiver	: receiver_state Protect.t;
}

type 'a request = unit -> ((t -> unit) * (('a -> unit) -> unit -> (receive_handler * receive_finish)))

let rec ignore_loop : receive_handler =
  Cont (fun _str -> (ignore_loop, fun () -> Printf.printf "Ignoring finish after loop\n%!"))

let reader (fd, state) =
  let buf = String.create 1024 in
  let lb = LineBuffer.create () in
  let rec get_next_handler () : (receive_handler * receive_finish) option =
    let handler =
      Protect.access state (
	fun st ->
	  match BatDeque.front st.line_callbacks with
	    | None -> (fun () -> None)
	    | Some ((handler : unit -> (receive_handler * receive_finish)), remaining) ->
		st.line_callbacks <- remaining;
		fun () -> Some (handler ())
      ) ()
    in
      handler
  in
  let rec loop (handler : (receive_handler * receive_finish) option) =
    let n = Unix.read fd buf 0 (String.length buf) in
      if n > 0 
      then (
	let strs = LineBuffer.append_substring lb buf 0 n in
	let process_json handler str =
	  let open Json in
	  let json =
	    try Some (from_string str)
	    with _ -> None
	  in
	    match json +> "r" with
	    | None ->
	      (* don't handle these at the moment *) 
	      Printf.fprintf stderr "Cnc.reader.loop.process_json: cannot parse %s\n%!" str;
	      handler
	    | Some r ->
	      let handler =
		match handler with
		| None -> get_next_handler ()
		| Some handler -> Some handler
	      in
	      ( match handler with
	      | None -> None
	      | Some (Cont f, _finish) -> 
		let (_handler, finish) = f (to_string r) in
		Protect.access state (
		  fun st ->
		    st.received_ack <- st.received_ack + 1;
		);
		finish ();
		None
	      )
	  in
	let handler =
	  List.fold_left
	    (fun (handler : (receive_handler * receive_finish) option) str ->
	       Printf.printf "CNC<-%s\n%!" str;
	       let handler : (receive_handler * receive_finish) option =
		 match str with
		 | str when Pcre.pmatch ~pat:"{" str ->
		   process_json handler str
		 | str when Pcre.pmatch ~pat:"ok" str ->
		   (match handler with
		   | None -> BatOption.may (fun (_, finish) -> finish ()) (get_next_handler ())
		   | Some (_, finish) -> finish ());
		   Protect.access state (
		     fun st ->
		       st.received_ack <- st.received_ack + 1;
		   );
		   None
		 | "start" ->
		   Protect.access state (
		     fun st -> 
{'fraction_non_alphanumeric': 0.11981419189492232, 'fraction_numerical': 0.0032035880185808103, 'mean_word_length': 2.5416903006239364, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 86, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2695902', 'n_tokens_mistral': 2168, 'n_tokens_neox': 1967, 'n_words': 655}