(* O'Caml high level Freenet client module                          *)
(* by Travis Bemann                                                 *)
(*                                                                  *)
(* This program is free software; you can redistribute it and/or    *)
(* modify it under the terms of the GNU Lesser General Public       *)
(* License as published by the Free Software Foundation; either     *)
(* version 2 of the License, or (at your option) any later version. *)
(*                                                                  *)
(* This program is distributed in the hope that it will be useful,  *)
(* but WITHOUT ANY WARRANTY; without even the implied warranty of   *)
(* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the    *)
(* GNU Lesser General Public License for more details.              *)

open Fstream
open Fieldset

(* Every error condition you can think of *)
exception Bad_control of string
exception Future_dbr
exception Block_missing of string * exn
exception Part_not_found of string
exception Default_not_found
exception Insert_map_part_uri
exception Bad_map_control

(* Control metadata representation datastructures *)
type control_date_redirect =
    { dbr_target : string;
      dbr_offset : int;
      dbr_increment : int }
      
type control_split_file =
    { spl_length : int;
      spl_blocks : string list;
      spl_blocks_check : string list }
      
type control =
    Redirect of string
  | Date_redirect of control_date_redirect
  | Split_file of control_split_file
  | No_control
      
(* Map(file) metadata representation datastructures *)
type control_info = control * fieldset
      
type map =
    { map_items : (string, control_info) Hashtbl.t;
      map_default : string option }

(* The length of the day in seconds *)
let day_len = 86400

(* All the node info and parameters you can think of; makes life *)
(* easy by keeping them all together and keeping them from all *)
(* having to be passed to functions as separate parameters *)
type node_info =
    { nin_node : Fcp.node;
      nin_addr : string;
      nin_port : int;
      nin_block_len : int;
      nin_threads_max : int;
      nin_attempts_max : int;
      nin_dnf_retry_htl_mult : float;
      nin_splitfile_block_len_try : int }

(* Construct a map from an association list of map part names and *)
(* control datastructure and info metadata pairs and a default *)
(* option. *)
let make_map parts default =
  let map = { map_items = Hashtbl.create (List.length parts);
	      map_default = default } in
  List.iter (fun (name, item) -> Hashtbl.add map.map_items name item)
    parts;
  begin match default with
    Some name ->
      begin try ignore (Hashtbl.find map.map_items name)
      with Not_found -> raise (Part_not_found name)
      end
  | None -> ()
  end;
  map

(* Convert an integer to a hexadecimal representation *)
let hex_of_int n = Printf.sprintf "%x" n

(* Parse date based redirect control metadata *)
let parse_date_redirect metadata =
 let target =
   try string_of_field (field_of_name metadata "Target")
   with Not_found | Failure _ ->
     raise (Bad_control "No date redirect target")
 and offset =
   try int_of_field (field_of_name metadata "Offset")
   with
     Not_found -> 0
   | Failure _ -> raise (Bad_control "Invalid date redirect offset")
 and increment =
   try int_of_field (field_of_name metadata "Increment")
   with
     Not_found -> day_len
   | Failure _ ->
       raise (Bad_control "Invalid date redirect increment")
 in Date_redirect { dbr_target = target;
		    dbr_offset = offset;
		    dbr_increment = increment }

(* Parse splitfile control metadata *)
let parse_splitfile metadata =
  let size =
    try int_of_field (field_of_name metadata "Size")
    with
      Not_found -> raise (Bad_control "Missing splitfile size")
    | Failure _ -> raise (Bad_control "Invalid splitfile size")
  and count =
    try int_of_field (field_of_name metadata "BlockCount")
    with
      Not_found -> raise (Bad_control "Missing splitfile block count")
    | Failure _ -> raise (Bad_control "Invalid splitfile block count")
  and count_check =
    try int_of_field (field_of_name metadata "CheckBlockCount")
    with
      Not_found -> 0
    | Failure _ ->
	raise (Bad_control "Invalid splitfile check block count")
  and blocks =
    try fieldset_of_field (field_of_name metadata "Block")
    with Not_found | Failure _ ->
      raise (Bad_control "Missing splitfile blocks")
  and blocks_check =
    try fieldset_of_field (field_of_name metadata "CheckBlock")
    with Not_found | Failure _ -> []
  in
  let rec get_blocks metadata index blocks =
    if index > 0 then
      let block =
	try string_of_field (field_of_name metadata
			       (hex_of_int index))
	with Not_found | Failure _ ->
	  raise (Bad_control "Missing splitfile blocks")
      in get_blocks metadata (index - 1) (block :: blocks)
    else blocks
  in Split_file { spl_length = size;
		 spl_blocks = get_blocks blocks count [];
		 spl_blocks_check =
		   get_blocks blocks_check count_check [] }

(* Parse control metadata *)
let control_of_fieldset metadata =
  try Redirect
      (string_of_field (field_of_path metadata "Redirect.Target"))
  with Not_found | Failure _ ->
    try parse_date_redirect
	(fieldset_of_field (field_of_name metadata "DateRedirect"))
    with Not_found | Failure _ ->
      try parse_splitfile
	  (fieldset_of_field (field_of_name metadata "SplitFile"))
      with Not_found | Failure _ -> No_control

(* Convert an integer into a valid field value. *)
let field_val_of_int n = Field_string (hex_of_int n)

(* Generate control metadata from a control datastructure *)
let fieldset_of_control = function
    Redirect uri ->
      field_make [] "Redirect.Target" (Field_string uri)
  | Date_redirect { dbr_target = target;
		    dbr_offset = offset;
		    dbr_increment = increment } ->
      fields_make []
	[("DateRedirect.Target", Field_string target);
	 ("DateRedirect.Offset", field_val_of_int offset);
	 ("DateRedirect.Increment", field_val_of_int increment)]
  | Split_file { spl_length = size;
		 spl_blocks = blocks;
		 spl_blocks_check = blocks_check } ->
      let rec make_block_fields blocks index fields =
	match blocks with
	  block :: rest ->
	    make_block_fields rest (index + 1)
	      (field_make fields (hex_of_int index)
		 (Field_string block))
	| [] -> fields
      in
      fields_make []
	[("SplitFile.Size", field_val_of_int size);
	 ("SplitFile.BlockCount",
	  field_val_of_int (List.length blocks));
	 ("SplitFile.CheckBlockCount",
	  field_val_of_int (List.length blocks_check));
	 ("SplitFile.Block",
	  Field_fieldset (make_block_fields blocks 1 []));
	 ("SplitFile.CheckBlock",
	  Field_fieldset (make_block_fields blocks_check 1 []))]
  | No_control -> []      

(* Auxiliary function to get a line within a string by index and *)
(* return the line without the newline and the index of the start *)
(* of the next line *)
let get_line data index =
  let len = String.length data in
  let rec step index =
    if index < len then
      match String.get data index with
	'\n' -> (index, index + 1)
      | '\r' ->
	  if index + 1 < len then
	    match String.get data (index + 1) with
	      '\n' -> (index, index + 2)
	    | _ -> (index, index + 1)
	  else (index, index + 1)
      | _ -> step (index + 1)
    else (index, index)
  in
  let end_line, next_line = step index in
  (String.sub data index (end_line - index), next_line)

(* Auxiliary function to create a new empty part in a list of *)
(* metadata parts only if one is needed *)
let part_new = function
    [] :: [] -> [] :: []
  | x :: y as parts -> [] :: parts
  | [] -> [] :: []

(* Auxiliary function to add a line to the metadata part at *)
(* the start of a list of metadata parts *)
let part_add parts item =
  match parts with
    part :: rest -> (item :: part) :: rest
  | [] -> assert false
		 
(* Parse raw metadata info a list of metadata parts each consisting *)
(* individually of a list of metadata lines *)
let separate_parts raw =
  let len = String.length raw in
  let rec step_base index parts =
    if index < len then
      let line, next = get_line raw index in
      match Strutil.strip line with
	"Version" -> step_version next parts
      | "Document" -> step_part next (part_new parts)
      | "End" -> parts
      | _ -> step_base next parts
    else parts
  and step_part index parts =
    if index < len then
      let line, next = get_line raw index in
      match Strutil.strip line with
	"EndPart" -> step_base next parts
      | "End" -> parts
      | "" -> step_part next parts
      | line -> step_part next (part_add parts line)
    else parts
  and step_version index parts =
    if index < len then
      let line, next = get_line raw index in
      match Strutil.strip line with
	"EndPart" -> step_base next parts
      | "End" -> parts
      | _ -> step_version next parts
    else parts
  in step_base 0 []

(* Turn raw metadata into a list of metadata parts *)
let parse_metadata_raw raw =
  List.rev_map (fun part -> field_list_parse part)
    (separate_parts raw)

(* Turn a list of metadata parts into a map(file) datastructure *)
let map_of_parts parts =
  let pairs =
    let part_to_pair = fun metadata -> 
      let name =
	try string_of_field (field_of_name metadata "Name")
	with Not_found | Failure _ -> ""
      and info =
	try fieldset_of_field (field_of_name metadata "Info")
	with Not_found | Failure _ -> []
      and control = control_of_fieldset metadata in
      (name, (control, info)) in
    List.rev_map part_to_pair parts
  in
  let hash = Hashtbl.create (List.length pairs) in
  List.iter (fun (name, item) -> Hashtbl.add hash name item) pairs;
  { map_items = hash; map_default = None }

(* Turn raw metadata into a map(file) datastructure *)
let map_of_metadata_raw raw = map_of_parts (parse_metadata_raw raw)

(* Turn a control datastructure and info metadata pair into a *)
(* metadata part *)
let fieldset_of_control_info (control, info) =
  fieldset_list_merge [fieldset_of_control control;
		       field_make [] "Info" (Field_fieldset info)]

(* Turn a map(file) datastructure into raw metadata *)
let metadata_raw_of_map map =
  let buf = Buffer.create 0 in
  Buffer.add_string buf "Version\nRevision=1\n";
  let items = Hashtbl.copy map.map_items in
  begin match map.map_default with
    Some name ->
      begin try Hashtbl.add items "" (Hashtbl.find items name)
      with Not_found -> raise Default_not_found
      end
  | None -> ()
  end;
  Hashtbl.iter
    begin fun name item ->
      let fields =
	fieldset_list_merge [fieldset_of_control_info item;
			     field_make [] "Name" (Field_string name)]
      in
      Buffer.add_string buf "EndPart\nDocument\n";
      Buffer.add_string buf (fields_print fields);
    end
    items;
  Buffer.add_string buf "End\n";
  Buffer.contents buf

(* Get a control info pair out of a map(file) datastructure by name *)
let map_part map part = Hashtbl.find map.map_items part

(* The regexp for splitting URIs into an actual URI and a part name *)
let split_uri_regexp = Str.regexp_string "//"

(* Split a URI into an actual URI and a part name; if no part name *)
(* is specified then the part name is returned as "" *)
let split_uri uri =
  match Str.bounded_split_delim split_uri_regexp uri 2 with
    [uri; part] -> (uri, part)
  | uri :: [] -> (uri, "")
  | [] -> raise (Fcp.Uri_error uri)
  | _ -> assert false

(* Strip the freenet: protocol indicator from a URI. *)
let uri_strip_proto uri =
  if String.length uri <= 8 then uri
  else if String.sub uri 0 8 <> "freenet:" then uri
  else String.sub uri 8 ((String.length uri) - 8)

(* Split the key type and remainder of a URI. *)
let uri_split_type uri =
  let suri = uri_strip_proto uri in
  if String.length suri > 4 then
    if String.get suri 3 = '@' then
      (String.sub suri 0 3,
       String.sub suri 4 ((String.length suri) - 4))
    else ("KSK", suri)
  else ("KSK", suri)

(* Generate a date based redirect URI from a target URI and an item *)
(* in seconds. *)
let generate_dbr_uri ~uri ~item =
  let key_type, key_data = uri_split_type uri in
  match key_type with
    "SSK" ->
      let sub, name =
	try Strutil.split2 key_data '/'
	with Not_found -> raise (Fcp.Uri_error uri)
      in Printf.sprintf "freenet:SSK@%s/%x-%s" sub item name
  | "KSK" -> Printf.sprintf "freenet:KSK@%x-%s" item key_data
  | _ ->
      raise (Bad_control "Date-based redirects must be SSKs or KSKs")

(* Generate a date based redirect URI from a target URI, increment in*)
(* seconds, and index in increments. *)
let generate_dbr_uri_item ~uri ~increment ~index =
  generate_dbr_uri uri (increment * index)

(* Generate a date based redirect URI relative to the present, from *)
(* a target URI, offset in seconds, increment in seconds, and index *)
(* in increments. *)
let generate_dbr_uri_relative ~uri ~offset ~increment ~index =
  let time = int_of_float (Unix.time ()) in
  if offset + (increment * index) < time then
    let item = (((time - offset) / increment) + index) * increment in
    generate_dbr_uri uri item
  else raise Future_dbr

(* Generate a date based redirect URI relative to the present by a *)
(* single increment, from a target URI, offset in seconds, and *)
(* increment in seconds. *)
let generate_dbr_uri_next ~uri ~offset ~increment =
  generate_dbr_uri_relative uri offset increment 1

(* Generate a date based redirect URI from a date based redirect *)
(* datastructure and an index from the present. *)
let expand_dbr dbr index =
  generate_dbr_uri_relative dbr.dbr_target dbr.dbr_offset
    dbr.dbr_increment index

(* Retry attempting to request a file until the maximum number of *)
(* attempts are done; handle increasing HTL to deal with *)
(* DataNotFound and such; if a request is successful, return the *)
(* request transaction, if it isn't, return the exact last *)
(* exception raised. *)
let request_retry node_info uri htl =
  let rec step attempts htl except =
    if attempts < node_info.nin_attempts_max then
      try node_info.nin_node#request uri htl
      with
	Fcp.Data_not_found as except ->
	  step (attempts + 1)
	    (int_of_float ((float_of_int htl) *.
			     node_info.nin_dnf_retry_htl_mult))
	    except
      | Fcp.Route_not_found _ as except ->
	  step (attempts + 1) htl except
    else raise except
  in step 0 htl Fcp.Finished

(* Request the raw map(file) datastructure of a file at a specified *)
(* URI; note that if a part indicator is used then the file that *)
(* the proper map(file) part points to will be used *as a whole*, *)
(* while if no map(file) part indicator is used then the entire *)
(* file's map(file) datastructure willl be returned, for ALL parts. *)
let rec request_map_raw node_info uri htl =
  let uri, part = split_uri uri in
  let transact = request_retry node_info uri htl in
  let map = map_of_metadata_raw transact#recv_metadata in
  if part = "" then
    (map, transact)
  else
    let control, _ =
      try map_part map part
      with Not_found -> raise Bad_map_control
    in
    transact#cancel;
    match control with
      Redirect uri -> request_map_raw node_info uri htl
    | Date_redirect dbr ->
	request_map_raw node_info (expand_dbr dbr 0) htl
    | Split_file _ -> raise Bad_map_control
    | No_control -> raise Bad_map_control

(* Request a specified map(file) part, handling map(file) redirects *)
(* and such in the process; tries to be smart about figuring out *)
(* whether redirects and such refer to the map(file) as a whole OR *)
(* just to an individual map(file) part. *)
let rec request_map node_info uri htl part =
  let uri, part_extra = split_uri uri in
  let map, transact = request_map_raw node_info uri htl in
  try (map_part map part, transact)
  with
    Not_found ->
      let control, _ =
	try map_part map part_extra
	with Not_found -> raise (Part_not_found part)
      in
      transact#cancel;
      match control with
	Redirect uri -> request_map node_info uri htl part
      | Date_redirect dbr ->
	  request_map node_info (expand_dbr dbr 0) htl part
      | Split_file _ -> raise (Part_not_found part)
      | No_control -> raise (Part_not_found part)

(* Request solely the raw control datastructure and info metadata *)
(* pair of a specified map(file) part, WITHOUT following any sort *)
(* of redirection. *)
let request_map_raw_part node_info uri htl =
  let uri, part = split_uri uri in
  let map, transact = request_map_raw node_info uri htl in
  try map_part map part
  with Not_found ->
    transact#cancel;
    raise (Part_not_found part)

(* Write the contents of a request transaction to an input fstream *)
let transact_to_stream transact stream =
  let metadata_skip () =
    let _ = transact#recv_metadata in
    stream#output_string transact#recv_data_prev
  in
  let rec step () =
    if transact#read_len < transact#len then
      try
	stream#output_string transact#recv_block;
	step ()
      with
	Fcp.Finished -> ()
      | Fcp.Restarted ->
	  stream#clear;
	  try
	    metadata_skip ();
	    step ()
	  with Fcp.Finished -> ()
    else ()
  in
  try
    metadata_skip ();
    step ()
  with Fcp.Finished -> ()
    
(* The high level request routine that will be used for most *)
(* purposes. *)
let rec request node_info uri htl stream =
  let uri, part = split_uri uri in
  let (control, info), transact =
    request_map node_info uri htl part in
  match control with
    Redirect uri -> request node_info uri htl stream
  | Date_redirect dbr ->
      request node_info (expand_dbr dbr 0) htl stream
  | Split_file split ->
      request_splitfile node_info split htl stream;
      info
  | No_control ->
      transact_to_stream transact stream;
      info

(* Request a splitfile; note that this does properly handle *)
(* multithreading of requests for each individual splitfile block. *)
and request_splitfile node_info split htl stream =
(* All the mutexes for making interthread communication safe *)
  let data_mutex = Mutex.create ()
  and error_mutex = Mutex.create ()
  and threads_mutex = Mutex.create ()
(* All the modifiable objects used by multiple threads *)
  and data = Array.make (List.length split.spl_blocks) None
  and error = ref None
  and threads = ref 0 in
(* Actually request a block within a separate thread and return the *)
(* results, either a block of data OR an exception, to the *)
(* controlling thread *)
  let request_block block index =
    Mutex.lock threads_mutex;
    threads := !threads + 1;
    Mutex.unlock threads_mutex;
    let request_block_thread () =
      try
	let buf = new fstream_out_buffer in
	ignore (request node_info block htl (buf :> fstream_out));
	Mutex.lock data_mutex;
	Mutex.lock threads_mutex;
	data.(index) <- Some buf#contents;
	threads := !threads - 1;
	Mutex.unlock data_mutex;
	Mutex.unlock threads_mutex
      with except ->
	Mutex.lock error_mutex;
	Mutex.lock threads_mutex;
	error := Some (Block_missing (block, except));
	threads := !threads - 1;
	Mutex.unlock error_mutex;
	Mutex.unlock threads_mutex
    in ignore (Thread.create request_block_thread ())
(* A bunch of utility functions used in multiple places by the *)
(* controlling thread *)
  and check_data index =
    Mutex.lock data_mutex;
    let item = data.(index) in
    Mutex.unlock data_mutex;
    item
  and clear_data index =
    Mutex.lock data_mutex;
    data.(index) <- None;
    Mutex.unlock data_mutex
  and check_error () =
    Mutex.lock error_mutex;
    let error = !error in
    Mutex.unlock error_mutex;
    error
  and check_threads () =
    Mutex.lock threads_mutex;
    let threads = !threads in
    Mutex.unlock threads_mutex;
    threads
  in
(* Control the actual block requesting threads, creating new ones *)
(* when there is less than a maximum number of block request *)
(* threads running simultaneously, writing blocks returned in data *)
(* by threads to the output fstream, and catching exceptions *)
(* signalled by the block request threads *)
  let rec dispatch blocks block_index output_index =
    begin match check_error () with
      None -> ()
    | Some except -> raise except
    end;
    let output_index =
      match check_data output_index with
	None -> output_index
      | Some block ->
	  clear_data output_index;
	  stream#output_string block;
	  output_index + 1
    in
    if check_threads () < node_info.nin_threads_max then
      match blocks with
	block :: rest ->
	  request_block block block_index;
	  dispatch rest (block_index + 1) output_index
      | [] -> wait output_index
    else begin
      Thread.yield ();
      dispatch blocks block_index output_index
    end
  and wait output_index =
    if output_index < List.length split.spl_blocks then
      match check_error () with
	None ->
	  begin match check_data output_index with
	    None -> wait output_index
	  | Some block ->
	      clear_data output_index;
	      stream#output_string block;
	      wait (output_index + 1)
	  end
      | Some except -> raise except
    else ()
  in
  dispatch split.spl_blocks 0 0

(* Try to insert a file by URI until number of attempts equals *)
(* maximum number of attempts, and then raise the *exact* last *)
(* exception raised. *)
let insert_retry node_info uri htl metadata_len data_len =
  let rec step attempts except =
    if attempts < node_info.nin_attempts_max then
      try node_info.nin_node#insert uri htl metadata_len data_len
      with Fcp.Route_not_found _ as except ->
	step (attempts + 1) (Some except)
    else match except with
      Some except -> raise except
    | None -> assert false
  in step 0 None

(* The maximum size of a non-CHK file in Freenet *)
let non_chk_len_max = 32678

(* The basis for all practical insertion within Hlfreenet; note *)
(* that this WILL complain if you try to insert a URI containing *)
(* a part indicator. *)
let rec insert node_info uri htl map stream =
  match split_uri uri with
    (_, "") ->
      let metadata = metadata_raw_of_map map in
      let map, stream =
	if ((String.length metadata) + stream#length
	      > non_chk_len_max) && (uri <> "CHK@")
	then begin
	  let chk = insert_chk node_info htl map stream in
	  ((make_map [("", (Redirect chk, []))] None),
	   (new fstream_in_string ""))
	end else (map, stream)
      in
      let transact = insert_retry node_info uri htl
	  (String.length metadata) stream#length in
      transact#metadata_block metadata;
      let buf = String.create node_info.nin_block_len in
      let rec step () =
	try
	  let len = stream#input_buf buf 0 node_info.nin_block_len in
	  if len > 0 then begin
	    if len = node_info.nin_block_len then
	      transact#block buf
	    else transact#block (String.sub buf 0 len);
	    step ()
	  end else ()
	with
	  End_of_file -> ()
	| except ->
	    transact#cancel;
	    raise except
      in
      step ();
      transact#wait_all
  | (_, _) -> raise Insert_map_part_uri

(* Insert a file as a CHK *)
and insert_chk node_info htl control_info stream =
  try insert node_info "CHK@" htl control_info stream
  with Fcp.Key_collision uri -> uri

(* Utility function for getting the log2 of integers and returning *)
(* it as an integer *)
let log2_int n =
  int_of_float ((log (float_of_int n)) /. (log 2.0))

(* Insert the individual parts for a splitfile but NOT the *)
(* splitfile itself (which is actually the splitfile metadata *)
(* and may actually just be an item in a map(file), for instance), *)
(* but rather return the generated control datastructure *)
let insert_splitfile_blocks node_info htl (stream : fstream_in) =
  let data_len = stream#length in
(* Determine the splitfile block length to use; note that this has *)
(* the function of mucking with the block length to make it a power *)
(* of 2 *)
  let block_len =
    if 1 lsl (log2_int node_info.nin_splitfile_block_len_try)
	<> node_info.nin_splitfile_block_len_try then
      2 lsl (log2_int node_info.nin_splitfile_block_len_try)
    else node_info.nin_splitfile_block_len_try
  in
  let block_count =
    if data_len mod block_len = 0 then
      data_len / block_len
    else (data_len / block_len) + 1
  in
(* All the mutexes for making interthread communication safe *)
  let chks_mutex = Mutex.create ()
  and error_mutex = Mutex.create ()
  and threads_mutex = Mutex.create ()
(* All the modifiable objects used by multiple threads *)
  and chks = Array.make block_count None
  and error = ref None
  and threads = ref 0 in
(* Insert a block as a CHK in a separate thread, and signal back *)
(* either the actual CHK or an exception *)
  let insert_block index data =
    Mutex.lock threads_mutex;
    threads := !threads + 1;
    Mutex.unlock threads_mutex;
    let insert_block_thread () =
      try
	let stream = new fstream_in_string data in
	let chk = insert_chk node_info htl
	    (make_map [("", (No_control, []))] None) stream in
	Mutex.lock chks_mutex;
	Mutex.lock threads_mutex;
	chks.(index) <- Some chk;
	threads := !threads - 1;
	Mutex.unlock chks_mutex;
	Mutex.unlock threads_mutex
      with except ->
	Mutex.lock error_mutex;
	Mutex.lock threads_mutex;
	error := Some except;
	threads := !threads - 1;
	Mutex.unlock error_mutex;
	Mutex.unlock threads_mutex
    in ignore (Thread.create insert_block_thread ())
(* A bunch of utility functions used in multiple places by the *)
(* controlling thread *)
  and check_error () =
    Mutex.lock error_mutex;
    let error = !error in
    Mutex.unlock error_mutex;
    error
  and check_threads () =
    Mutex.lock threads_mutex;
    let threads = !threads in
    Mutex.unlock threads_mutex;
    threads
  in
(* Create splitfile block insertion threads, catch any exceptions *)
(* signalled by a splitfile block insertion thread, and once done *)
(* spawning all the splitfile block insertion threads, and wait for *)
(* all insertions to complete. *)
  let rec dispatch index =
    begin match check_error () with
      None -> ()
    | Some except -> raise except
    end;
    if index < block_count then
      if check_threads () < node_info.nin_threads_max then
	let len =
	  if data_len - stream#pos >= block_len then
	    block_len
	  else data_len - stream#pos
	in
	insert_block index (stream#really_input len);
	dispatch (index + 1)
      else begin
	Thread.yield ();
	dispatch index
      end
    else wait ()
  and wait () =
    begin match check_error () with
      None -> ()
    | Some except -> raise except
    end;
    if check_threads () > 0 then begin
      Thread.yield ();
      wait ()
    end else match check_error () with
      None -> ()
    | Some except -> raise except
  in
  dispatch 0;
(* Actually generate the splitfile datastructure *)
  { spl_length = data_len;
    spl_blocks =
      List.map
        begin function
	    Some chk -> chk
	  | None -> assert false
	end
        (Array.to_list chks);
    spl_blocks_check = [] }

(* Insert an actual splitfile's parts and the file itself; note *)
(* that this is not what you want if the splitfile is to be an *)
(* item in a map(file); then you just want the map(file) part *)
(* referring to the splitfile containing the control datastructure *)
(* returned by insert_splitfile_blocks. *)
let insert_splitfile node_info uri htl info stream =
  let control = insert_splitfile_blocks node_info htl stream in
  insert node_info uri htl
    (make_map [("", (Split_file control, info))] None)
    (new fstream_in_string "")

(* Same as insert_splitfile except that a CHK is inserted and *)
(* Fcp.Key_collision is caught so it is ignored. *)
let insert_splitfile_chk node_info htl info stream =
  try insert_splitfile node_info "CHK@" htl info stream
  with Fcp.Key_collision uri -> uri

(* Generate a CHK from a stream. *)
let generate_chk node_info map (stream : fstream_in) =
  let metadata = metadata_raw_of_map map in
  let transact = node_info.nin_node#generate_chk
      ~metadata_len:(String.length metadata) ~data_len:stream#length
  in
  transact#metadata_block metadata;
  let buf = String.create node_info.nin_block_len in
  let rec step () =
    try
      let len = stream#input_buf buf 0 node_info.nin_block_len in
      if len > 0 then begin
	if len = node_info.nin_block_len then
	  transact#block buf
	else transact#block (String.sub buf 0 len);
	step ()
      end else ()
    with
      End_of_file -> ()
    | except ->
	transact#cancel;
	raise except
  in
  step ();
  transact#generate_chk

(* This class basically forms the primary user interfaces to *)
(* Hlfreenet, but does NOT encapsulate a lot of its innards (which *)
(* are encapsulated using the OCaml module system instead.) *)
class node_hl ~addr ~port ~threads_max ~block_len ~attempts_max
   ~dnf_retry_htl_mult ~splitfile_block_len_try =
  object (self)
    inherit Fcp.node addr port

(* Private method to generate node info datastructure *)
    method private info =
      { nin_node = (self :> Fcp.node);
        nin_addr = addr;
	nin_port = port;
	nin_block_len = block_len;
	nin_threads_max = threads_max;
	nin_attempts_max = attempts_max;
	nin_dnf_retry_htl_mult = dnf_retry_htl_mult;
	nin_splitfile_block_len_try =
	splitfile_block_len_try }

(* Simple accessor methods *)
    method threads_max = threads_max
    method block_len = block_len
    method attempts_max = attempts_max
    method dnf_retry_htl_mult = dnf_retry_htl_mult
    method splitfile_block_len_try = splitfile_block_len_try

(* Do a high level request, return the resulting info metadata. *)
    method request_hl ~uri ~htl ~stream =
      request self#info uri htl stream

(* Do a high level request, return the resulting control *)
(* datastructure and info metadata pair; do not follow control *)
(* metadata. *)
    method request_hl_control_info ~uri ~htl =
      request_map_raw_part self#info uri htl

(* Do a high level request, return the following map; do not follow *)
(* control metadata. *)
    method request_hl_map ~uri ~htl =
      let map, _ = request_map_raw self#info uri htl in map

(* Do a high level insert of a file with a map datastructure. *)
    method insert_hl ~uri ~htl ~map ~stream =
      insert self#info uri htl map stream

(* Do a high level insert of a file with a control datastructure *)
(* and info metadata pair. *)
    method insert_hl_no_map ~uri ~htl ~control_info ~stream =
      insert self#info uri htl (make_map [("", control_info)] None)
	stream

(* Do a high level insert of a file with info metadata. *)
    method insert_hl_info ~uri ~htl ~info ~stream =
      insert self#info uri htl
	(make_map [("", (No_control, info))] None) stream

(* Do a high level insert of a file as a CHK with a map *)
(* datastructure. *)
    method insert_hl_chk ~htl ~map ~stream =
      insert_chk self#info htl map stream

(* Do a high level insert of a file as a CHK with a control *)
(* datastructure and info metadata pair. *)
    method insert_hl_chk_no_map ~htl ~control_info ~stream =
      insert_chk self#info htl (make_map [("", control_info)] None)
	stream

(* Do a high level insert of a file as a CHK with info metadata. *)
    method insert_hl_chk_info ~htl ~info ~stream =
      insert_chk self#info htl
	(make_map [("", (No_control, info))] None) stream

(* Do a high level insert of a file as a group of splitfile blocks *)
(* and return the resulting splitfile datastructure. *)
    method insert_hl_splitfile_blocks ~htl ~stream =
      insert_splitfile_blocks self#info htl stream

(* Do a high level insert of a file as a splitfile and insert the *)
(* corresponding splitfile control metadata and info metadata at *)
(* a specified URI. *)
    method insert_hl_splitfile ~uri ~htl ~info ~stream =
      insert_splitfile self#info uri htl info stream

(* Do a high level insert of a file as a splitfile in a CHK and *)
(* insert the corresponding splitfile control metadata and info *)
(* metadata at a specified URI, while catching Fcp.Key_collision *)
(* so it is ignored. *)
    method insert_hl_splitfile_chk ~htl ~info ~stream =
      insert_splitfile_chk self#info htl info stream

(* Do a high level CHK generation from a stream and map *)
(* datastructure. *)
    method generate_chk_hl ~map ~stream =
      generate_chk self#info map stream

(* Do a high level CHK generation from a stream and control *)
(* datastructure and info metadata pair. *)
    method generate_chk_hl_no_map ~control_info ~stream =
      generate_chk self#info (make_map [("", control_info)] None)
	stream

(* Do a high level CHK generation from a stream and info metadata. *)
    method generate_chk_hl_info ~info ~stream =
      generate_chk self#info
	(make_map [("", (No_control, info))] None) stream
  end