<commit_before>open Batteries
open Amqp_types

exception Unknown_frame of int
exception Decode_error

let protocol_header = "AMQP\x00\x00\x09\x01"

let peer_properties = Table
type peer_properties = table

module Transport = struct
  type t = { i: IO.input; o: unit IO.output; }

  let connect () =
    let i, o = Unix.(open_connection ~autoclose:false (ADDR_INET (inet_addr_loopback, 5672))) in
    IO.output o protocol_header 0 (String.length protocol_header) |> ignore;
    IO.flush o;
    { i; o; }

end

let print_bytes bytes =
  String.iter (fun c -> Printf.printf "%02x " (Char.code c)) bytes;
  Printf.printf "\n%!";

module Framing = struct

  (* Imperative. We will change that later *)
  type hdr = (int * int)
  type channel_id = int

  type channel_state =
    | Ready
    | Waiting of hdr * int * Buffer.t

  type message =
    | Method of string
    | Body of string

  type callback = hdr -> message -> unit

  type channel = {  callback: callback;
                    mutable state: channel_state;
                 }

  let channels : (int, channel) Hashtbl.t = Hashtbl.create 0

  let frame_end = 0xce

  let frame          = Octet :: Short :: Longstr :: Octet :: Nil
  let read_frame     = read frame
  let write_frame    = write frame

  let method_frame   = Short :: Short :: Nil
  let content_header = Short :: Short :: Longlong :: Short :: Nil

  let read_method_frame = read method_frame
  let write_method_frame : (string IO.output -> 'a)  = write method_frame

  let read_content_header = read content_header
  let write_content_header  : (string IO.output -> 'a) = write content_header

  let write output channel ((cid, mid), message) =
    let (data, tpe) = match message with
      | Method data -> data, Amqp_spec.frame_method
      | Body data -> data, Amqp_spec.frame_header (* TODO *)
    in
    let hdr_data =
      write_method_frame (IO.output_string ()) cid mid
      |> IO.close_out
    in
    write_frame (IO.output_string ()) tpe channel (hdr_data ^ data) frame_end
    |> IO.close_out
    |> tap print_bytes
    |> (fun x -> IO.output output x 0 (String.length x))
    |> ignore;
    IO.flush output

  let read input =
    let (tpe, channel_id, data, magic) = read_frame (Tuple4.curry identity) input in
    assert (magic = frame_end);
    Printf.printf "Channel: %d\n%!" channel_id;
    let channel = Hashtbl.find channels channel_id in
    match tpe with
    | n when n = Amqp_spec.frame_method ->
      (* Standard method message *)
      assert (channel.state = Ready);
      let hdr = read_method_frame (Tuple2.curry identity) (IO.input_string data) in<commit_msg>Removed unused code<commit_after>open Batteries
open Amqp_types

exception Unknown_frame of int
exception Decode_error

let protocol_header = "AMQP\x00\x00\x09\x01"

module Transport = struct
  type t = { i: IO.input; o: unit IO.output; }

  let connect () =
    let i, o = Unix.(open_connection ~autoclose:false (ADDR_INET (inet_addr_loopback, 5672))) in
    IO.output o protocol_header 0 (String.length protocol_header) |> ignore;
    IO.flush o;
    { i; o; }
end

module Framing = struct

  (* Imperative. We will change that later *)
  type hdr = (int * int)
  type channel_id = int

  type channel_state =
    | Ready
    | Waiting of hdr * int * Buffer.t

  type message =
    | Method of string
    | Body of string

  type callback = hdr -> message -> unit

  type channel = {  callback: callback;
                    mutable state: channel_state;
                 }

  let channels : (int, channel) Hashtbl.t = Hashtbl.create 0

  let frame_end = 0xce

  let frame          = Octet :: Short :: Longstr :: Octet :: Nil
  let read_frame     = read frame
  let write_frame    = write frame

  let method_frame   = Short :: Short :: Nil
  let content_header = Short :: Short :: Longlong :: Short :: Nil

  let read_method_frame = read method_frame
  let write_method_frame : (string IO.output -> 'a)  = write method_frame

  let read_content_header = read content_header
  let write_content_header  : (string IO.output -> 'a) = write content_header

  let write output channel ((cid, mid), message) =
    let (data, tpe) = match message with
      | Method data -> data, Amqp_spec.frame_method
      | Body data -> data, Amqp_spec.frame_header (* TODO *)
    in
    let hdr_data =
      write_method_frame (IO.output_string ()) cid mid
      |> IO.close_out
    in
    write_frame output tpe channel (hdr_data ^ data) frame_end
    |> IO.flush

  let read input =
    let (tpe, channel_id, data, magic) = read_frame (Tuple4.curry identity) input in
    assert (magic = frame_end);
    Printf.printf "Channel: %d\n%!" channel_id;
    let channel = Hashtbl.find channels channel_id in
    match tpe with
    | n when n = Amqp_spec.frame_method ->
      (* Standard method message *)
      assert (channel.state = Ready);
      let hdr = read_method_frame (Tuple2.curry identity) (IO.input_string data) in
{'fraction_non_alphanumeric': 0.09715447154471545, 'fraction_numerical': 0.007520325203252033, 'mean_word_length': 2.890118577075099, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 28, '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': '3908306', 'n_tokens_mistral': 1635, 'n_tokens_neox': 1559, 'n_words': 503}