(* $Id: http_client.ml 128 2004-08-01 11:48:03Z gerd $
 * ----------------------------------------------------------------------
 *
 *)

(* Reference documents:
 * RFC 2068, 2616:      HTTP 1.1
 * RFC 2069, 2617:      Digest Authentication
 *)


(* TODO:
 * - Minor problem: if there is no user/password given for a realm, the client
 *   initiates a second request although this request is senseless.
 *)

open Unix

let listrm = List.remove_assoc;;
  (* This is Ocaml-2.02. In 2.01, use List.remove *)

exception Header_is_incomplete;;
exception Body_is_incomplete;;
exception Body_maybe_complete;;

exception Bad_message of string;;

exception Http_error of (int * string);;
exception Http_protocol of exn;;
exception No_reply;;

(* Internally used exceptions: *)

exception No_exception;;    (* serves as a guard *)
exception Abort_on_protocol_error;;
exception Broken_pipe;;


type auth_method =
    Auth_basic of string                                (* realm *)
  | Auth_digest of (string * string * string * string)  (* realm, nonce,
							 * opaque, stale *)

type token =
    Word of string
  | Special of char             (* tspecial or CTL, but not double quotes *)
;;


type secret = unit

type 'message_class how_to_reconnect =
    Send_again         (* Send the request automatically again *)
  | Request_fails      (* Drop the request *)
  | Inquire of ('message_class -> bool)
                       (* If the function return 'true' send again, otherwise
			* drop the request.
			*)
;;

type 'message_class how_to_redirect =
    Redirect           (* Perform the redirection *)
  | Do_not_redirect    (* No redirection *)
  | Redirect_inquire of ('message_class -> bool)
                       (* If the function return 'true' redirect, otherwise
			* do not redirect
			*)
;;


let better_unix_error f arg =
  try
    f arg
  with
    Unix_error (e,syscall,param) ->
      let error = error_message e in
      if param = "" then
	failwith error
      else
	failwith (param ^ ": " ^ error)


let rec syscall f =
  (* Invoke system call, and handle EINTR *)
  try
    f()
  with
      Unix.Unix_error(Unix.EINTR,_,_) ->
	(* "interrupted system call": A signal happened while the system
	 * blocked.
	 * Simply restart the call.
	 *)
	syscall f
;;


let string_of_token_list l =           (* for verbosity *)
  String.concat ", "
    (List.map
       (function 
	    Word w -> w
	  | Special c -> "'" ^ String.make 1 c ^ "'")
       l)
;;

(* Parsing suitable for most header lines. See RFC 2068. *)

let parse_header_value v =
  let is_control c = let code = Char.code (c) in code < 32 || code = 127 in
  let is_special c = (c = '('  || c = ')' || c = '<' || c = '>' or
		      c = '@'  || c = ',' || c = ';' || c = ':' or
		      c = '\\' || c = '"' || c = '/' || c = '[' or
		      c = ']'  || c = '?' || c = '{' || c = '}' or
		      c = '\t' || c = ' ' || c = '=') in
  let l = String.length v in

  let rec parse_quoted k word =
    if k < l then begin
      let c = v.[k] in
      if c='"' then
	k, word
      else
	if c='\\' then begin
	  if k < l-1 then
	    parse_quoted (k+2) (word ^ String.make 1 (v.[k+1]))
	  else
	    failwith "http_client: cannot parse header line"
	end
	else
	  parse_quoted (k+1) (word ^ String.make 1 c)
    end
    else
      failwith "http_client: cannot parse header line"
  in

  let rec parse k word =
    if k < l then begin
      let c = v.[k] in
      if c <> '"' && (is_control c || is_special c) then begin
	if (c = ' ' || c = '\t') then
	  (if word <> "" then [Word word] else []) @
	  parse (k+1) ""
	else
	  (if word <> "" then [Word word] else []) @
	  [Special c] @
	  parse (k+1) ""
      end
      else
	if c = '"' then begin
	  let k', word' = parse_quoted (k+1) "" in
	    (if word <> "" then [Word word] else []) @
	    [Word word'] @
	    parse (k'+1) ""
	end
	else
	  parse (k+1) (word ^ String.make 1 c)
    end
    else []

  in
  let p = parse 0 "" in
  (*
    List.iter
    (function Word s -> prerr_endline ("Word " ^ s) |
              Special c -> prerr_endline ("Special " ^ String.make 1 c))
    p;
   *)
  p
;;


let hex_digits = [| '0'; '1'; '2'; '3'; '4'; '5'; '6'; '7';
		    '8'; '9'; 'a'; 'b'; 'c'; 'd'; 'e'; 'f' |];;

let encode_hex s =
  (* encode with lowercase hex digits *)
  let l = String.length s in
  let t = String.make (2*l) ' ' in
  for n = 0 to l - 1 do
    let x = Char.code s.[n] in
    t.[2*n]   <- hex_digits.( x lsr 4 );
    t.[2*n+1] <- hex_digits.( x land 15 );
  done;
  t
;;


type synchronization =
    Sync_with_handshake_before_request_body of float
  | Sync
  | Pipeline of int
;;


type http_options = 
    { synchronization : synchronization;
      maximum_connection_failures : int;
      maximum_message_errors : int;
      inhibit_persistency : bool;
      connection_timeout : float;
      number_of_parallel_connections : int;
      verbose_status : bool;
      verbose_request_header : bool;
      verbose_response_header : bool;
      verbose_request_contents : bool;
      verbose_response_contents : bool;
      verbose_connection : bool;
    }
;;

(**********************************************************************)
(***                          BUFFERS                               ***)
(**********************************************************************)

(* Similar to Buffer, but may be accessed in a more efficient way. *)

module B = Netbuffer;;

type buffer_type = B.t;;


(**********************************************************************)
(***                  THE MESSAGE CONTAINER                         ***)
(**********************************************************************)

(* The class 'message' represents the logical view on messages. A message
 * consists of
 * - a set of header attributes and a message body intended
 *   to be sent to the server (the request)
 * - a set of header attributes and a message body returned from the
 *   server (the reply)
 * 
 * The class has methods helping to transform the request into a 
 * sequence of octets (bytes) and to interpret a stream of octets as 
 * reply.
 *
 * The class is virtual because an auxiliary method needed to form the
 * request depends on the type of the HTTP request ("HTTP method"). This
 * method, 'prepare', is defined in the subclasses 'get', 'post', 'put' etc.
 *)


class virtual message =
  object (self)
    val mutable host = ""
    val mutable port = 80
    val mutable path = ""
    val mutable request = ""
    val mutable request_uri = ""
    val mutable request_method = ""
    val mutable header = []

    val mutable no_proxy = false

    val mutable using_proxy = false

    val mutable body = ""

    val mutable received = B.create 1
    val mutable received_header = []
    val mutable received_entity = 0     (* where entity begins in 'received' *)
    val mutable received_contents = ""  (* received contents (decoded) *)

    val mutable status_code = 0
    val mutable status_text = ""
    val mutable status_version = ""

    val mutable served = false
    val mutable reconnect_mode = (Request_fails : message how_to_reconnect)
    val mutable redirect_mode = (Do_not_redirect : message how_to_redirect)
    val mutable error_counter = 0
    val mutable error_exception = No_exception

    method virtual prepare : bool -> unit

    method using_proxy = using_proxy
      (* whether this message contacts its peer as a proxy *)

    method is_served = served
    method set_served (_:secret) = served <- true

    method set_unserved (_:secret) = 
      (* Resets 'served', 'error_counter', and 'error_exception' *)
      served <- false;
      error_counter <- 0;
      error_exception <- No_exception

    method get_host() = host
    method get_port() = port

    method get_uri() =
      let http_prefix =
	"http://" ^ host ^ (if port = 80 then "" else ":" ^ string_of_int port)
      in
      match request_uri with
	  "*" ->
	    http_prefix
	| "" ->
	    assert false
	| path when path.[0] = '/' ->
	    http_prefix ^ path
	| full_uri ->
	    full_uri

    method get_req_body() = body
    method get_request (_:secret) = request
    method get_req_header () = header
    method assoc_req_header name =
      List.assoc name header
    method assoc_multi_req_header name =
      List.map snd (List.filter (fun (n,_) -> n = name) header)
    method set_req_header name0 value =
      let name = String.lowercase name0 in
      let rec set l =
	match l with
	  [] -> [name, value]
	|	(n,v) :: l' -> if name = n then (name,value)::l' else (n,v) :: set l'
      in
      header <- set header
    method get_req_uri () = request_uri
    method get_req_method () = request_method

    method get_resp_header () = 
      if error_exception = No_exception then
	received_header
      else
	raise (Http_protocol error_exception)

    method assoc_resp_header name =
      if error_exception = No_exception then
	List.assoc name received_header
      else
	raise (Http_protocol error_exception)

    method assoc_multi_resp_header name =
      if error_exception = No_exception then
	List.map snd (List.filter (fun (n,_) -> n = name) received_header)
      else
	raise (Http_protocol error_exception)

    method get_resp_body () =
      (* raise Http_error if status is not 2xx *)
      if error_exception = No_exception then begin
	let _,code,_ = self # dest_status() in
	if code >= 200 && code < 300 then
	  received_contents
	else
	  raise (Http_error(code, received_contents))
      end
      else
	raise (Http_protocol error_exception)

    method dump_header prefix h =
      List.iter
	(fun (n,v) ->
	  prerr_endline (prefix ^ n ^ ": " ^ v))
	h


    method no_proxy() =
      (* force not to use a proxy for this request *)
      no_proxy <- true

    method is_proxy_allowed() =
      not no_proxy


    method get_reconnect_mode = reconnect_mode

    method set_reconnect_mode m =
      reconnect_mode <- m

    method get_redirect_mode = redirect_mode

    method set_redirect_mode m =
      redirect_mode <- m

    method get_error_counter () = error_counter

    method set_error_counter () n =
      error_counter <- n

    method set_error_exception () x =
      match x with
	  Http_error(_,_) -> assert false   (* not allowed *)
	| _ ->
	    error_exception <- x

    method set_response response p_init_header =
      received <- response;
      error_exception <- No_exception;
      if p_init_header then begin
	status_code <- 0;
	received_header <- [];
	received_entity <- 0;
      end


    method init_query (_:secret) query =
      (* initializer *)
      try
	let h, pt, ph = Http_client_aux.match_query query in
	host <- h;
	port <- pt;
	if String.length ph >= 1 then
	  if ph.[0] <> '/' then
	    failwith "http_client: bad URL";
	path <- ph
      with
	Not_found ->
	  failwith "http_client: bad URL"


    method private set_request use_proxy use_asterisk req =
      self # set_req_header "host" (host ^ ":" ^ string_of_int port);
      (* self # set_req_header "te" "trailers"; *) (* We accept trailers *)
      request_method <- req;
      reconnect_mode <- begin match req with
	  ("GET"|"HEAD") -> Send_again
	| _ -> Request_fails
      end;
      redirect_mode <- begin match req with
	  ("GET"|"HEAD") -> Redirect
	| _ -> Do_not_redirect
      end;
      if use_proxy then
        (* absolute URI *)
	let query = "http://" ^ host ^ ":" ^ string_of_int port ^ path in
        request_uri <- query
      else begin
        request_uri <- if path = "" then
                          (if use_asterisk then "*" else "/")
		       else path
      end;
      request <- req ^ " " ^ request_uri ^ " HTTP/1.1";
      using_proxy <- use_proxy;


    method dest_status () =
      (* analyzes "status" line in 'received' *)
      (* Raises Bad_message if the status line is malformed. 
       * Raises Not_found if the status line is not yet complete.
       *)

      if error_exception = No_exception then begin
	if status_code <> 0 then
          (status_version, status_code, status_text)
	else begin
          let nl = B.index_from received 0 '\n' in (* or Not_found *)
	  let status_line = B.sub received 0 nl in
	  let version, code, text = 
	    try
	      Http_client_aux.match_status status_line (* or Not_found *)
	    with
		Not_found ->
		  raise (Bad_message ("Bad status line: " ^ status_line))
	  in
	  status_version <- version;
	  status_code <- code;
	  status_text <- text;
	  (status_version, status_code, status_text)
	end
      end
      else
	raise (Http_protocol error_exception)


    method decode_header (_:secret) =
      (* Decode header in 'received' and store it in 'received_header'. *)
      (* Set also 'received_entity'. *)
      (* Raises Bad_message if the header contains a bad line.
       * Raises Header_is_incomplete if the header is incomplete.
       *)

      try
	(* If 'received_header' is already set, both 'received_header' and
	 * 'received_entity' are valid.
	 *)
	if received_header = [] then begin
	  let first_nl = B.index_from received 0 '\n' in (* or Not_found *)
	  (* If the first line is malformed, throw an exception immediately: *)
	  ignore(self # dest_status());
	  (* Try to decode the header after the status line: *)
          let header, length = self # decode_header_at () received (first_nl+1) in
          received_header <- header;
          received_entity <- first_nl + length + 1
	end
      with
          Not_found ->
	    (* status line not terminated *)
	    raise Header_is_incomplete


    method decode_header_at (_:secret) message position =
      (* This method is also used during interpretation of chunked encoding. *)
      (* Raises Header_is_incomplete if "end of header" is missing. *)
      (* Raises Bad_message if the header contains a bad line. *)

      let length = ref 0 in

      let decode_line line =
	try Http_client_aux.match_header_line line
	with
	  Not_found -> 
	    raise (Bad_message "Bad header line")
      in

      let rec get_header pos prefix =
        (* next line: *)
	let nl =  try B.index_from message pos '\n'
	with Not_found -> raise Header_is_incomplete in
	let line =
	  if nl > pos && (B.unsafe_buffer message).[ nl - 1 ] = '\r' then
	    B.sub message pos (nl-pos-1)
	  else
	    B.sub message pos (nl-pos)
	in

	if line = "" then begin
	  length := nl+1-position;
	  if prefix <> "" then [ decode_line prefix ] else []
	end
	else begin
	  if List.mem line.[0] [' '; '\t'] then begin
	    (* continuation; RFC 2068 says:
	     * "All linear white space, including folding, has the same 
	     *  semantics as SP."
	     * Because of this linear white space is transformed into
	     * a single SP (space) character.
	     *)
	    (* Search the first non-space, non-tab character: *)
	    let line_length = String.length line in
	    let rec substring_non_space i =
	      if i < line_length then
		match line.[i] with
		    (' '|'\t') -> substring_non_space (i+1)
		  | _          -> String.sub line i (line_length - i)
	      else
		(* a totally empty line *)
		""
	    in
	    let transformed_line = " " ^ substring_non_space 0 in
	    get_header (nl+1) (prefix ^ transformed_line)
	  end  
	  else begin
	    let h = if prefix <> "" then [ decode_line prefix ] else [] in
	    h @ get_header (nl+1) line
	  end
	end
      in

      assert(position <= B.length message);
      let lines = get_header position "" in

      List.flatten
	(List.map
	   (fun (n,v) -> if n = "" then [] else [n,v])
	   lines),
      !length


    method decode_body
	     (_:secret) verbose_response_header verbose_response_contents =
      (* Assumes that headers are already decoded.
       * Handles transfer encodings, currently only "chunked".
       * Note that this may add further entity headers.
       * Raises Bad_message if there are any problems with the message.
       *)

      let received_length = B.length received in

      let rec chunks pos =
	(* get all chunks beginning at 'pos' *)
	assert (pos <= received_length);
	let nl =
	  try
	    B.index_from received pos '\n'
	  with
	    Not_found -> 
	      raise (Bad_message "Cannot decode chunk; LF not found")
	in
	let length_line = B.sub received pos (nl-pos) in
	let length =
	  try
	    let hexnum = Http_client_aux.match_hex length_line in
	    int_of_string ("0x" ^ hexnum)
	  with
	  | _ -> raise (Bad_message "Cannot decode chunk; bad hex number")
	in
	if length = 0 then begin
	  (* side-effect: decode any remaining footer *)
	  let footer, _ = 
	    try
	      self # decode_header_at () received (nl+1) 
	    with
		Header_is_incomplete ->
		  raise (Bad_message "Incomplete footer")
	  in
	  if verbose_response_header then
	    self # dump_header "HTTP Footer " footer;
	  received_header <- received_header @ footer;
	  []
	end else
	  let k = nl + 1 + length in    (* here must be a CRLF *)
	  let pos' =
	    try
	      if k < received_length then begin
		match (B.unsafe_buffer received).[k] with
		    '\r' ->
		      if k+1 < received_length then begin
			if (B.unsafe_buffer received).[k+1] <> '\n' then 
			  raise Not_found;
			k+2
		      end
		      else raise Not_found
		  | '\n' -> k+1            (* accept also a single LF *)
		  | _    -> raise Not_found
	      end
	      else raise Not_found
	    with 
		Not_found ->
		  raise (Bad_message "Cannot decode chunk; missing CRLF")
	  in
	  B.sub received (nl+1) length :: chunks pos'
      in

      let transfer_encoding =
	String.lowercase
	  (try List.assoc "transfer-encoding" received_header
	   with Not_found -> "") in

      begin
	match transfer_encoding with
	  "chunked" ->
	    assert(received_entity <= B.length received);
	    received_contents <- String.concat "" (chunks received_entity)
		(* Note: 'content-length' is ignored (see RFC 2068, 4.4) *)

	 | "" ->
	  (* If there is a "content-length" use this *)
	  (* NOTE: RFC 2068 demands to ignore "content-length" if a body
	   * is not allowed for that response.
	   *)
	     let no_body_expected =
	       request_method = "HEAD" || status_code = 204 || status_code = 304
	     in
	     let length =
	      if no_body_expected 
	      then 0           (* HEAD responses never have a body *)
	      else
		try
		  int_of_string (List.assoc "content-length" received_header)
		with
		    Not_found -> B.length received - received_entity
		  | _ -> 
		      raise (Bad_message "Bad Content-Length header field")
	    in
	    (* The followging error must be given due to RFC 2068, 4.4: *)
	    if length > B.length received - received_entity then
	      raise (Bad_message "Message does not match Content-Length");
	    received_contents <-
	      B.sub received received_entity length;
	    received <- B.create 1
	|	_ ->
	    raise (Bad_message ("Unknown transfer encoding '" ^
				transfer_encoding ^ "'"))
      end;
      if verbose_response_contents then
	prerr_endline ("HTTP Response:\n" ^ received_contents)


    method received_body_is_complete (_:secret) =
      (* Similar to 'body_is_complete' but uses values from slots.
       * Returns (header_length, body_length) on success.
       * Exceptions: See 'body_is_complete'
       *)
      let body_length = 
	self # body_is_complete () received_header received received_entity in
      (received_entity, body_length)


    method body_is_complete (_:secret) header message body_pos =
      (* Only checks if the body is valid *)
      (* 'header':   already decoded header *)
      (* 'message':  the full message that has been received *)
      (* 'body_pos': where the body begins in 'message' *)
      (* Raises:
       * - Body_is_incomplete if the body is incomplete because of
       *   incomplete chunks or because of 'content-length'
       * - Body_maybe_complete for unframed messages (HTTP/1.0)
       * - Bad_message if the message is not only incomplete,  but contains
       *   really bad material
       * Otherwise, returns the total length of the body
       * (i.e. the body can be found in 
       *     String.sub message body_pos total_length)
       *)
      (* Note: This method does not depend on the object's state *)

      let message_length = B.length message in
      assert(body_pos <= message_length);

      let rec chunks pos =
	(* check all chunks beginning at 'pos' *)
	assert (pos <= message_length);
	let nl =
	  try
	    B.index_from message pos '\n'
	  with
	    Not_found -> raise Body_is_incomplete
	in
	let length_line = B.sub message pos (nl-pos) in
	let length =
	  try
	    let hexnum = Http_client_aux.match_hex length_line in
	    int_of_string ("0x" ^ hexnum)
	  with
	    _ -> raise (Bad_message "Cannot decode chunk; bad hex number")
	in
	if length = 0 then begin
	  let footer, footer_length =
	    try self # decode_header_at () message (nl+1)
	    with Header_is_incomplete -> raise Body_is_incomplete
          in
	  (nl + 1 + footer_length) - body_pos
	  
	end else
	  let k = nl + 1 + length in    (* here must be a CRLF *)
	  let pos' =
	    if k < message_length then begin
	      match (B.unsafe_buffer message).[k] with
		  '\r' ->
		    if k+1 < message_length then begin
		      if (B.unsafe_buffer message).[k+1] <> '\n' then 
			raise (Bad_message "Cannot decode chunk; missing CRLF");
		      k+2
		    end
		    else raise Body_is_incomplete
		| '\n' -> k+1            (* accept also a single LF *)
		| _    -> 
		    raise (Bad_message "Cannot decode chunk; missing CRLF")
	    end
	    else raise Body_is_incomplete
	  in
	  chunks pos'
      in

      let transfer_encoding =
	String.lowercase
	  (try List.assoc "transfer-encoding" header
	   with Not_found -> "") in

      let beginning = body_pos in                  (* first position of body *)

      begin
	match transfer_encoding with
	  "chunked" ->
	    chunks beginning
	| "" ->
	  (* If there is a "content-length" use this *)
	    let length =
	      if request_method = "HEAD"
	      then 0
	      else 
		try
		  int_of_string (List.assoc "content-length" header)
		with
		    Not_found -> raise Body_maybe_complete
		  | _ -> raise (Bad_message "Bad Content-Length header field")
	    in
	    if length > B.length message - beginning then
	      raise Body_is_incomplete;
	    length
	| _ ->
	    raise (Bad_message ("Unknown transfer encoding '" ^
				transfer_encoding ^ "'"))
      end
  end
;;


(******** SUBCLASSES IMPLEMENTING HTTP METHODS ************************)

class get the_query =
  object (self)
    inherit message

    val query = the_query

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "GET"
  end
;;


class trace the_query max_hops =
  object (self)
    inherit message

    val query = the_query
    val hops = max_hops

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "TRACE";
      self # set_req_header "max-forwards" (string_of_int hops)
  end
;;


class options the_query =
  object (self)
    inherit message

    val query = the_query

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy true "OPTIONS"
  end
;;


class head the_query =
  object (self)
    inherit message

    val query = the_query

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "HEAD"
  end
;;


class post the_query the_params =
  object (self)
    inherit message

    val query = the_query
    val params = the_params

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "POST";
      self # set_req_header "content-type" "application/x-www-form-urlencoded";
      let s = List.map (fun (n,v) -> n ^ "=" ^ Netencoding.Url.encode v) params in
      body <- String.concat "&" s
  end
;;


class post_raw the_query the_contents =
  object (self)
    inherit message

    val query = the_query
    val cont = the_contents

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "POST";
      body <- cont
  end
;;


class put the_query the_contents =
  object (self)
    inherit message

    val query = the_query
    val cont = the_contents

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "PUT";
      body <- cont
  end
;;


class delete the_query =
  object (self)
    inherit message

    val query = the_query

    initializer
      self # init_query () query

    method prepare use_proxy =
      self # set_request use_proxy false "DELETE"
  end
;;


(**********************************************************************)
(***                  AUTHENTICATION METHODS                        ***)
(**********************************************************************)

(* Authentication methods specify the behaviour of the client in the 
 * case the server requires authorization.
 *)


class basic_auth_method =
  (* The "Basic" authentication method which sends a password as clear text *)
  object (self)
    val mutable userdb = []    (* pairs (realm, (user, password)) *)
    val mutable current_realm = ""
    val mutable current_host = ""
    val mutable current_port = -1

    method name = "basic"

    method set_realm realm user password =
      userdb <- (realm, (user,password)) ::
	          listrm realm userdb

    method get_credentials () =
      List.assoc current_realm userdb


    method www_authenticate (req : message) param_toks =
      (* Called on response codes 401 and 407. *)
      let p_alist = self # scan_params param_toks in
      let get n = try List.assoc n p_alist with Not_found -> "" in
      current_realm <- get "realm";
      current_host <- req # get_host();
      current_port <- req # get_port()

    method set_authorization (req : message) name_authorization =
      (* Do not send the password to a host that is different from the host
       * which requested authorization the last time.
       * 
       * Raise Not_found if authorization fails already at this stage.
       *)
      if req # get_host() <> current_host or
	req # get_port() <> current_port then raise Not_found;
      let (user,password) = self # get_credentials() in  (* or Not_found *)
      let basic_cookie = Netencoding.Base64.encode (user ^ ":" ^ password) in
      let cred = "Basic " ^ basic_cookie in
      req # set_req_header name_authorization cred


    method update (req : message) (param_toks:token list) =
      (* Called on any response code *)
      ()

    (* Note: The method 'update' is thought as a place to process information
     * such as 'nextnonce' in the digest scheme. However, 'update' requires
     * that requests and responses are handled in a strictly sequential way,
     * because we need to know the last response to set up the next request.
     * If pipelining is in effect, 'update' cannot be reasonably included into
     * the protocol.
     * Currently, 'update' is not invoked at all.
     *)

    method private scan_params p =
      let rec scan p =
	match p with
	  [] ->
	    []
	| Word name :: Special '=' :: Word value :: p' ->
	    begin
	      match p' with
		Special ',' :: p'' ->
		  (name, value) :: scan p''
	      |	[] ->
		  [name,value]
	      |	_ ->
		  failwith "error scanning header parameters"
	    end
	| _ ->
	    failwith "error scanning header parameters"
      in
      scan p
  end
;;


class digest_auth_method =
  (* The "Digest" authentication method (see RFC 2069) which works by 
   * modifying a so-called "challenge" from the server by an algorithm
   * involving a secret password and sending the computed result, the
   * "credentials" back to the server.
   *)
  object (self)
    inherit basic_auth_method

    val mutable current_nonce = ""
    val mutable current_opaque = ""

    method name = "digest"

    method www_authenticate req param_toks =
      (* Called on response codes 401 and 407 *)
      let p_alist = self # scan_params param_toks in
      let get n = try List.assoc n p_alist with Not_found -> "" in
      let algorithm = get "algorithm" in
      if algorithm = "MD5" || algorithm = "" then begin
	current_realm <- get "realm";
	current_nonce <- get "nonce";
	current_opaque <- get "opaque";
        (* domain: not interpreted *)
        (* stale: not interpreted -- we just give always another chance *)
	current_host <- req # get_host();
	current_port <- req # get_port()
      end

(*
    method debug realm nonce opaque =
      current_realm <- realm;
      current_nonce <- nonce;
      current_opaque <- opaque
*)

    method set_authorization req name_authorization =
      (* Do not send the password to a host that is different from the host
       * which requested authorization the last time
       *)
      if req # get_host() <> current_host or
	req # get_port() <> current_port then raise Not_found;
      let h s = encode_hex (Digest.substring s 0 (String.length s)) in
      let (user,password) = self # get_credentials() in  (* or Not_found *)
      let uri = req # get_req_uri() in
      let a1 = user ^ ":" ^ current_realm ^ ":" ^ password in
      let a2 = req # get_req_method() ^ ":" ^ uri in
      let response = h (h a1 ^ ":" ^ current_nonce ^ ":" ^ h a2) in
      
      let digest_response = "Digest username=\"" ^ user ^ "\", realm=\"" ^
			    current_realm ^ "\", nonce=\"" ^ current_nonce ^ 
			    "\", uri=\"" ^ uri ^ "\", response=\"" ^ response ^ 
			    "\", opaque=\"" ^ current_opaque ^ "\"" in
      req # set_req_header name_authorization digest_response

    method update req param_toks =
      let p_alist = self # scan_params param_toks in
      let get n = try List.assoc n p_alist with Not_found -> "" in
      let nextnonce = get "nextnonce" in
      if nextnonce <> "" then begin
	current_nonce <- nextnonce
      end

  end
;;


(**********************************************************************)
(***           THE BUFFER FOR THE REPLY BEING RECEIVED              ***)
(**********************************************************************)

(* The class 'recv_buffer' is a queue of octets which can be filled by
 * a Unix.read call at its end, and from which octets can be removed
 * at its beginning ("consuming octets").
 * There is also functionality to remove 1XX responses at the beginning
 * of the buffer, and to interpret the beginning of the buffer as HTTP
 * status line.
 * The idea of the buffer is that octets can be added at the end of the
 * buffer while the beginning of the buffer is interpreted as the beginning
 * of the next message. Once enough octets have been added that the message
 * is complete, it is removed (consumed) from the buffer, and the possibly
 * remaining octets are the beginning of the following message.
 *)


class recv_buffer fd =
  object (self)
    val fd = fd
    val io_buf = String.create 8192
    val mutable contents = B.create 8192
    val mutable eof = false
    val mutable timeout = false

    method unix_read () =
      if not eof && not timeout then begin
	let n = 
	  syscall (fun() -> Unix.read fd io_buf 0 (String.length io_buf)) in
	if n = 0 then
	  eof <- true
	else begin
	  B.add_sub_string contents io_buf 0 n;
	end
      end
	    

    method set_timeout =
      timeout <- true

    method contents =  
      (* returns what has been read but not yet consumed *)
      contents

    method eof = 
      (* end of stream indicator *)
      eof

    method timeout =
      timeout

    method consume n = 
      (* discards the first n bytes *)
      B.delete contents 0 n

    method drop_'continue'_responses =
      (* Removes all "1xx CONTINUE" responses at the beginning of the buffer.
       * Returns (c100, n) where
       * - c100: whether there was a 100 response
       * - n:    the number of 1xx response lines 
       *)
      let c100 = ref false in
      let n = ref 0 in
      try
	while true do
	  let nl = B.index_from contents 0 '\n' in  (* or Not_found *)
	  let line = B.sub contents 0 nl in
	  let version, code, text = Http_client_aux.match_status line
	    (* or Not_found *)			      
	  in
	  if code >= 100 && code < 200 then begin
	    (* After the status line, there may be header lines which are
	     * terminated by an empty line. Note that 1xx responses do not
	     * have a message body.
	     *)
	    let rec search_empty_line pos =
	      let pos_lineend = 
		B.index_from contents pos '\n' in  (* or Not_found *)
	      if pos = pos_lineend or
		 (pos + 1 = pos_lineend && (B.unsafe_buffer contents).[pos] = '\r') then
		(* empty line found *)
		pos_lineend + 1 
	      else
		search_empty_line (pos_lineend + 1)
	    in
	    if code = 100 then c100 := true;
	    let length_of_continue_response = search_empty_line (nl+1) in
	    self # consume length_of_continue_response;
	    incr n
	  end
	  else 
	    raise Not_found
	done;
	assert false
      with
	  Not_found -> !c100, !n          (* terminates the loop *)

    method status_line =
      (* Returns the status line at the beginning of the buffer as tuple
       * ((version, code, text), length),
       * where the triple represents the contents of the status line,
       * and the 'length' component contains the length of the line
       * (including the terminating line feed).
       * Raises Not_found if the buffer cannot be interpreted as status line.
       *)
      let nl = B.index_from contents 0 '\n' in  (* or Not_found *)
      let line = B.sub contents 0 nl in
      let triple = Http_client_aux.match_status line (* or Not_found *) in
      triple, (nl+1)

  end
;;

(**********************************************************************)
(***                PROTOCOL STATE OF A SINGLE MESSAGE              ***)
(**********************************************************************)

(* The class 'message_transport' is a container for the message and the
 * associated protocol state, and defines the methods to send
 * the request, and to receive the reply.
 * The protocol state stored here mainly controls the send and receive
 * buffers (e.g. how many octets have already been sent? Are the octets
 * received so far already a complete message or not?)
 *)


type message_state =
    Unprocessed     (* Not even a byte sent or received *)
  | Sending         (* A block has been sent, but it was not the last *)
  | Waiting         (* The message has been sent, but nothing was replied yet *)
  | Receiving       (* The beginning of a reply has been received *)
  | Complete        (* The whole reply has been received *)
;;


class message_transport 
  (m:message) 
  (f_done:message->unit)
  the_options
  =
  object (self) 
    val mutable state = Unprocessed
    val indicate_done = f_done
    val msg = m
    val mutable string_to_send = ""
    val mutable send_position = 0
    val mutable header_length = 0

    (* The variable 'auth_required' stores whether there was already a
     * 401 response ("Unauthorized") for this message.
     *)
    val mutable auth_required = false

    val mutable verbose_status = the_options.verbose_status
    val mutable verbose_request_header = the_options.verbose_request_header
    val mutable verbose_response_header = the_options.verbose_response_header
    val mutable verbose_request_contents = the_options.verbose_request_contents
    val mutable verbose_response_contents = the_options.verbose_response_contents
    val mutable verbose_connection = the_options.verbose_connection

    method state = state

    method auth_required = auth_required

    method got_401 = 
      (* There was a 401 response ("Unauthorized"). *)
      auth_required <- true

    method f_done = indicate_done

    method reset =
      (* resets transmission state (before trying to reconnect the server) *)
      state <- Unprocessed;
      string_to_send <- "";
      send_position <- 0;
      auth_required <- false;

    method send fd only_header =
      (* 'only_header': sends only the header lines. Invoke as follows:
       * while not "header_is_sent" do "send fd true" done.
       * It is then allowed to continue with "send fd false" invocations
       * if necessary.
       *)
      assert (state = Unprocessed || state = Sending);
      if state = Unprocessed then begin

	let body = msg # get_req_body() in

	(* 'content-length' must be set, see RFC 2068, 4.4 *)
	msg # set_req_header "content-length" 
	                     (string_of_int (String.length body));

	(* initialize 'string_to_send' and 'send_position' *)
	let request_string = msg # get_request() in
	let header = msg # get_req_header() in
	let crlf = "\r\n" in

	string_to_send <- (request_string ^ crlf ^
			   String.concat 
			     "" 
			     (List.map 
				(fun (n,v) -> n ^ ": " ^ v ^ crlf) 
				header) ^
			   crlf ^
			   body);
	send_position <- 0;

	header_length <- (String.length string_to_send - String.length body);

	if verbose_status then
	  prerr_endline ("HTTP request: " ^ request_string);
	if verbose_request_header then
	  msg # dump_header "HTTP request " header;
	if verbose_request_contents then
	  prerr_endline ("HTTP request body:\n" ^ body);
      end;


      let l_all = String.length string_to_send  in

      (* If 'only_header': Send at most 'header_length' bytes. *)
      let l = 
	if only_header 
	then header_length 
	else l_all in

      (* Send at most 16384 bytes because we can then get input again after this
       * amount of bytes. This input may be an early error message, i.e. sent
       * before the request is been fully transmitted; and the client can react
       * on this situation by interrupting the sending process.
       * Note that you can write even a large amount of bytes at once on a
       * system with enough free memory.
       *)
      let n_to_send = min 16384 (l - send_position) in
      let n = 
	syscall (fun () -> Unix.write fd string_to_send send_position n_to_send)
      in
      send_position <- send_position + n;
      
      if send_position >= l_all then
	state <- Waiting
      else
	state <- Sending;


    method header_is_sent =
      (* 'true' if the header lines have been sent. *)
      state <> Unprocessed && (send_position >= header_length)


    method abort_sending =
      (* Continue directly with the response, even if the message has not been
       * completely sent.
       *)
      string_to_send <- "";
      send_position <- 0;
      state <- Waiting


    method receive (rbuf : recv_buffer) =
      (* This method is invoked if some additional octets have been received
       * that 
       * - may begin this reply
       * - or continue this reply
       * - or make this reply complete
       * - or make this reply complete and begin another reply
       * It is checked if the reply is complete, and if so, it is recorded
       * and the octets forming the reply are removed from the input buffer.
       * Raises Bad_message if the message is malformed.
       *)
      assert (state = Waiting || state = Receiving);
      let received_buffer = rbuf # contents in
      try
	(* Set the 'received' slot of the message, and initialize it the
	 * first time. Note that the following piece of code makes the
	 * assumption that this method is invoked with state = Waiting
	 * the first time.
	 *)
	let need_to_init_header = (state = Waiting) in
	msg # set_response received_buffer need_to_init_header;

	(* Try to decode the header: *)
	msg # decode_header();

(*	prerr_endline ("Content-length = " ^ 
		       try List.assoc "content-length" header
		       with Not_found -> "-" );
 *)

	(* Try to decode the body: *)
	let header_length, body_length =
	  msg # received_body_is_complete () in

	(* Verbose output: *)
	let (version, code, code_as_string) = msg # dest_status() in
	if verbose_status then begin
	  prerr_endline ("HTTP response code: " ^ string_of_int code);
	  prerr_endline ("HTTP response code: " ^ code_as_string);
	  prerr_endline ("HTTP response protocol: "  ^ version);
	end;
	if verbose_response_header then begin
	  msg # dump_header "HTTP response " (msg # get_resp_header());
	end;

	(* If that worked, really decode the body and store the result: *)
	msg # decode_body () verbose_response_header verbose_response_contents;

	(* This means that there is a framed reply which is complete, and
	 * that this reply consists of header_length + body_length octets.
	 *)
	let total_length = header_length + body_length in

	(* Remove the processed octets from the input buffer: *)
	rbuf # consume total_length;

	(* Indicate that the message has been completely processed: *)
	msg # set_served();
	state <- Complete;
      with
	  Header_is_incomplete -> (* prerr_endline "H_incomplete";  *) ()
	| Body_is_incomplete   -> (* prerr_endline "B_incomplete";  *) ()
	| Body_maybe_complete  -> (* prerr_endline "B_mb_complete"; *) ()
	      (* Got a response without content-length header field, and
	       * not in chunked encoding; the header is complete, and a part
	       * of the body has been received. It is possible that the body
	       * is complete or that it is incomplete. 
	       *)


    method receive_eof (rbuf : recv_buffer) =
      (* This method is invoked instead of 'receive' if the EOF marker was
       * read.
       * Raises Bad_message if the message is malformed.
       *)
      assert (state = Waiting || state = Receiving);
      let received_buffer = rbuf # contents in
      try
	(* Set the 'received' slot of the message, and initialize it the
	 * first time. Note that the following piece of code makes the
	 * assumption that this method is invoked with state = Waiting
	 * the first time.
	 *)
	let need_to_init_header = (state = Waiting) in
	msg # set_response received_buffer need_to_init_header;

	(* Try to decode the header: *)
	msg # decode_header();

	let total_length =
	  try 
	    let header_length, body_length =
	      msg # received_body_is_complete ()
	    in
	    header_length + body_length
	  with
	      Body_maybe_complete ->
		(* Got a response without content-length header field, and
		 * not in chunked encoding; the header is complete, and a part
		 * of the body has been received. It is possible that the body
		 * is complete or that it is incomplete. 
		 *)
		B.length received_buffer
	in

	(* Verbose output: *)
	let (version, code, code_as_string) = msg # dest_status() in
	if verbose_status then begin
	  prerr_endline ("HTTP response code: " ^ string_of_int code);
	  prerr_endline ("HTTP response code: " ^ code_as_string);
	  prerr_endline ("HTTP response protocol: "  ^ version);
	end;
	if verbose_response_header then begin
	  msg # dump_header "HTTP response " (msg # get_resp_header());
	end;

	(* If that worked, really decode the body and store the result: *)
	msg # decode_body () verbose_response_header verbose_response_contents;

	(* Remove the processed octets from the input buffer: *)
	rbuf # consume total_length;

	(* Indicate that the message has been completely processed: *)
	msg # set_served();
	state <- Complete
      with
	  Header_is_incomplete ->
	    raise (Bad_message "Header is incomplete")
	| Body_is_incomplete ->
	    raise (Bad_message "Message body is incomplete")


    method indicate_pipelining =
      (* Return 'true' iff the reply is HTTP/1.1 compliant and does not
       * contain the 'connection: close' header.
       * Raises Bad_message if the status line is malformed.
       *)
      assert (state = Complete);
      let version, _, _ = msg # dest_status() in
      if version = "HTTP/1.1" then begin  
	(* TODO: Better: version >= HTTP/1.1 && version < HTTP/2.0 *)
	let conn_header =
	  try msg # assoc_resp_header "connection"
	  with Not_found -> ""
	in
	conn_header <> "close"
      end
      else false


    method indicate_sequential_persistency =
      (* Returns 'true' if only persistency without pipelining
       * is possible.
       *)
      assert (state = Complete);
      let proxy_connection =
	try msg # assoc_resp_header "proxy-connection"
	with Not_found -> ""
      in
      let connection = 
	try msg # assoc_resp_header "connection"
	with Not_found -> ""
      in
      (String.lowercase connection = "keep-alive") ||
      (String.lowercase proxy_connection = "keep-alive")


    method postprocess =
      (* TODO: *)
      (* If the message indicates a redirection which is allowed to be 
       * automatically followed, the message is again pushed onto the
       * pipeline.
       * If the server requests authentication, the necessary credentials
       * are set, and the message is again pushed onto the connection
       * queue.
       * Otherwise, nothing happens.
       *)
      indicate_done msg

    method message = msg
  
  end
;;


(**********************************************************************)
(**********************************************************************)
(**********************************************************************)
(***                                                                ***)
(***           THE PROTOCOL STATE OF THE CONNECTION                 ***)
(***                                                                ***)
(**********************************************************************)
(**********************************************************************)
(**********************************************************************)

(* This is the core of this HTTP implementation: The object controlling
 * the state of the connection to a server (or a proxy).
 *)

type sockstate =
    Down
  | Up_rw 
  | Up_r
;;


class connection the_esys 
                 (the_peer_host,the_peer_port)
                 (the_proxy_user,the_proxy_password) 
                 the_auth_methods
		 the_options
  =
  object (self)
    val mutable esys = the_esys
    val mutable group = None

    (* timeout_groups: Unixqueue groups that must be deleted when the 
     * connection is closed. These groups represent timeout conditions.
     *)
    val mutable timeout_groups = []

    val mutable peer_host = the_peer_host
    val mutable peer_port = the_peer_port

    val mutable socket = stdin
    val mutable socket_state = Down
    val mutable recv = new recv_buffer stdin

    val mutable write_queue = Queue.create()
    val mutable read_queue = Queue.create()

    (* polling_wr is 'true' iff the write side of the socket is currently
     * polled.
     *)

    val mutable polling_wr = false

    (* 'connecting' is 'true' if the 'connect' system call cannot connect
     * immediately, and continues to connect in the background.
     * While 'connecting' the socket is in non-blocking mode.
     *)

    val mutable connecting = false

    (* 'connect_pause': seconds to wait until the next connection is tried.
     * There seems to be a problem with some operating systems (namely
     * Linux 2.2) which do not like immediate reconnects if the previous
     * connection did not end in a sane state.
     * A value of 0.5 seems to be sufficient for reconnects (see below).
     *)			      

    val mutable connect_pause = 0.0

    (* The following two variables control whether pipelining is enabled or
     * not. The problem is that it is unclear how old servers react if we
     * sent to them several requests at once. The solution is that the first
     * "round" of requests and replies is done in a compatibility mode: After
     * the first request has been sent sending stops, and the client waits
     * until the reply is received. If the reply indicates HTTP/1.1 and does
     * not contain a "connection: close" header, all further requests and 
     * replies will be performed in pipelining mode, i.e. the requests are
     * sent independent of whether the replies of the previous requests have
     * been received or not.
     *
     * enable_persistency: 'true' means that the server is able to process
     *    the requests in pipelining mode. 'false' means that the type of the
     *    server is unknown, and because of this compatibility mode is 
     *    preferred.
     * sending_first_message: 'true' means that the first request has not yet
     *    been completely sent to the server. 
     * done_first_message: 'true' means that the reply of the first request
     *    has been arrived.
     *)

    val mutable enable_persistency = false
    val mutable sending_first_message = true
    val mutable done_first_message = false

    val mutable synchronization = the_options.synchronization

    (* 'inhibit_pipelining_byserver': becomes true if the server is able
     * to keep persistent connections but is not able to use pipelining.
     * (HTTP/1.0 "keep alive" connections)
     *)

    val mutable inhibit_pipelining_byserver = false
 
    (* Premature shutdowns: HTTP specifies that if the client sees an EOF
     * before it has got a complete status line, a special algorithm to
     * reliably reconnect the server must be used (see Section 8.2 of
     * RFC 2068).
     * This is implemented as follows:
     *
     * ["pe": abbreviates "premature end".]
     * pe_got_some_status: 'true' means that there was already a status line.
     * pe_trial: if 0, the client is in a normal state. If > 0, this variable
     *    counts the number of premature shutdowns.
     * pe_waiting_for_status: 'true' means that the header of the first
     *    request has been sent, and that we are waiting until 
     *    a repsonse arrives, or until a timeout has elapsed
     * pe_timer: the event group of a timer
     * 
     * If we get a status line, we can set 'pe_got_some_status' to true, and
     * 'pe_trial' to 0.
     * If we get an EOF before the status line, i.e. when 'pe_got_some_status'
     * is still true, increment 'pe_trial'.
     * If 'pe_trial' > 0, the following special behaviour is enabled:
     *
     * - Only the header of the request is sent (it is the first request on
     *   this connection) - special code in the class 'message_transport'.
     *   'pe_waiting_for_status' is set to true. The timer is started.
     *   The output event handler is removed.
     * - If we get data from the server: if we 'pe_got_some_status', the special
     *   behaviour is disabled. (Otherwise wait for the next input event.)
     * - If we get an EOF from the server: increment 'pe_trial'; cancel the
     *   timer; set up a new timer 
     * - If the timeout happens: Disable the special behaviour, too.
     *
     * "Disabling" means: Set 'pe_trial' to 0; set 'pe_waiting_for_status' to
     * false; set '!pe_timer_enabled' to false; remove the timer; 
     * Create a new output handler if necessary (i.e. if output is pending).
     *)

    val mutable pe_got_some_status = false
    val mutable pe_trial = 0
    val mutable pe_waiting_for_status = false
    val mutable pe_timer = None

    (* 'connect_started': when the last 'connect' operation started.
     * 'connect_time': how many seconds the last 'connect' lasted. 
     *)

    val mutable connect_started = 0.0
    val mutable connect_time = 0.0

    (* If a connection does not lead to any type of response, the client 
     * simply re-opens a new connection and tries again. The following 
     * variables limit the number of attempts until the client gives up.
     *
     * 'totally_failed_connections': counts the number of failed connections
     *      without any response from the server.
     * 'max_totally_failed_connection': the maximum number of such connections
     *      until the client gives up
     * 'got_message_status': true if there was a status line with code
     *      2xx to 5xx (status lines 1xx do not count here).
     *)

    val mutable totally_failed_connections = 0
    val mutable max_totally_failed_connections = 
	                               the_options.maximum_connection_failures
    val mutable got_message_status = false

    (* 'close_connection' indicates that a HTTP/1.0 response was received or 
     * that a response contained a 'connection: close' header.
     *)

    val mutable close_connection = false

    (* 'no_persistency': 'true' inhibits that more than one request/reply round
     * is done over one connection.
     *)

    val mutable no_persistency = the_options.inhibit_persistency

    (* After 'maximum_number_of_errors' the request is not tried again,
     * but the failure is reported.
     *)

    val mutable maximum_number_of_errors = the_options.maximum_message_errors


    (* waiting_for_status100: In Sync_with_handshake_before_request_body
     * mode, this reference counts special events after the request header
     * has been sent:
     *   0: request header is not yet sent
     *   1: request header has just been sent, now waiting for status 100
     *  >1: number of timeouts since then
     *
     * Important detail: this variable is intentionally a reference
     * which is shared with the timeout closure. So:
     *
     * waiting_for_status100 <- ref 0: this means that a new
     *   reference is created, and if there is an existing timeout closure
     *   the closure will not occasionally change the value of the new
     *   reference.
     *   This is right after a message has been processed (reinitialization).
     * waiting_for_status100 := 0: this means that the current reference
     *   is changed.
     *   This is right while processing the current message.
     *)

    val mutable waiting_for_status100 = ref 0


    (* Proxy authorization: If 'proxy_user' is non-empty, the variables
     * 'proxy_user' and 'proxy_password' are interpreted as user and
     * password for proxy authentication. More precisely, once the proxy
     * responds with status code 407, 'proxy_credentials_required' becomes
     * true, and all following requests will include the credentials identifying
     * the user (with the "basic" authentication method).
     * If the proxy responds again with code 407, this reaction will not
     * be handled again but will be visible to the outside.
     *)

    val mutable proxy_user = the_proxy_user
    val mutable proxy_password = the_proxy_password

    val mutable proxy_credentials_required = false
    val mutable proxy_cookie = ""


    (* Server authorization: Once the server responds with status code
     * 401, the requested authentication method is triggered. The variable
     * 'auth_methods' contains implementations of such methods as objects.
     * These objects have methods which add the necessary header fields
     * containing the credentials.
     * Once an authentication method has been triggered, it is likely that
     * the next message must be authorized, too. Because of this, the
     * method used last is stored in 'current_auth_method' and automatically
     * activated again for the next message.
     *)
    val mutable auth_methods = the_auth_methods
    val mutable current_auth_method = None


    (* 'critical_section': true if further additions to the queues and
     * every change of the internal state must be avoided. In this case,
     * additions are put into 'deferred_additions' and actually added later.
     *)

    val mutable deferred_additions = Queue.create()
    val mutable critical_section = false


    val mutable options = the_options

    val mutable verbose_status = the_options.verbose_status
    val mutable verbose_request_header = the_options.verbose_request_header
    val mutable verbose_response_header = the_options.verbose_response_header
    val mutable verbose_request_contents = the_options.verbose_request_contents
    val mutable verbose_response_contents = the_options.verbose_response_contents
    val mutable verbose_connection = the_options.verbose_connection


    method length =
      (* Returns the number of open requests (requests without response) *)
      Queue.length read_queue + Queue.length deferred_additions


    method add (m : message) f_done =
      (* add 'm' to the read and write queues *)
      ignore(self # add_msg m f_done)


    method private add_msg m f_done =
      (* In contrast to 'add', the new message transporter is returned. *)

      (* Check if DNS lookup succeeds *)
      if socket_state = Down then
	ignore(self # inet_addr);

      (* If handshake is demanded: add the "expect" header now: *)
      begin match synchronization with
	  Sync_with_handshake_before_request_body _ ->
	    if m # get_req_body() <> "" then
	      m # set_req_header "expect" "100-continue";
	| _ -> ()
      end;

      (* If proxy authentication is enabled, add the necessary header fields: 
       * (See also the code and the comments in method 
       * 'postprocess_complete_message')
       *)
      if proxy_credentials_required  &&  m # using_proxy then begin
	(* Note: We only add the proxy password if 'using_proxy' returns true
	 * because this indicates that the peer was contacted as proxy.
	 *)
	(* If the cookie is not yet computed, do this now: *)
	if proxy_cookie = "" then begin
	  proxy_cookie <- Netencoding.Base64.encode (proxy_user ^ ":" ^ proxy_password);
	end;
	(* Add the "proxy-authorization" header: *)
	m # set_req_header 
	  "proxy-authorization"
	  ("Basic " ^ proxy_cookie);
      end;

      (* If there is an authentication method, apply it now: *)

      (* Note: It is questionable to apply the 'current_auth_method' here.
       * Suggest this method is demanded upon request x1, and that x2 to xN 
       * are already on the queue. In this case, x1 is added again to queue, and
       * the new method is applied to it; but x2 to xN still are unaffected;
       * the first message for which 'current_auth_method' has an effect is
       * the message x{N+1}.
       * It may be cleverer to apply 'current_auth_method' just before
       * it is sent over the network.
       *)

      begin match current_auth_method with
	  None ->
	    (* No auth method. *)
	    ()
	| Some a ->
	    (* Apply the method 'a' only if the request does not already
	     * contain the credentials:
	     *)
	    if
	      try let _ = m # assoc_req_header "authorization" in false
	      with Not_found -> true
	    then begin
	      (* The "authorization" header is not present. Set it now: *)
	      a # set_authorization m "authorization";
	    end
      end;
      
      (* Create the transport container for the message and add it to the
       * queues:
       *)
      let trans = new message_transport m f_done options in

      if critical_section then begin
	(* This 'add' invocation was done in a callback. This may interfer
	 * with other queue modifications.
	 *)
	Queue.add trans deferred_additions;
      end
      else begin
	Queue.add trans write_queue;
	Queue.add trans read_queue;

        (* If there is currently no event group, we are detached from the
         * event system (i.e. not receiving events). Attach again.
         *)
	if group = None then self # attach_to_esys;

        (* Update the polling state. *)
	self # maintain_polling;
      end;

      trans


    method leave_critical_section : unit =
      (* Move the entries from 'deferred_additions' to the real queues. *)

      Queue.iter
	(fun trans ->
	   Queue.add trans write_queue;
	   Queue.add trans read_queue;
	)
	deferred_additions;

      if Queue.length deferred_additions > 0 then begin

        (* If there is currently no event group, we are detached from the
	 * event system (i.e. not receiving events). Attach again.
	 *)
	if group = None then self # attach_to_esys;

        (* Update the polling state. *)
	self # maintain_polling;

	Queue.clear deferred_additions
      end
 

    method private add_again m_trans =
      (* add 'm_trans' again to the read and write queues (as a consequence
       * of authorization)
       *)

      let m = m_trans # message in
      let f_done = m_trans # f_done in

      (* Reset all transmission state contained in 'm': *)
      m # set_unserved();

      self # add_msg m f_done


    method set_options p =
      options <- p;
      synchronization <- options.synchronization;
      max_totally_failed_connections <- options.maximum_connection_failures;
      no_persistency <- options.inhibit_persistency;
      maximum_number_of_errors <- options.maximum_message_errors;
      verbose_status <- options.verbose_status;
      verbose_request_header <- options.verbose_request_header;
      verbose_response_header <- options.verbose_response_header;
      verbose_request_contents <- options.verbose_request_contents;
      verbose_response_contents <- options.verbose_response_contents;
      verbose_connection <- options.verbose_connection;


    method private attach_to_esys =
      assert (group = None);
      let g = Unixqueue.new_group esys in
      group <- Some g;
      self # reinitialize;


    method private reinitialize =
      assert (socket_state = Down);

      let g = match group with
	  Some x -> x
	| None -> assert false
      in

      connecting <- false;
      enable_persistency <- false;
      sending_first_message <- true;
      done_first_message <- false;
      close_connection <- false;
      polling_wr <- false;
      critical_section <- false;
      inhibit_pipelining_byserver <- false;
      waiting_for_status100 <- ref 0;

      pe_got_some_status <- false;
        (* The other pe_* variables survive the lifetime of connections. *)

      got_message_status <- false;

      let g1 = Unixqueue.new_group esys in
      Unixqueue.once 
	esys
	g1
	connect_pause
	(fun () -> 
	   try
	     connect_pause <- 0.0;
	     self # connect_server;        (* may raise Exit on fatal errors *)
	     let timeout_value = options.connection_timeout in
	     Unixqueue.add_resource esys g (Unixqueue.Wait_in socket, 
					    timeout_value);
	     Unixqueue.add_close_action esys g (socket, (fun _ -> self # shutdown));
	     Unixqueue.add_handler esys g (self # handler);
	     self # maintain_polling;
	   with
	       Exit -> ()
	)


    method private maintain_polling =

      (* If one of the following conditions is true, we need not to poll
       * the write side of the socket:
       * - The write_queue is empty but the read_queue not
       * - The difference between the read and the write queue is too big
       * - We wait for the reply of the first request send to a server
       * - The write side of the socket is closed
       * - pe_waiting_for_status is true.
       * - waiting_for_status100 is 1
       *)

      let timeout_value = options.connection_timeout in

      let actual_max_drift =
	if inhibit_pipelining_byserver then 0 else
	  match synchronization with
	      Pipeline drift -> drift
	    | _              -> 0
		(* 0: Allow no drift if pipelining is not allowed *)
      in

      if (Queue.length write_queue = 0 && Queue.length read_queue > 0) or
	 (Queue.length read_queue - Queue.length write_queue > actual_max_drift) or
	 (not done_first_message && not sending_first_message) or
         socket_state <> Up_rw or
	 pe_waiting_for_status or
	 !waiting_for_status100 = 1
      then begin
	if polling_wr then begin
	  let g = match group with
	      Some x -> x
	    | None -> assert false
	  in
(*  prerr_endline "REMOVE";  *)
	  Unixqueue.remove_resource esys g (Unixqueue.Wait_out socket);
	end;
	polling_wr <- false
      end;

      (* On the other hand, all of the following conditions must be true
       * to enable polling again:
       * - The write_queue is not empty, or
       *   both the write_queue and the read_queue are empty
       * - The difference between the read and the write queue is small enough
       * - We send the first request to a server, or do pipelining
       * - The write side of the socket is open
       * - pe_waiting_for_status is false.
       * - waiting_for_status100 is not 1
       *)

(*
let prerr_bool b =
prerr_endline (if b then "T" else "F") in
prerr_bool (Queue.length write_queue > 0 or
	   (Queue.length write_queue = 0 && Queue.length read_queue = 0));
prerr_bool done_first_message;
prerr_bool sending_first_message;
prerr_bool (socket_state = Up_rw);
prerr_bool pe_waiting_for_status;
*)

      if (Queue.length write_queue > 0 or
	   (Queue.length write_queue = 0 && Queue.length read_queue = 0)) &&
	 (Queue.length read_queue - Queue.length write_queue <= actual_max_drift) &&
	 (done_first_message || sending_first_message) &&
	 socket_state = Up_rw &&
	 not pe_waiting_for_status &&
         !waiting_for_status100 <> 1
      then begin
	if not polling_wr then begin
	  let g = match group with
	      Some x -> x
	    | None -> assert false
	  in
(*  prerr_endline "ADD";   *)
	  Unixqueue.add_resource esys g (Unixqueue.Wait_out socket, 
					 timeout_value
					);
	end;
	polling_wr <- true;
      end;


    method private shutdown =

      if verbose_connection then 
	prerr_endline "HTTP connection: Shutdown!";
      begin match socket_state with
	  Down -> ()
	| (Up_rw | Up_r) -> 
	    if verbose_connection then 
	      prerr_endline "HTTP connection: Closing socket!";
	    syscall (fun () -> Unix.close socket)
      end;
      socket_state <- Down;
      ( match group with
	    Some g -> 
	      Unixqueue.clear esys g;
	      group <- None;
	  | None ->  ()
      );
      List.iter (Unixqueue.clear esys) timeout_groups;
      timeout_groups <- []


    method private clear_timeout g =
      Unixqueue.clear esys g;
      timeout_groups <- List.filter (fun x -> x <> g) timeout_groups;


    method private abort_connection =

      (* By removing the input and output resources, the event queue is told
       * that nothing more will be done with the group g, and because of this
       * the queue invokes the 'close action' (here self # shutdown) and
       * cleans up the queue.
       *)
      match group with
	  Some g -> 
	    Unixqueue.remove_resource esys g (Unixqueue.Wait_in socket);
	    if polling_wr then begin
	      Unixqueue.remove_resource esys g (Unixqueue.Wait_out socket);
	      polling_wr <- false;
	    end;
	    assert (group = None);
        | None -> 
	    ()


    method private inet_addr =
      try
	(* TODO: 'inet_addr_of_string' may block *)
	inet_addr_of_string (Http_client_aux.match_ip peer_host)
      with
	  Not_found ->
	    try
	      let h = gethostbyname peer_host in
	      h.h_addr_list.(0)
	    with Not_found ->
	      raise 
		(Sys_error 
		   ("Http_client: host name lookup failed for " ^ peer_host));

    method private connect_server =
      (* raises Exit if connection not possible *)

      if verbose_connection then
	prerr_endline ("HTTP connection: Connecting to server " ^ peer_host);

      let addr = self # inet_addr in

      let s = syscall (fun () -> Unix.socket PF_INET SOCK_STREAM 0) in
      connect_started <- Unix.gettimeofday();
      (* Connect in non-blocking mode: *)
      Unix.set_nonblock s;
      begin try
	syscall (fun () -> Unix.connect s (ADDR_INET (addr, peer_port)));
	let t1 = Unix.gettimeofday() in
	connect_time <- t1 -. connect_started;
	connecting <- false;
	if verbose_connection then
	  prerr_endline "HTTP connection: Connected!";
      with
	  Unix.Unix_error(Unix.EINPROGRESS,_,_) ->
	    (* The 'connect' has not yet been finished. *)
	    connecting <- true;
	    (* The 'connect' operation continues in the background.
	     * It is guaranteed that the socket becomes writeable if
	     * the connection is established.
	     * (Of course, it becomes readable if there is already data
	     * to read, but if the other side does not send us anything
	     * only writeability is indicated.)
	     * If the connection fails: This situation is not very well
	     * described in the manual pages. The "Single Unix Spec"
	     * says nothing about this case. In the Linux manpages I 
	     * found that it is possible to read the O_ERROR socket option
	     * (see connect(2)). By experimenting I found out that the socket
	     * indicates readability, and that the following "read" syscall
	     * then reports the error correctly.
	     * The O_ERROR socket option is not supported by O'Caml, so
	     * the latter is assumed.
	     *)
	| Unix.Unix_error(_,_,_) as err ->
	    (* Something went wrong. E.g. network is unreachable. *)
	    (* We cannot call abort_connection or shutdown, because the
	     * state is not fully initialized. So clean up everything
	     * manually.
	     *)
	    ( match group with
		  Some g ->
		    Unixqueue.clear esys g;         (* clean up esys *)
		    group <- None
		| None -> ()
	    );
	    Unix.close s;                 (* close socket explicitly *)
	    (* We cannot call cleanup_on_eof() here because this would
	     * recursively try to connect again. Instead: drop all messages
	     *)
	    Queue.iter 
	      (fun m ->
		 m # message # set_error_exception () err;
		 self # critical_postprocessing m;  (* because m is dropped *)
	      )
	      read_queue;
	    Queue.clear read_queue;
	    Queue.clear write_queue;
	    raise Exit
      end;

      socket <- s;
      socket_state <- Up_rw;
      recv <- new recv_buffer socket;


    method private check_connection =
      (* You need to call this method only if 'connecting' is true, and of
       * course if the socket is either readable or writeable.
       * The socket is set to blocking mode, again. The connect time
       * is measured and recorded.
       * TODO: find out if a socket error happened in the meantime.
       *)
      if connecting then begin
	(* No longer clear the nonblock flag. The rest of the code handles
	 * the EAGAIN situations that may happen.
	 *)
	(* Unix.clear_nonblock socket; *)
	let t1 = Unix.gettimeofday() in
	connect_time <- t1 -. connect_started;
	connecting <- false;
	if verbose_connection then
	  prerr_endline "HTTP connection: Got connection status";
      end


    method private handler _ _ ev =

      let g = match group with
	  Some x -> x
	| None -> 
	    (* This is possible while shutting down the socket *)
	    raise Equeue.Reject
      in
      match ev with
	  Unixqueue.Input_arrived (g0,fd0) ->
	    if g0 = g then self # handle_input else raise Equeue.Reject
	| Unixqueue.Output_readiness (g0,fd0) ->
	    if g0 = g then self # handle_output else raise Equeue.Reject
	| Unixqueue.Timeout (g0, _) ->
	    if g0 = g then self # handle_timeout else raise Equeue.Reject
	| _ ->
	    raise Equeue.Reject


    (**********************************************************************)
    (***                    THE TIMEOUT HANDLER                         ***)
    (**********************************************************************)

    method private handle_timeout =
      (* No network packet arrived for a period of time.
       * May happen while connecting to a server, or during operation.
       *)

      recv # set_timeout;
      self # handle_input;


    (**********************************************************************)
    (***                     THE INPUT HANDLER                          ***)
    (**********************************************************************)

    method private handle_input =
      (* Data have arrived on the 'socket'. First we receive as much as we
       * can; then the data are interpreted as sequence of messages.
       *)
      
      (* Ignore this event if the socket is Down (this may happen
       * if the input side is closed while there are still input
       * events in the queue):
       *)
      if socket_state = Down then
	raise Equeue.Reject;

      if verbose_connection then 
	prerr_endline "HTTP connection: Input event!";

      let g = match group with
	  Some x -> x
	| None -> assert false
      in

      let end_of_queueing = ref false in
      (* 'end_of_queueing': stores whether there was an EOF or not *)


      (************ ACCEPT THE RECEIVED OCTETS ************)

      if not recv # timeout then
	begin try
	  if connecting then
	    self # check_connection;
	  recv # unix_read();
	with
	    Unix.Unix_error(Unix.EAGAIN,_,_) ->
	      (* This should normally not happen, but there are operating
	       * systems with buggy socket implementations.
	       *)
	      if verbose_connection then
		prerr_endline("HTTP connection: Strange socket behaviour (cannot read although the socket is readable)");
	      ();  (* Ignore! *)
	  | Unix.Unix_error(e,a,b) as err ->
	      if verbose_connection then
		prerr_endline("HTTP connection: Unix error " ^
			      Unix.error_message e);
	      self # reset_after_unix_error err;
	      raise (Unix.Unix_error(e,a,b));
	end;

      if recv # eof || recv # timeout then begin
	(* shutdown the connection, and clean up the event system: *)
	if verbose_connection && recv # eof then 
	  prerr_endline "HTTP connection: Got EOF!";
	if verbose_connection && recv # timeout then 
	  prerr_endline "HTTP connection: Connection timeout!";
	
	if not(recv # eof) then begin
	  (* Perhaps we are not connected: *)
	  try
	    syscall (fun () -> Unix.shutdown socket Unix.SHUTDOWN_SEND);
	  with
	      Unix.Unix_error(Unix.ENOTCONN,_,_) ->
		()
	end;
	self # abort_connection;
	end_of_queueing := true
      end;


      (************ TRY TO INTERPRET THE OCTETS AS MESSAGES **************)

      let rec read_loop() =
	if B.length(recv # contents) > 0 then begin
	  let this = Queue.peek read_queue in    (* may raise Queue.Empty *)

	  (*** Process 'this' ***)

	  begin match this # state with
	      Unprocessed ->
		(* 'this' is the message object which is the next object on
		 * the read queue. If the state is Unprocessed, there was no
		 * request, and it is impossible that we received octets of
		 * a reply never requested. So this is a protocol error.
		 *
		 * Exception: "100 CONTINUE" repsonses are allowed; they 
		 * may be left from the previous request.
		 *)

	        (* Drop all "Continue" responses: 
		 *
		 * n_continue: the number of "continue" responses found.
		 *)

		let _, n_continue = 
		  recv # drop_'continue'_responses in

		if n_continue > 0 then
		  pe_got_some_status <- true;  (* counts as status line *)

		if B.length(recv # contents) > 0 then begin
		  if verbose_connection then
		    prerr_endline "HTTP connection: protocol error (A)";
		  raise Abort_on_protocol_error;
		end

  	  (*** --- ***)

	    | Sending ->
		(* This 'state' means: At this moment a request is being
		 * sent to the server. 
		 * I.e. we received octets for a request we are currently
		 * sending.
		 *)
		(* It is possible that the server responds to requests which
		 * have not yet been completely sent to the server. This can
		 * be a simple "100 CONTINUE" message (which can be ignored), 
		 * but can also be an error message.
		 * As the current state is "Sending", this request is a bit 
		 * longer (more than one network packet), and because of this
		 * it is best to abort the request. This is only possible by
		 * terminating the TCP connection (the alternative would be
		 * to use chunked encoding for requests if it is known that the
		 * server is HTTP/1.1 compliant; chunked messages can be
		 * aborted by sending a zero-length chunk and an empty footer).
		 *)
		assert (try Queue.peek write_queue == this
			with Queue.Empty -> false);

	        (* Drop all "Continue" responses: 
		 *
		 * n_continue: the number of "continue" responses found.
		 *)

		let c100, n_continue = 
		  recv # drop_'continue'_responses in

		if n_continue > 0 then
		  pe_got_some_status <- true;  (* counts as status line *)
		if c100 then
		  waiting_for_status100 := 2; (* update current reference *)

	        (* If there are still bytes in the recv buffer, try to interpret
		 * them as error message, and abort the connection.
		 *)

		if B.length(recv # contents) > 0 then begin
		  if verbose_connection then 
		    prerr_endline "HTTP connection: Got asynchronous response";
		  
		  begin try
		    ignore(recv # status_line);
		    pe_got_some_status <- true;  (* There is a status line *)
		    got_message_status <- true;
		    waiting_for_status100 := 2;
		  with
		      Not_found ->
			if verbose_connection then
			  prerr_endline "HTTP connection: protocol error (B)";
			raise Abort_on_protocol_error;
		  end;

		  if not (recv # eof) then begin
		    (* abort the connection *)

		    if verbose_connection then
		      prerr_endline "HTTP connection: Sending asynchronous EOF";

		    syscall
		      (fun () ->
			 Unix.shutdown socket Unix.SHUTDOWN_SEND);

		    socket_state <- Up_r;
		    (* The next "self # maintain_polling" will remove the
		     * output resource from the event queue.
		     *)
		  end;
		
		  (* Interpret the response: *)
		  
		  ignore (Queue.take write_queue);    (* This is done! *)
		  this # abort_sending;

		  assert (this # state = Waiting);
		  read_loop();
		end

  	  (*** --- ***)

	    | (Waiting | Receiving) ->
		(* Waiting: The request has been sent, and this is the first
		 *   piece of response received so far.
		 * Receiving: The response is not yet complete.
		 *
		 * In this case, try to interpret the received bytes as
		 * response.
		 *)

	        (* Drop all "Continue" responses: 
		 *
		 * n_continue: the number of "continue" responses found.
		 *)

		let c100, n_continue = 
		  recv # drop_'continue'_responses in

		if n_continue > 0 then
		  pe_got_some_status <- true;  (* counts as status line *)
		if c100 then
		  waiting_for_status100 := 2; (* update current reference *)

                (* Find out whether there is a status line: *)

		begin try
		  ignore(recv # status_line);  (* or raise Not_found *)
		  pe_got_some_status <- true;  (* There is a status line *)
		  got_message_status <- true;
		  waiting_for_status100 := 2;
		with
		    Not_found -> ()
		end;

		(* Note that the following two methods, 'receive_eof' and
		 * 'received', really expect that 'state = Waiting' for the
		 * first invocation.
		 * Otherwise, they do not initialize themselves correctly.
		 *)

		(* These methods may raise Bad_message. This indicates
		 * a badly synchronized stream, and the request should be
		 * repeated on a new connection if possible.
		 *)

		begin try
		  if recv # eof then
		    this # receive_eof recv 
		      (* Handles unframed messages differently *)
		  else
		    this # receive recv;
		with
		    Bad_message s as x ->
		      if verbose_connection then
			prerr_endline ("HTTP connection: Protocol error (Bad message: " ^ s ^ ")");
		      let m = this # message in
		      m # set_error_exception () x;
		      m # set_error_counter () (m # get_error_counter() + 1);
		      raise Abort_on_protocol_error;
		end;

	        (* Continue only if 'this' message is done *)
		if this # state = Complete then begin
		  (* The response is complete, and can be processed. *)
		  (* Update the knowledge about the server and the connection:*)

		  let able_to_pipeline = this # indicate_pipelining in

		  (* 'able_to_pipeline': is true if we assume that the server
		   * is HTTP/1.1-compliant and thus is able to manage pipelined
		   * connections.
		   * Update first 'close_connection': This variable becomes
		   * true if the connection is not assumed to be pipelined
		   * which forces that the CLIENT closes the connection 
		   * immediately (see the code in the output handler).
		   *)

		  let only_sequential_persistency =
		    this # indicate_sequential_persistency in

		  (* 'sequential_persistency': is true if the connection is
		   * HTTP/1.0, and the server indicated a persistent connection.
		   * In this case, pipelining is disabled.
		   *)

		  if only_sequential_persistency then begin
		    (* 'close_connection': not set.
		     * 'enable_persistency': set 
		     *)
		    if verbose_connection then 
		      prerr_endline "HTTP connection: using HTTP/1.0 style persistent connection";
		    enable_persistency <- true;
		    inhibit_pipelining_byserver <- true;
		  end
		  else begin
		    close_connection  <- close_connection  || not able_to_pipeline;
		    (* Update also 'enable_persistency': This variable enables
		     * the output handler for another request on the same
		     * connection. In this case the connection persists until
		     * the SERVER closes it.
		     *)
		    
		    enable_persistency <- enable_persistency || able_to_pipeline;
		  end;

		  (* Remember that the first request/reply round is over: *)

		  done_first_message <- true;

		  (* Reinitialize waiting_for_status100: *)

		  waiting_for_status100 <- ref 0;   (* new reference! *)

		  (* Remove this message from the queue: *)

		  ignore (Queue.take read_queue);

		  (* postprocess 'this' (may raise exceptions! (callbacks)) *)

		  self # postprocess_complete_message this;

		  (* continue with the next message *)

		  read_loop();
		end

  	  (*** --- ***)

	    | Complete ->
		(* This must not happen because complete messages are 
		 * immediately removed from the read_queue.
		 *)
		assert false
	  end
	end
      in         (* of "let rec read_loop() = " *)

      begin try
	(* Start the interpretation formulated in 'read_loop' and catch
	 * exceptions.
	 *)
	read_loop();
      with
	  Queue.Empty ->            (* i.e. 'read_queue' is empty *)
	    if B.length(recv # contents) > 0 then begin
	      (* No more responses expected, but still octets to interpret.
	       * This is a protocol error, too.
	       *)
	      if verbose_connection then begin
		let n = B.length(recv # contents) in
		prerr_endline "HTTP connection: Extra octets -- aborting connection";
		prerr_endline ("Number of extra bytes: " ^ string_of_int n);
		if n < 20 then
		  prerr_endline ("Extra bytes: " ^ 
				 encode_hex(B.contents(recv # contents)));
	      end;

	      self # abort_connection;
	      end_of_queueing := true;
	    end

	| Abort_on_protocol_error ->
	    (* Close the current connection immediately by sending EOF, and
	     * by dropping any received material.
	     *)
	    if verbose_connection then
	      prerr_endline "HTTP connection: Aborting the connection after protocol error";

	    self # abort_connection;
	    end_of_queueing := true;
      end;

      (*************** HANDLING OF PREMATURE EOF SITUATIONS ***********)

      if pe_trial > 0 then begin
	if pe_got_some_status then begin
	  (* Disable PE handling *)

	  if verbose_connection then
	    prerr_endline "HTTP connection: resuming normal operation";

	  pe_trial <- 0;
	  pe_waiting_for_status <- false;

	  self # pe_disable_timer;

	  (* The output handler will be added by 'maintain_polling' if
	   * necessary.
	   *)

	end 
	else begin
	  if !end_of_queueing then begin   (* also if protocol error happened *)
	    (* If we got another EOF, continue PE handling. *)

	    if verbose_connection then
	      prerr_endline "HTTP connection: continuing special handling after further premature EOF";

	    pe_trial <- pe_trial + 1;
	    pe_waiting_for_status <- false;   (* not yet. *)

	    self # pe_disable_timer;

	    (* Create a new timer: *)
	    let tm = Unixqueue.new_group esys in
	    let en = ref true in

	    let timeout = ref (max 1.0 connect_time) in
	    for k = 1 to pe_trial do timeout := !timeout *. 2.0 done;

	    Unixqueue.once esys tm !timeout (self # pe_timeout en);
	    
	    pe_timer <- Some tm;
	    timeout_groups <- tm :: timeout_groups;

	    (* The output handler will be added by 'maintain_polling' if
	     * necessary.
	     *)

	  end;
	end
      end
      else 
	if not pe_got_some_status && !end_of_queueing then begin
	  (* Enable PE handling *)         (* also if protocol error happened *)

	  if verbose_connection then
	    prerr_endline "HTTP connection: enabling special code after premature EOF";

	  pe_trial <- 1;
	  pe_waiting_for_status <- false;   (* not yet. *)
	  
	  (* Disable an old timer, if any: [Unlikely] *)
	  self # pe_disable_timer;

	  (* Start the timer: *)
	  let tm = Unixqueue.new_group esys in
	  let en = ref true in
	  Unixqueue.once 
	    esys tm ((max 1.0 connect_time) *. 2.0) (self # pe_timeout en);

	  pe_timer <- Some tm;
	  timeout_groups <- tm :: timeout_groups;

	  (* And now close the connection, and create a new one. *)
	end;

      (************** CLOSE THE CONNECTION IF NECESSARY, ****************)
      (************** AND PREPARE RECONNECTION           ****************)

      if !end_of_queueing then begin
	assert (group = None);
	self # cleanup_on_eof No_exception;
      end;

      (*************** UPDATE THE POLLING STATE **************)

      (* If there were 'add' invocations from callbacks, move these additions
       * to the real queues now.
       *)
      self # leave_critical_section; 

      self # maintain_polling;


    (************** CLOSE THE CONNECTION IF NECESSARY, ****************)
    (************** AND PREPARE RECONNECTION           ****************)

    method private cleanup_on_eof err : unit =
      assert (group = None);

      (* If the socket is closed, it is necessary to check whether all 
       * requests sent so far have got their replies. 
       * Cases:
       * - write_queue and read_queue are empty: all is done.
       * - write_queue and read_queue have the same number of elements:
       *   reconnect
       * - else: some replies are missing. The requests are NOT tried again
       *   by default because the operations might not be idempotent. 
       *   The messages carry a flag with them indicating whether reconnection
       *   is allowed or not.
       * It is not possible that the write_queue is longer than the read_queue.
       *)

      (* First check if the connection was a total failure, i.e. if not
       * even a status line (other than 1xx) was received. In this case
       * increase the counter for totally failed connections. If the
       * counter exceeeds a limit, all messages on the queues are discarded.
       *)
	
      if not got_message_status then begin
	(* It was a total failure. *)
	totally_failed_connections <- totally_failed_connections + 1;
	if verbose_connection then
	  prerr_endline "HTTP connection: total failure";
	
	if totally_failed_connections > max_totally_failed_connections then
	  begin
	    (* Set the error exception of all remaining messages, and
	     * clear the queues.
	     *)
	    Queue.iter 
	      (fun m ->
		 (* If the error exception of m is not yet set, set it
		  * to err or Bad_message
		  *)
		 begin try
		   ignore(m # message # get_resp_header());
		   (* It was No_exception. So change it *)
		   m # message # set_error_exception ()
		     (if err = No_exception then
			Bad_message "Unknown reason (e.g. unexpected eof, timeout)"
		      else
			err
		     )
		 with
		     _ -> ()
		 end;
		 self # critical_postprocessing m;  (* because m is dropped *)
	      )
	      read_queue;
	    
	    Queue.clear read_queue;
	    Queue.clear write_queue;
	    
	    (* Reset state variables *)

	    totally_failed_connections <- 0;
	    pe_trial <- 0;
	    pe_waiting_for_status <- false;
	    self # pe_disable_timer;
	    
	    (* Simply continue with the following piece of code, which will
	     * no nothing.
	     *)

	  end
	  
      end
      else 
	(* This connection breaks the series of total failures (if there 
	 * was such a series.
	 *)
	totally_failed_connections <- 0;

      (* Now examine the queues, and decide what to do. *)
	
      let n_read  = Queue.length read_queue in
      let n_write = Queue.length write_queue in
      if n_read > 0 || n_write > 0 then begin
	assert (n_read >= n_write);
	assert (group = None);
	
	connect_pause <- 0.5;
	
	if n_read = n_write then begin
	  (* Under normal operating conditions:
	   * We have got exactly the same number of responses as we sent
	   * requests. This means that we can simply continue by attaching
	   * again to the event system, and requesting the rest.
	   * 
	   * But...
	   * It is possible that there is another request which is currently
	   * being sent. Because of this we reset the first member of the
	   * write queue, forcing it to be sent again.
	   * This may only happen if the server does not like our request
	   * and closes the connection while the request is being sent.
	   *)
	  assert (Queue.length write_queue > 0);
	  (Queue.peek write_queue) # reset;
	  self # attach_to_esys;         (* Attach again and continue *)
	end
	else begin
	  (* ASSERTION:
	   *     read_queue  = q1 . q2
	   *     write_queue =      q2
	   * i.e. the extra elements of the read queue are at the beginning
	   * of the read queue.
	   * 
	   * PLAN: Make that
	   *     read_queue  = q2 . q1'
	   *     write_queue = q2 . q1'
	   * where q1' are the elements of q1 for which a reconnection is
	   * allowed. Reset the error exception for these elements.
	   * For the other elements (q1 \ q1') leave the error
	   * exception as it is, but change every No_exception into 
	   * No_reply.
	   *)
	  for i = 1 to n_read - n_write do
	    let m_trans = Queue.take read_queue in
	    let m = m_trans # message in
	    (* Test: Are reconnections allowed? *)
	    if m # get_error_counter() <= maximum_number_of_errors then begin
	      let do_reconnect =
		match m # get_reconnect_mode with
		    Send_again -> true
		  | Request_fails -> false
		  | Inquire f ->
		      (* Ask the function 'f' whether to reconnect or not: *)
		      begin 
			try f m    (* returns true or false *)
			with
			    (* The invocation of 'f' may raise an exception.
			     * It is printed to stderr (there is no other
			     * way to report it).
			     *)
			    x ->
			      prerr_string "Exception caught in Http_client: ";
			      prerr_endline (Printexc.to_string x);
			      false
		      end
		      
	      in
	      if do_reconnect then begin
		(* Ok, this request is tried again. *)
		m # set_error_exception () No_exception;  (* reset 'm'... *)
		m_trans # reset;                      (* ...and 'm_trans' *)
		Queue.add m_trans write_queue; 
		                              (* ...and add it to the queue of
		                               * open...
					       *)
		Queue.add m_trans read_queue; 
                                      (* ...and to the unfinished requests. *)
	      end
	      else begin
		(* Drop this message because reconnection is not allowed *)
		(* If the error exception of this message is No_exception,
		 * change it into No_reply. 
		 *)
		begin try
		  ignore(m # get_resp_header());
		  (* It was No_exception. So change it *)
		  m # set_error_exception () No_reply;
		with
		    _ -> ()
		end;
		(* We do not reconnect, so postprocess now. *)
		self # critical_postprocessing m_trans;
	      end
	    end
	    else begin
	      (* drop this message because of too many errors *)
	      (* We do not reconnect, so postprocess now. *)
	      self # critical_postprocessing m_trans;
	    end;
	  done;

	  let n_read  = Queue.length read_queue in
	  let n_write = Queue.length write_queue in
	  assert (n_read = n_write);
	  
	  (* It is now possible that n_read = n_write = 0, in which case
	   * no more is to do, or that there are remaining requests.
	   *)

	  if n_write > 0 then begin
	    assert (Queue.peek read_queue == Queue.peek write_queue);
	    (* Process the queues: *)
	    if group = None then self # attach_to_esys;
	  end
	  else begin
	    (* Nothing to do, so disable all remaining timers: *)
 	    self # pe_disable_timer;
	  end
	end;
      end;


    method critical_postprocessing m =
      critical_section <- true;
      try
	m # postprocess;
	critical_section <- false
      with
	  any ->
	    critical_section <- false;
	    raise any


    (**************** RESET AFTER SERIOUS ERRORS **************************)

    method private reset_after_unix_error err =
      (* Close the socket; clear the Unixqueue *)
      self # abort_connection;

      (* Reset PE timer *)
      self # pe_disable_timer;

      (* Cleanup of the read/write queues: *)
      assert (group = None);
      self # cleanup_on_eof err;

      (* If there were 'add' invocations from callbacks, move these additions
       * to the real queues now.
       *)
      self # leave_critical_section; 

      (* This object should now be in a sane state; i.e. if the Unixqueue
       * is restarted, the remaining requests will be processed as usual.
       *)


    (********************* RESET COMPLETELY *******************************)

    method reset =
      (* Close the socket; clear the Unixqueue *)
      self # abort_connection;

      (* Reset PE timer *)
      self # pe_disable_timer;

      (* Discard all messages on the queues. *)
      
      Queue.iter 
	(fun m ->
	   m # message # set_error_exception 
	     ()
	     No_reply;
	   self # critical_postprocessing m;     (* because m is dropped *)
	)
	read_queue;
	    
      Queue.clear read_queue;
      Queue.clear write_queue;
      
      (* Reset state variables *)

      totally_failed_connections <- 0;
      pe_trial <- 0;
      pe_waiting_for_status <- false;

      (* If there were 'add' invocations from callbacks, move these additions
       * to the real queues now.
       *)
      self # leave_critical_section;



    (*************** HANDLING OF PREMATURE EOF SITUATIONS *****************)

    method private pe_timeout is_enabled () =
      if verbose_connection then
	prerr_endline "HTTP connection: timeout; resuming normal operation";
      
      (* Disable PE handling: *)
      pe_trial <- 0;
      pe_waiting_for_status <- false;

      ( match pe_timer with
	    Some g ->
	      timeout_groups <- List.filter (fun x -> x <> g) timeout_groups;
	  | _ ->
	      assert false;
      );
      pe_timer <- None;

      (* The output handler will be added by 'maintain_polling' if
       * necessary.
       *)
      self # maintain_polling;


    method private pe_disable_timer=
      (* Disable the current timer, if any: *)
      begin match pe_timer with        (* Remove timer *)
	  None -> ()
	| Some g -> self # clear_timeout g
      end;
      pe_timer <- None;

    (**********************************************************************)
    (***                     AUTHENTICATION                             ***)
    (**********************************************************************)

    method private postprocess_complete_message msg_trans =
      (* This method is invoked for every complete reply. The following
       * cases are handled at this stage of processing:
       *
       * - Status code 407: The proxy demands authorization. If the request
       *   already contains credentials for the proxy, this status code
       *   isn't handled here. Otherwise, the request is added again onto
       *   the queue, and a flag ('proxy_credentials_required') is set which 
       *   forces that the proxy credentials must be added for every new 
       *   request.
       *   Note: The necessary authentication header fields are added in
       *   the 'add' method.
       *
       * - Status code 401: The content server demands authorization.
       *   The reply of the server contains the name of the method to be
       *   used. This method is looked up (from 'auth_methods'), and
       *   becomes the new 'current_auth_method'. The credentials to be
       *   sent with the request are computed by this method. The request
       *   is added again to the message queues.
       *   The 'current_auth_method' computes the credentials of every 
       *   request sent from now on to the server (even for requests not
       *   replied with a 401 code). This credentials may be wrong because
       *   there may be several realms on the server demanding different
       *   authentication. 
       *   To avoid endless looping, it is stored in every 'message_transport'
       *   object whether there was already a 401 reply or not. If there
       *   was such a reply, and we get again a 401 code, this situation
       *   is not handled, but passed over to the user of this class.
       *
       * All other status codes are not handled here. Note that it is not
       * possible to react on the redirection codes because this object
       * represents the connection to exactly one server.
       * As default behaviour, the method 'postprocess' of the 
       * message_transport object is invoked; this method incorporates
       * all the intelligence not coded here.
       *)

      let default_action() =
	self # critical_postprocessing msg_trans;
      in

      let scan_auth_header h =
	let p = parse_header_value h in
	begin match p with
	    Word m :: p' ->
	      String.lowercase m, p'
	  | _ -> 
	      raise Not_found
	end
      in

      let msg = msg_trans # message in
      let (_, code, _) = msg # dest_status() in
      match code with
	  407 ->
	    (* --------- Proxy authorization required: ---------- *)
	    if
	      try 
		let _ = msg # assoc_req_header "proxy-authorization" in
		if verbose_status then
		  prerr_endline "HTTP auth: proxy authentication required again";
		false
	      with Not_found -> true
	    then begin
	      (* The request did not contain the "proxy-authorization" header.
	       * Enable proxy authentication if there is a user/password pair.
	       * Otherwise, do the default action.
	       *)
	      if msg # using_proxy then begin
		if verbose_status then
		  prerr_endline "HTTP auth: proxy authentication required";
		if proxy_user <> "" then begin
		  (* We have a user/password pair: Enable proxy authentication
		   * and add 'msg' again to the queue of messages to be
		   * processed.
		   * Note: Following the HTTP protocol standard, the header
		   * of the response contains a 'proxy-authenticate' field
		   * with the authentication method and the realm. This is
		   * not supported; the method is always "basic" and realms
		   * are not distinguished.
		   *)
		  if not proxy_credentials_required then begin
		    proxy_credentials_required <- true;
		    proxy_cookie <- "";
		  end;
		  if verbose_status then
		    prerr_endline "HTTP auth: proxy credentials added";
		  ignore (self # add_again msg_trans);
		end
		else 
		  (* No user/password pair: We cannot authorize ourselves. *)
		  if verbose_status then
		    prerr_endline "HTTP auth: user/password missing";
		  default_action()
	      end
	      else
		(* The server was not contacted as a proxy, but it demanded
		 * proxy authorization. Regard this as an intrusion.
		 *)
		if verbose_status then
		  prerr_endline "HTTP auth: intrusion by proxy authentication";
		default_action()
	    end
	    else 
	      (* The request did already contain "proxy-authenticate". *)
	      default_action()
	      
	| 401 ->
	    (* -------- Content server authorization required: ---------- *)
	    (* If this is the second 401 response, do not react on it. *)
	    if msg_trans # auth_required then begin
	      if verbose_status then
		prerr_endline "HTTP auth: server authentication required again";
	      default_action()
	    end
	    else begin
	      if verbose_status then
		prerr_endline "HTTP auth: server authentication required";
	      (* The first 401 response *)
	      msg_trans # got_401;
	      (* Analyze the received header. If we get a Not_found exception,
	       * analysis failed.
	       * Most of the following functions or methods calls may raise
	       * Not_found, so we do not comment them in detail.
	       *)
	      try
		let challenge = msg # assoc_resp_header "www-authenticate" in
		let auth_method_name, params = scan_auth_header challenge in
		if verbose_status then begin
		  prerr_endline ("HTTP auth: method=" ^ auth_method_name);
		  prerr_endline ("HTTP auth: params=" ^ string_of_token_list params);
		end;
		(* Lookup the authentication method: *)
		let auth_method = List.assoc auth_method_name auth_methods in
		(* Note: 'auth_method' may be the same method as 
		 * 'current_auth_method' or not. This does not matter.
		 *)
		if verbose_status then
		  prerr_endline "HTTP auth: method found";
		(* Update knowledge of the auth method: *)
		auth_method # www_authenticate msg params;
		if verbose_status then
		  prerr_endline "HTTP auth: method analyzed requirements";
		(* Set the credentials: *)
		auth_method # set_authorization msg "authorization";
		if verbose_status then
		  prerr_endline "HTTP auth: method added credentials";
		(* Add the message again to the queue: *)
		let new_msg_trans = self # add_again msg_trans in
		new_msg_trans # got_401;
		(* Make 'auth_method' the new current method: *)
		current_auth_method <- Some auth_method;
		if verbose_status then
		  prerr_endline "HTTP auth: augmented request added";
	      with
		  Not_found -> 
		    default_action()
	    end

	| _ ->
	    (* ----------------------- default: ------------------------- *)
	    default_action()


    (**********************************************************************)
    (***                     THE OUTPUT HANDLER                         ***)
    (**********************************************************************)

    method private handle_output =

      (* Ignore this event if the socket is not Up_rw (this may happen
       * if the output side is closed while there are still output
       * events in the queue):
       *)
      if socket_state <> Up_rw then
	raise Equeue.Reject;

      if verbose_connection then 
	prerr_endline "HTTP connection: Output event!";

      let g = match group with
	  Some x -> x
	| None -> assert false
      in

      if connecting then
	self # check_connection;

      let rec write_loop () =
	let this = Queue.peek write_queue in
	begin match this # state with
	    (Unprocessed | Sending) ->

	      (* TODO: After all requests are processed, the pipeline is 
	       * closed. It must be possible to keep it open for a certain
	       * amount of time. 
	       * Allow the user to specify a timeout value, and in this
	       * case, close the socket only after the timeout happened.
	       * A timeout may be forced.
	       *)

	      (* if no_persistency, set 'connection: close' *)

	      if no_persistency && this # state = Unprocessed then begin
		this # message # set_req_header "connection" "close";
	      end;

	      (* If synchronization = Sync_with_handshake_before_request_body:
	       * transmit only the header of the request, and set
	       * waiting_for_status100. Furthermore, add a timeout
	       * handler that resets this variable after some time.
	       *)

	      let do_handshake =
		match synchronization with
		    Sync_with_handshake_before_request_body timeout ->
		      !waiting_for_status100 = 0
		  | _ ->
		      false
	      in

	      if enable_persistency || sending_first_message then begin
		begin try
		  (* Note: pe_trial > 0 forces that only the header is sent
		   * if the PE handling is activated.
		   *)
		  this # send socket (pe_trial > 0  ||  do_handshake);

		  (* Further PE handling: set pe_waiting_for_status. 
		   * If the header is sent, set pe_waiting_for_status to
		   * true, which forces the output handler to be removed.
		   *)
		  if pe_trial > 0 && this # header_is_sent then begin
		    pe_waiting_for_status <- true;
		    if verbose_connection then
		      prerr_endline "HTTP connection: waiting for status line";
		  end;

		  (* If a handshake is requested: set the variable and the
		   * timer.
		   *)
		  if do_handshake  &&  this # header_is_sent  &&
		     this # state <> Waiting 
		  then begin
		    match synchronization with
			Sync_with_handshake_before_request_body timeout ->
			  let w = ref 1 in
			  waiting_for_status100 <- w;
			  let tm = Unixqueue.new_group esys in
			  timeout_groups <- tm :: timeout_groups;
			  Unixqueue.once
			    esys tm timeout
			    (fun () ->
			       if !w = 1 && w == waiting_for_status100 then (
				 w := 2;
				 self # maintain_polling;
			       );
			       self # clear_timeout tm;
			    );
			  if verbose_connection then
			    prerr_endline "HTTP connection: waiting for 100 CONTINUE";
		      | _ -> assert false
		  end;

		with
		    Unix.Unix_error(Unix.EPIPE,_,_) ->
		      (* Broken pipe: This can happen if the server decides
		       * to close the connection in the same moment when the
		       * client wants to send another request after the 
		       * connection has been idle for a period of time.
		       * Reaction: Close our side of the connection, too,
		       * and open a new connection. The current request will
		       * be silently resent because it is sure that the
		       * request was not received completely; it does not
		       * matter whether the request is idempotent or not.
		       *
		       * Broken pipes are very unlikely because this must 
		       * happen between the 'select' and 'write' system calls.
		       *)
		      if verbose_connection then
			prerr_endline "HTTP connection: broken pipe";
		      this # reset;  (* force that the request is sent again *)
		      raise Broken_pipe

		  | Unix.Unix_error(Unix.EAGAIN,_,_) ->
		      (* This should normally not happen, but there are
		       * operating systems with buggy socket implementations.
		       *)
		      if verbose_connection then
			prerr_endline("HTTP connection: Strange socket behaviour (cannot write although the socket is writeable)");
		      ();  (* Ignore! *)
		      
		  | Unix.Unix_error(e,a,b) as err ->
		      if verbose_connection then
			prerr_endline("HTTP connection: Unix error " ^
				      Unix.error_message e);
		      this # reset;
		      (* Unlike Broken_pipe, it is not clear that the other
		       * side ends the connection, too.
		       * So invoke special termination code.
		       *)
		      self # reset_after_unix_error err;
		      raise (Unix.Unix_error(e,a,b));
		end;
		if sending_first_message && this # state = Waiting then
		  sending_first_message <- false;
		if this # state = Waiting then
		  ignore (Queue.take write_queue);
	      end
	  | (Waiting | Receiving | Complete) ->
	      (* continue with the next message *)
	      ignore (Queue.take write_queue);
	      write_loop()
	end
      in

      let broken_connection = ref false in     (* indicates Broken_pipe *)
      
      begin try
	write_loop()
      with
	  Queue.Empty -> ()
	| Broken_pipe -> broken_connection := true
      end;

      (* Close the connection under the following conditions:
       * - broken pipe
       * - no_persistency is requested
       * - the header "connection: close" was read
       * - the write queue AND the read queue is empty
       *)

      if (Queue.length write_queue = 0 && Queue.length read_queue = 0) or
	 (no_persistency && not sending_first_message) or
         close_connection or
	 !broken_connection
      then begin
	assert (socket_state = Up_rw);

	if verbose_connection then 
	  prerr_endline "HTTP connection: Sending EOF!";

	if not !broken_connection then
	  syscall (fun () -> Unix.shutdown socket Unix.SHUTDOWN_SEND);

	socket_state <- Up_r;

	(* Note on broken pipes:
	 * We can assume that the read side of the socket is still active:
	 * Either the server is still sending its last response, or the
	 * server closed the socket completely, and we will see an EOF.
	 * The case that a proper shutdown of the connection is not possible
	 * at all cannot be handled anyway; in this case we are waiting for
	 * the EOF forever.
	 *)
      end;

      self # maintain_polling;

  end
;;


(**********************************************************************)
(**********************************************************************)
(**********************************************************************)
(***                                                                ***)
(***                 THE PIPELINE INTERFACE                         ***)
(***                                                                ***)
(**********************************************************************)
(**********************************************************************)
(**********************************************************************)

(* The following class, 'pipeline' defines the interface for the outside
 * world.
 *)

class pipeline =
  object (self)
    val mutable esys = Unixqueue.create_unix_event_system()

    val mutable proxy = ""
    val mutable proxy_port = 80
    val mutable proxy_auth = false
    val mutable proxy_user = ""
    val mutable proxy_password = ""

    val mutable www_auth = ( [] : (string * basic_auth_method) list )

    val mutable no_proxy_for = []

    val mutable connections = Hashtbl.create 10

    val mutable open_messages = 0

    val mutable open_connections = 0

    val mutable options =
	    { (* Default values: *)
	      synchronization = Pipeline 5;
	      maximum_connection_failures = 1;
	      maximum_message_errors = 2;
	      inhibit_persistency = false;
	      connection_timeout = 300.0;
	      number_of_parallel_connections = 2;
	      verbose_status = false;
	      verbose_request_header = false;
	      verbose_response_header = false;
	      verbose_request_contents = false;
	      verbose_response_contents = false;
	      verbose_connection = false;
	    }

    method set_event_system new_esys =
      esys <- new_esys;
      Hashtbl.clear connections;

    method add_authentication_method m =
      let name = m # name in
      www_auth <- (name,m) :: listrm name www_auth;

    method set_proxy the_proxy the_port =
      (* proxy="": disables proxy *)
      proxy       <- the_proxy;
      proxy_port  <- the_port;
      ()

    method set_proxy_auth user passwd =
      (* sets 'user' and 'password' if demanded by a proxy *)
      proxy_auth     <- user <> "";
      proxy_user     <- user;
      proxy_password <- passwd


    method avoid_proxy_for l =
      (* l: List of hosts or domains *)
      no_proxy_for <- l


    method set_proxy_from_environment() =
      (* Is the environment variable "http_proxy" set? *)
      let http_proxy =
	try Sys.getenv "http_proxy" with Not_found -> "" in
      begin try
	let (user,password,host,port,path) = 
	  Http_client_aux.match_http http_proxy in
	self # set_proxy (Netencoding.Url.decode host) port;
	match user with
	  Some user_s ->
	    begin match password with
	      Some password_s ->
		self # set_proxy_auth (Netencoding.Url.decode user_s) (Netencoding.Url.decode password_s)
	    | None -> ()
	    end
	| None -> ()
      with
	Not_found -> ()
      end;

      (* Is the environment variable "no_proxy" set? *)
      let no_proxy =
	try Sys.getenv "no_proxy" with Not_found -> "" in
      let no_proxy_list =
	Http_client_aux.split_words_by_commas no_proxy in
      self # avoid_proxy_for no_proxy_list;


    method reset () =
      (* deletes all pending requests; closes connection *)

      (* Reset all connections: *)
      Hashtbl.iter
	(fun _ cl ->
	   List.iter
	     (fun c ->
		c # reset)
	     !cl)
	connections

      

    method private add_with_callback_no_redirection (request : message) f_done =
      request # set_unserved();

      let host = request # get_host() in
      let port = request # get_port() in

      let proxy_allowed = 
	request # is_proxy_allowed() &
	not
          (List.exists
             (fun dom ->
                if dom <> "" &
                   dom.[0] = '.' &
		   String.length host > String.length dom
                then
                  let ld = String.length dom in
                  String.lowercase(String.sub 
                                     host 
                                     (String.length host - ld) 
                                     ld)
                  = String.lowercase dom
                else
                  dom = host)
             no_proxy_for)
      in

      request # prepare (proxy <> "" && proxy_allowed);
      
      (* find out the effective peer: *)
      let peer, peer's_port =
	if proxy = "" || not proxy_allowed then
	  host, port
	else
	  proxy, proxy_port
      in
      
      (* Find out if there is already a connection to this peer: *)

      let conn = 
	let connlist = 
	  try
	    Hashtbl.find connections (peer, peer's_port) 
	  with
	      Not_found ->
		let new_connlist = ref [] in
		Hashtbl.add connections (peer, peer's_port) new_connlist;
		new_connlist
	in
	if List.length !connlist < options.number_of_parallel_connections 
	  then begin
	    let new_conn = new connection
	                     esys
	                     (peer, peer's_port)
	                     (proxy_user, proxy_password) 
			     www_auth
			     options in
	    open_connections <- open_connections + 1;
	    connlist := new_conn :: !connlist;
	    new_conn
	  end 
	  else begin
	    (* Find the connection with the lowest number of queue entries: *)
	    List.fold_left
	      (fun best_conn a_conn ->
		 if a_conn # length < best_conn # length then
		   a_conn
		 else
		   best_conn)
	      (List.hd !connlist)
	      (List.tl !connlist)
	  end
      in
      
      (* Add the request to the queue of this connection: *)

      conn # add request 
	(fun m ->
	   f_done m;
	   (* Update 'open_connections', 'connections', and 'open_messages' *)
	   let l = conn # length in
	   if l = 0 then begin
	     open_connections <- open_connections - 1;
	     let connlist =
	       try
		 Hashtbl.find connections (peer, peer's_port);
	       with
		   Not_found -> assert false
	     in
	     connlist := List.filter (fun c -> c != conn) !connlist;
	     if !connlist = [] then
	       Hashtbl.remove connections (peer, peer's_port);
	   end;
	   self # update_open_messages;
	);

      open_messages <- open_messages + 1;

    method private update_open_messages =
      open_messages <- 0;
      Hashtbl.iter
	(fun _ cl ->
	   List.iter
	     (fun c ->
		open_messages <- open_messages + (c # length))
	     !cl)
	connections;


    method add_with_callback (request : message) f_done =
      self # add_with_callback_no_redirection
	request
	(fun m ->
	   try
	     let (_,code,_) = m # dest_status() in
	     match code with
		 (301|302) ->
		   (* Simply repeat the request with a different URI *)
		   let do_redirection =
		     match m # get_redirect_mode with
			 Redirect -> true
		       | Do_not_redirect -> false
		       | Redirect_inquire f ->
			   (* Ask the function 'f' whether to redirect: *)
			   begin 
			     try f m    (* returns true or false *)
			     with
			     (* The invocation of 'f' may raise an exception.
			      * It is printed to stderr (there is no other
			      * way to report it).
			      *)
				 x ->
				   prerr_string 
				     "Exception caught in Http_client: ";
				   prerr_endline (Printexc.to_string x);
				   false
			   end
		   in

		   if do_redirection then begin
		     let location = m # assoc_resp_header "location" in
		       (* or raise Not_found *)
		     let location' =
		       if location <> "" && location.[0] = '/' then
			 (* Problem: "Location" header must be absolute due
			  * to RFC specs. Now it is relative (with full path).
			  * Workaround: Interpret relative to old server
			  *)
			 let host = m # get_host() in
			 let port = m # get_port() in
			 let prefix =
			   "http://" ^ host ^ 
			   (if port = 80 then "" else ":" ^ string_of_int port)
			 in
			 prefix ^ location
		       else
			 location in
		     m # init_query () location';
		     self # add_with_callback m f_done
		   end
		     else f_done m

	       | _ -> 
		   f_done m
	     with
		 (Http_protocol _ | Not_found) -> 
		   f_done m
	)


    method add request =
      self # add_with_callback request (fun _ -> ())

    method run () =
      (* Runs through the requests in the pipeline. If a request can be
       * fulfilled, i.e. the server sends a response, the status of the
       * request is set and the request is removed from the pipeline.
       * If a request cannot be fulfilled (no response, bad response,
       * network error), an exception is raised and the request remains in
       * the pipeline (and is even the head of the pipeline).
       *
       * Exception Broken_connection:
       *  - The server has closed the connection before the full request
       *    could be sent. It is unclear if something happened or not.
       *    The application should figure out the current state and
       *    retry the request.
       *  - Also raised if only parts of the response have been received
       *    and the server closed the connection. This is the same problem.
       *    Note that this can only be detected if a "content-length" has
       *    been sent or "chunked encoding" was chosen. Should normally
       *    work for persistent connections.
       *  - NOT raised if the server forces a "broken pipe" (normally
       *    indicates a serious server problem). The intention of
       *    Broken_connection is that retrying the request will probably
       *    succeed.
       *)

	 Unixqueue.run esys

    method get_options = options

    method set_options p =
      options <- p;
      Hashtbl.iter
	(fun _ cl ->
	   List.iter
	     (fun c ->
		c # set_options p)
	     !cl)
	connections

    method number_of_open_messages = open_messages

    method number_of_open_connections = open_connections

  end
;;

(**********************************************************************)
(**********************************************************************)
(**********************************************************************)
(***                                                                ***)
(***                 THE CONVENIENCE MODULE                         ***)
(***                                                                ***)
(**********************************************************************)
(**********************************************************************)
(**********************************************************************)

(* This module is intended for beginners and for simple applications
 * of this HTTP implementation.
 *)


module Convenience =
  struct

    class simple_basic_auth_method f =
      object
	inherit basic_auth_method
	method get_credentials () = f current_realm
      end

    class simple_digest_auth_method f =
      object
	inherit digest_auth_method
	method get_credentials () = f current_realm
      end

    let http_url_decode url = Http_client_aux.match_http url

    let http_trials = ref 3
    let http_user = ref ""
    let http_password = ref ""

    let this_user = ref ""
    let this_password = ref ""

    let conv_verbose = ref false

    let auth_basic =
      new simple_basic_auth_method
	(fun realm ->
	  if !this_user <> "" then
	    !this_user, !this_password
	  else
	    if !http_user <> "" then
	      !http_user, !http_password
	    else
	      raise Not_found)

    let auth_digest =
      new simple_digest_auth_method
	(fun realm ->
	  if !this_user <> "" then
	    !this_user, !this_password
	  else
	    if !http_user <> "" then
	      !http_user, !http_password
	    else
	      raise Not_found)


    let get_default_pipe() =

      let p = new pipeline in

      p # set_proxy_from_environment();

      (* Add authentication methods: *)
      p # add_authentication_method auth_basic;
      p # add_authentication_method auth_digest;

      (* That's it: *)
      p


    let pipe = lazy (get_default_pipe())
    let pipe_empty = ref true

    let mutex = Http_client_aux.Mtx.create()

    let request m trials =
      Http_client_aux.Mtx.lock mutex;
      try
	let p = Lazy.force pipe in
	if not !pipe_empty then
	  p # reset();
	p # add_with_callback m (fun _ -> pipe_empty := true);
	pipe_empty := false;
	let rec next_trial todo e =
	  if todo > 0 then begin
	    try
	      p # run()
	    with
	    | Http_error(n,s) as e' ->
		if List.mem n [408; 413; 500; 502; 503; 504 ] then
		  next_trial (todo-1) e'
		else
		  raise e'
	    | e' -> 
		if !conv_verbose then (
		  prerr_endline "HTTP driver: Got exception; trying again";
		  prerr_endline ("HTTP driver: exception was: " ^
				 Printexc.to_string e');
		);
		next_trial (todo-1) e'
	  end
	  else
	    raise e
	in
	next_trial trials (Failure "bad number of http_trials");
	Http_client_aux.Mtx.unlock mutex;
      with
	any ->
	  Http_client_aux.Mtx.unlock mutex;
	  raise any

    let prepare_url url =
      Http_client_aux.Mtx.lock mutex;
      try
	this_user := "";
    	let (user,password,host,port,path) = http_url_decode url in
	begin match user with
	  Some user_s ->
	    this_user := Netencoding.Url.decode user_s;
	    this_password := "";
	    begin match password with
	      Some password_s ->
		this_password := Netencoding.Url.decode password_s
	    | None -> ()
	    end
	| None -> ()
	end;
	Http_client_aux.Mtx.unlock mutex;
	"http://" ^ host ^ ":" ^ string_of_int port ^ path
      with
	Not_found -> 
	  Http_client_aux.Mtx.unlock mutex;
	  url
      |	any -> 
	  Http_client_aux.Mtx.unlock mutex;
	  raise any

    let http_get_message url =
      let m = new get (prepare_url url) in
      request m !http_trials;
      m

    let http_get url = (http_get_message url) # get_resp_body()

    let http_head_message url =
      let m = new head (prepare_url url) in
      request m !http_trials;
      m

    let http_post_message url params =
      let m = new post (prepare_url url) params in
      request m 1;
      m

    let http_post url params = (http_post_message url params) # get_resp_body()

    let http_put_message url content =
      let m = new put (prepare_url url) content in
      request m !http_trials;
      m

    let http_put url content = (http_put_message url content) # get_resp_body()

    let http_delete_message url =
      let m = new delete (prepare_url url) in
      request m 1;
      m

    let http_delete url = (http_delete_message url) # get_resp_body()


    let http_verbose() =
      Http_client_aux.Mtx.lock mutex;
      try
	let p = Lazy.force pipe in
	let opt = p # get_options in
	p # set_options
	  { opt with verbose_status = true;
	             verbose_request_header = true;
		     verbose_response_header = true;
		     verbose_request_contents = true;
		     verbose_response_contents = true;
		     verbose_connection = true 
          };
	conv_verbose := true;
	Http_client_aux.Mtx.unlock mutex;
      with
	any ->
	  Http_client_aux.Mtx.unlock mutex;
	  raise any
  end