stream.ml

open Lwt.Infix

let add_to_table tbl str =
	Base.(Hashtbl.set tbl ~key:str ~data:(Hashtbl.find_or_add tbl str ~default:(fun () -> 0) + 1))

let stream_stdin () =
	Lwt_stream.from (fun () -> try Lwt.return (Some (input_line stdin)) with End_of_file -> Lwt.return None)

let main c tbl =
  Lwt_stream.iter_n ~max_concurrency:c (fun a -> Lwt.return (add_to_table tbl a)) (stream_stdin ());;

let () =
	let concur = Base.Int.of_string (Sys.argv.(1)) in
	let tbl = Base.(Hashtbl.create (module String)) in
		(main concur tbl >>= fun () -> Lwt.return (Base.(Hashtbl.iter_keys tbl ~f:(fun k -> Stdlib.print_endline (k ^ (Int.to_string (Hashtbl.find_exn tbl k))))))) |> ignore