(* $Id: ftp_client.ml 145 2005-07-28 00:04:03Z gerd $ *)

open Telnet_client
open Ftp_data_endpoint

exception FTP_protocol_error of exn
exception FTP_protocol_violation of string

let proto_viol s =
  raise(FTP_protocol_violation s)

let verbose = false

type cmd_state =
    [ `Init
    | `Success
    | `Proto_error
    | `Failure
    | `Rename_seq
    | `Restart_seq
    | `User_pass_seq
    | `User_acct_seq
    | `Pass_acct_seq
    | `Preliminary
    ]

type port =
    [ `Active of string * int * Unix.file_descr
    | `Passive of string * int
    | `Unspecified
    ]

type form_code =
    [ `Non_print | `Telnet | `ASA ]

type representation =
    [ `ASCII of form_code option
    | `EBCDIC of form_code option
    | `Image
    ]

type structure =
    [ `File_structure
    | `Record_structure
    ]

type transmission_mode =
    Ftp_data_endpoint.transmission_mode

type state =
    { cmd_state : cmd_state;
      ftp_connected : bool;
      ftp_data_conn : bool;
      ftp_user : string option;
      ftp_password : string option;
      ftp_account : string option;
      ftp_logged_in : bool;
      ftp_port : port;
      ftp_repr : representation;
      ftp_structure : structure;
      ftp_trans : transmission_mode;
      ftp_dir : string list;
    }


type cmd =
    [ `Connect
    | `USER of string
    | `PASS of string
    | `ACCT of string
    | `CWD of string
    | `CDUP
    | `SMNT of string
    | `QUIT
    | `REIN
    | `PORT
    | `PASV
    | `TYPE of representation
    | `STRU of structure
    | `MODE of transmission_mode
    | `RETR of string * (state -> Ftp_data_endpoint.local_receiver)
    | `STOR of string * (state -> Ftp_data_endpoint.local_sender)
    | `STOU of (unit -> Ftp_data_endpoint.local_sender)
    | `APPE of string * (state -> Ftp_data_endpoint.local_sender)
    | `ALLO of int * int option
    | `REST of string
    | `RNFR of string
    | `RNTO of string
    | `DELE of string
    | `RMD of string
    | `MKD of string
    | `PWD
    | `LIST of string option * (state -> Ftp_data_endpoint.local_receiver)
    | `NLST of string option * (state -> Ftp_data_endpoint.local_receiver)
    | `SITE of string
    | `SYST
    | `STAT of string option
    | `HELP of string option
    | `NOOP
    ]

let string_of_cmd =
  function
    | `Connect -> ""
    | `USER s -> "USER " ^ s ^ "\r\n"
    | `PASS s -> "PASS " ^ s ^ "\r\n"
    | `ACCT s -> "ACCT " ^ s ^ "\r\n"
    | `CWD s  -> "CWD " ^ s ^ "\r\n"
    | `CDUP   -> "CDUP\r\n"
    | `SMNT s -> "SMNT " ^ s ^ "\r\n"
    | `QUIT   -> "QUIT\r\n"
    | `REIN   -> "REIN\r\n"
    | `PORT   -> assert false   (* not done here *)
    | `PASV   -> "PASV\r\n"
    | `TYPE t -> "TYPE " ^ 
	         ( match t with
		     | `ASCII None -> "A"
		     | `ASCII (Some `Non_print) -> "A N"
		     | `ASCII (Some `Telnet) -> "A T"
		     | `ASCII (Some `ASA) -> "A C"
		     | `EBCDIC None -> "E"
		     | `EBCDIC (Some `Non_print) -> "E N"
		     | `EBCDIC (Some `Telnet) -> "E T"
		     | `EBCDIC (Some `ASA) -> "E C"
		     | `Image -> "I"
		 ) ^ "\r\n"
    | `STRU `File_structure -> "STRU F\r\n"
    | `STRU `Record_structure -> "STRU R\r\n"
    | `MODE `Stream_mode -> "MODE S\r\n"
    | `MODE `Block_mode -> "MODE B\r\n"
    | `RETR (s,_) -> "RETR " ^ s ^ "\r\n"
    | `STOR (s,_) -> "STOR " ^ s ^ "\r\n"
    | `STOU _     -> "STOU\r\n"
    | `APPE (s,_) -> "APPE " ^ s ^ "\r\n"
    | `ALLO(n,r)  -> "ALLO " ^ string_of_int n ^ 
	             (match r with
			| None -> ""
			| Some m -> " R " ^ string_of_int m) ^ "\r\n"
    | `REST s -> "REST " ^ s ^ "\r\n"
    | `RNFR s -> "RNFR " ^ s ^ "\r\n"
    | `RNTO s -> "RNTO " ^ s ^ "\r\n"
    | `DELE s -> "DELE " ^ s ^ "\r\n"
    | `RMD  s -> "RMD " ^ s ^ "\r\n"
    | `MKD  s -> "MKD " ^ s ^ "\r\n"
    | `PWD    -> "PWD\r\n"
    | `LIST(None,_) -> "LIST\r\n"
    | `LIST(Some s,_) -> "LIST " ^ s ^ "\r\n"
    | `NLST(None,_) -> "NLST\r\n"
    | `NLST(Some s,_) -> "NLST " ^ s ^ "\r\n"
    | `SITE s -> "SITE " ^ s ^ "\r\n"
    | `SYST   -> "SYST\r\n"
    | `STAT None -> "STAT\r\n"
    | `STAT(Some s) -> "STAT " ^ s ^ "\r\n"
    | `HELP None -> "HELP\r\n"
    | `HELP(Some s) -> "HELP " ^ s ^ "\r\n"
    | `NOOP -> "NOOP\r\n"


let port_re = Netstring_pcre.regexp ".*[^0-9](\\d+),(\\d+),(\\d+),(\\d+),(\\d+),(\\d+)"

let extract_port s =
  match Netstring_pcre.string_match port_re s 0 with
    | None ->
	proto_viol "Cannot parse specification of passive port"
    | Some m ->
	let h1 = Netstring_pcre.matched_group m 1 s in
	let h2 = Netstring_pcre.matched_group m 2 s in
	let h3 = Netstring_pcre.matched_group m 3 s in
	let h4 = Netstring_pcre.matched_group m 4 s in
	let p1 = Netstring_pcre.matched_group m 5 s in
	let p2 = Netstring_pcre.matched_group m 6 s in
	let p = int_of_string p1 * 256 + int_of_string p2 in
	(h1 ^ "." ^ h2 ^ "." ^ h3 ^ "." ^ h4, p)

let addr_re = Netstring_pcre.regexp "^(\\d+)\\.(\\d+)\\.(\\d+)\\.(\\d+)$"

let format_port (addr,p) =
  match Netstring_pcre.string_match addr_re addr 0 with
    | None ->
	failwith "Bad IP address"
    | Some m ->
	let h1 = Netstring_pcre.matched_group m 1 addr in
	let h2 = Netstring_pcre.matched_group m 2 addr in
	let h3 = Netstring_pcre.matched_group m 3 addr in
	let h4 = Netstring_pcre.matched_group m 4 addr in
	let p1 = string_of_int(p lsr 8) in
	let p2 = string_of_int(p land 0xff) in
	h1 ^ "," ^ h2 ^ "," ^ h3 ^ "," ^ h4 ^ "," ^ p1 ^ "," ^ p2

let set_ftp_port state value =
  ( match state.ftp_port with
      | `Active(_,_,fd) ->
	  Unix.close fd
      | _ ->
	  ()
  );
  { state with ftp_port = value }


type reply = int * string
    (* Reply code plus text *)


let init_state s =
  let s_name =
    match Unix.getsockname s with
      | Unix.ADDR_INET(addr,_) ->
	  Unix.string_of_inet_addr addr
      | _ ->
	  failwith "Not an internet socket"
  in
  { cmd_state = `Init;
    ftp_connected = true;
    ftp_data_conn = false;
    ftp_user = None;
    ftp_password = None;
    ftp_account = None;
    ftp_logged_in = false;
    ftp_port = `Unspecified;
    ftp_repr = `ASCII None;
    ftp_structure = `File_structure;
    ftp_trans = `Stream_mode;
    ftp_dir = [];
  }


let start_reply_re = Netstring_pcre.regexp "^[0-9][0-9][0-9]-"
let end_reply_re = Netstring_pcre.regexp "^[0-9][0-9][0-9] "


let is_active state =
  match state.ftp_port with
    | `Active _ -> true
    | _ -> false

let is_passive state =
  match state.ftp_port with
    | `Passive _ -> true
    | _ -> false

class ftp_client_pi
        ?(event_system = Unixqueue.create_unix_event_system())
        ?(onempty = fun _ -> ())
        ?(onclose = fun () -> ())
        sock =
  let ctrl_input_buffer = Netbuffer.create 500 in
  let ctrl_input, ctrl_input_shutdown = 
    Netchannels.create_input_netbuffer ctrl_input_buffer in
object(self)

  val queue = Queue.create()

  val mutable state = init_state sock

  val mutable data_engine = None

  val ctrl = new telnet_session

  val reply_text = Buffer.create 200
  val mutable reply_code = (-1)
  val mutable reply_callback = (fun _ _ -> ())

  val mutable interaction_state = 
    ( `Waiting `Connect 
	: [ `Ready 
          | `Connecting_pasv of cmd * Uq_engines.connect_status Uq_engines.engine
	  | `Listening_actv of cmd * Unixqueue.event Uq_engines.engine
          | `Transfer of cmd
	  | `Transfer_replied of cmd * int * string
	  | `Waiting of cmd
	  ] )
      (* `Ready: another command can be sent now
       * `Connecting_pasv: In passive mode, we are connecting to the
       *    remote port (done by the argument engine). This state is
       *    skipped when we are still connected.
       * `Listening_actv: In active mode, we are listening for the
       *    connect (done by the argument engine). This state is
       *    skipped when we are still connected.
       * `Transfer: a data transfer is in progress
       * `Transfer_replied: while the rest of the transfer is not yet
       *    done, the server already sent a reply
       * `Waiting: it is waited for the reply
       *)



  initializer (
    ctrl # set_connection (Telnet_socket sock);
    ctrl # set_event_system event_system;
    ctrl # set_callback self#receive_ctrl_reply;
    ctrl # attach();
  )

  method state = state

  method private receive_ctrl_reply got_synch =
    while not (Queue.is_empty ctrl#input_queue) do
      let tc = Queue.take ctrl#input_queue in
      match tc with
	| Telnet_data data ->
	    Netbuffer.add_string ctrl_input_buffer data;
	    self # parse_ctrl_reply()

	| Telnet_nop ->
	    ()

	| Telnet_will _
	| Telnet_wont _
	| Telnet_do _
	| Telnet_dont _ ->
	    ctrl # process_option_command tc

	| Telnet_sb _ 
	| Telnet_se ->
	    ()    (* Ignore subnegotiation *)

	| Telnet_eof ->
	    ctrl_input_shutdown();
	    self # parse_ctrl_reply()

	| Telnet_timeout ->
	    ()  (* TODO *)

	| _ ->
	    (* Unexpected telnet command *)
	    proto_viol "Unexpected command on Telnet level"
    done

  method private parse_ctrl_reply() =
    try
      while true do
	let line = ctrl_input # input_line() in  (* or exception! *)
	if verbose then prerr_endline ("< " ^ line);
	if Netstring_pcre.string_match start_reply_re line 0 <> None then (
	  let code = int_of_string (String.sub line 0 3) in
	  if reply_code <> (-1) && reply_code <> code then 
	    proto_viol "Parse error of control message";
	  reply_code <- code;
	  Buffer.add_string reply_text line;
	  Buffer.add_string reply_text "\n";
	)
	else
	  if Netstring_pcre.string_match end_reply_re line 0 <> None then (
	    let code = int_of_string (String.sub line 0 3) in
	    if reply_code <> (-1) && reply_code <> code then
	      proto_viol "Parse error of control message";
	    Buffer.add_string reply_text line;
	    Buffer.add_string reply_text "\n";
	    let text = Buffer.contents reply_text in
	    reply_code <- (-1);
	    Buffer.clear reply_text;
	    self # interpret_ctrl_reply code text;
	  )
	  else (
	    if reply_code = (-1) then
	      proto_viol "Parse error of control message"
	    Buffer.add_string reply_text line;
	    Buffer.add_string reply_text "\n";
	  )
      done
    with
      | Netchannels.Buffer_underrun ->
	  (* No complete line yet *)
	  ()
      | End_of_file ->
	  onclose()

  method private interpret_ctrl_reply code text =
    (* This method is called whenever a reply has been completely received.
     * This may happen in a number of situations:
     * - As greeting message
     * - As regular response to a sent FTP command
     * - When the data transfer is over. Note that the control response
     *   may be received before the end of the transfer is seen by the
     *   client.
     * - Within the data transfer
     * - At any other point in time, but this is regarded as protocol error
     *)
    let reply st cmd_state =
      let st' = { st with cmd_state = cmd_state } in
      state <- st';
      reply_callback st' (code,text)
    in
    let ready() =
      interaction_state <- `Ready in
    ( match interaction_state with
	| `Ready ->
	    proto_viol "Spontaneous control message"
	| `Waiting `Connect ->
	    (match code with
	       | 120 -> reply state `Preliminary
	       | 220 -> ready(); reply state `Success
	       | n when n >= 400 && n <= 599 ->
		   ready(); reply state `Failure
	       | _   -> proto_viol "Unexpected control message"
	    )
	| `Waiting (`USER s) ->
	    ( match code with
		| 230 -> 
		    ready(); 
                    reply { state with 
			      ftp_user = Some s;
			      ftp_password = None;
			      ftp_account = None;
			      ftp_logged_in = true } `Success
		| 530 -> 
		    ready();
		    reply { state with
			      ftp_logged_in = false } `Failure
		| 331 -> 
		    ready(); 
		    reply { state with 
			      ftp_user = Some s;
			      ftp_password = None;
			      ftp_account = None;
			      ftp_logged_in = false } `User_pass_seq
		| 332 -> 
		    ready();
		    reply { state with 
			      ftp_user = Some s;
			      ftp_password = None;
			      ftp_account = None;
			      ftp_logged_in = false } `User_acct_seq
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`PASS s) ->
	    ( match code with
		| 202 | 230 -> 
		    ready(); 
		    reply { state with
			      ftp_password = Some s;
			      ftp_account = None;
			      ftp_logged_in = true } `Success
		| 530 ->
		    ready();
		    reply { state with
			      ftp_logged_in = false } `Failure
		| 332 ->  
		    ready();
		    reply { state with 
			      ftp_password = Some s;
			      ftp_account = None;
			      ftp_logged_in = false } `Pass_acct_seq
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`ACCT s) ->
	    ( match code with
		| 202 | 230 -> 
		    ready(); 
		    reply { state with
			      ftp_account = Some s;
			      ftp_logged_in = true } `Success
		| 530 -> 
		    ready();
		    reply { state with
			      ftp_logged_in = false } `Failure
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`CWD s) ->
	    ( match code with
		| 200 | 250 -> 
		    ready(); 
		    let state' =
		      { state with ftp_dir = s :: state.ftp_dir } in
		    reply state' `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `CDUP ->
	    ( match code with
		| 200 | 250 -> 
		    ready();
		    let state' =
		      match state.ftp_dir with
			| [] -> state
			| _ :: dir ->
			    { state with ftp_dir = dir } in
		    reply state' `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`SMNT s) ->
	    ( match code with
		| 200 | 202 | 250 -> 
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `REIN ->
	    ( match code with
		| 120 -> 
		    reply state `Preliminary
		| 220 -> 
		    ready(); reply (init_state sock) `Success
		    (* CHECK: Close data connection? *)
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `QUIT ->
	    ( match code with
		| 200 | 221 -> 
		    reply state `Success;
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> proto_viol "Unexpected control message"
	    )
	| `Waiting `PORT ->
	    ( match code with
		| 200 -> 
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `PASV ->
	    ( match code with
		| 227 ->
		    let (addr,port) = extract_port text in
		    ready();
		    self # close_connection();
		    reply (set_ftp_port state (`Passive(addr,port))) `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`TYPE t) ->
	    ( match code with
		| 200 ->
		    ready();
		    reply { state with ftp_repr = t } `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`MODE m) ->
	    ( match code with
		| 200 ->
		    ready();
		    reply { state with ftp_trans = m } `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`STRU s) ->
	    ( match code with
		| 200 ->
		    ready();
		    reply { state with ftp_structure = s } `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`ALLO _) ->
	    ( match code with
		| 200 | 202 ->
		    ready();
		    reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`REST _) ->
	    ( match code with
		| 350 ->
		    ready();
		    reply state `Restart_seq
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`RNFR _) ->
	    ( match code with
		| 350 ->
		    ready(); reply state `Rename_seq
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`RNTO _) ->
	    ( match code with
		| 200 | 250 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`DELE _) ->
	    ( match code with
		| 200 | 250 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`RMD _) ->
	    ( match code with
		| 200 | 250 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`MKD _) ->
	    ( match code with
		| 200 | 250 | 257 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `PWD ->
	    ( match code with
		| 200 | 250 | 257 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
 	| `Waiting `SYST ->
	    ( match code with
		| 215 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
 	| `Waiting (`STAT _) ->
	    ( match code with
		| 211 | 212 | 213 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`HELP _) ->
	    ( match code with
		| 211 | 214 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting (`SITE _) ->
	    ( match code with
		| 200 | 202 | 250 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	| `Waiting `NOOP ->
	    ( match code with
		| 200 ->
		    ready(); reply state `Success
		| n when n >= 400 && n <= 599 ->
		    ready(); reply state `Failure
		| _   -> 
		    proto_viol "Unexpected control message"
	    )
	(* `STOR |`STOU | `APPE left out for now *)

	| `Connecting_pasv(_, conn_engine) ->
	    (* Somewhat unexpected reply! *)
	    ( match code with
		| 125 | 150 ->
		    reply state `Preliminary
		| n when n >= 400 && n <= 599 ->
		    conn_engine # abort();
		    ready();
		    reply state `Failure
		| _ ->
		    conn_engine # abort();
		    proto_viol "Unexpected control message"
	    )
	| `Listening_actv(_, acc_engine) ->
	    (* Somewhat unexpected reply! *)
	    ( match code with
		| 125 | 150 ->
		    reply state `Preliminary
		| n when n >= 400 && n <= 599 ->
		    acc_engine # abort();
		    ready();
		    reply state `Failure
		| _ ->
		    acc_engine # abort();
		    proto_viol "Unexpected control message"
	    )
	| `Transfer cmd ->
	    (* The transfer probably ends in the near future, just record
             * the reply, and wait for the end of the transfer.
             *)
	    ( match code with
		| 125 | 150 ->
		    reply state `Preliminary
		| n when n >= 400 && n <= 599 ->
		    self # close_connection();
		    ready();
		    reply state `Failure
		| _ ->
		    interaction_state <- `Transfer_replied(cmd,code,text)
	    )
	| `Transfer_replied cmd ->
	    (* Another reply! This is an error. *)
	    proto_viol "Unexpected control message"
	| `Waiting(`RETR(_,f))
	| `Waiting(`LIST(_,f))
	| `Waiting(`NLST(_,f)) ->
	    (* This state is only possible when the transfer has already
             * been completed.
             *)
	    ( match data_engine with
		| None -> ()  (* strange *)
		| Some e -> 
		    if e # descr_state <> `Clean then self # close_connection()
	    );
	    ( match code with
		| 125 | 150 ->
		    reply state `Preliminary
		| 226 ->
		    self # close_connection();
 		    ready();
		    reply state `Success
		| 250 ->
 		    ready();
		    reply state `Success
		| n when n >= 400 && n <= 599 ->
		    self # close_connection();
		    reply state `Failure
		| _ ->
		    proto_viol "Unexpected control message"
	    )
	| _ -> assert false

    );
    self # send_command_when_ready()


  method private send_command_when_ready() =
    if interaction_state = `Ready then (
      try
	assert(reply_code = (-1));
	let (cmd, onreply) = Queue.take queue in  (* or Queue.Empty *)
	interaction_state <- `Waiting cmd;
	( match cmd with
	    | `RETR(_,f)
	    | `LIST(_,f)
	    | `NLST(_,f) ->
		( match state.ftp_port with
		    | `Passive(_,_) ->
			(* In passive mode, connect now: *)
			self # setup_passive_receiver cmd f
			
		    | `Active(_,_,_) ->
			(* In active mode, accept the connection now *)
			self # setup_active_receiver cmd f

		    | `Unspecified ->
			failwith "FTP client: Usage error, one must send `PORT or `PASV before the transfer"
		)
	    | _ -> ()
	);
	let line = 
	  match cmd with
	    | `PORT ->
		let server_sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
		let addr =  (* of control connection *)
		  match Unix.getsockname sock with
		    | Unix.ADDR_INET(addr,_) -> addr
		    | _ -> assert false in
		let addr_str = Unix.string_of_inet_addr addr in
		Unix.bind server_sock (Unix.ADDR_INET(addr,0));
		Unix.listen server_sock 1;
		let port =
		  match Unix.getsockname server_sock with
		    | Unix.ADDR_INET(_,port) -> port
		    | _ -> assert false in
		state <- set_ftp_port state (`Active(addr_str,port,server_sock));
		let port_str = format_port (addr_str,port) in
		"PORT " ^ port_str ^ "\r\n"
	    | _ -> string_of_cmd cmd in
	if verbose then prerr_endline ("> " ^ line);
	Queue.push (Telnet_data line) ctrl#output_queue;
	ctrl # update();
	reply_callback <- onreply;
      with
	  Queue.Empty ->
	    onempty state
    )


  method private close_connection() =
    (* Terminates any transfer immediately and closes the connection *)
    match data_engine with
      | None -> ()
      | Some e -> 
	  ( match e # state with
	      | `Working _ -> e # abort()
	      | _ -> ()
	  );
	  let data_sock = e # descr in
	  Unix.close data_sock;
	  data_engine <- None;
	  state <- set_ftp_port state `Unspecified



  method private setup_passive_receiver cmd f =
    assert(is_passive state);
    ( match data_engine with
	| Some e ->
	    (* Continue with the old connection *)
	    if e # descr_state <> `Clean then
	      proto_viol "Data connection not clean";
	    (* Create a new engine taking the connection over *)
	    let data_sock = e # descr in
	    self # setup_receiver 
	      ~is_done:(fun () ->
			  match interaction_state with
			    | `Transfer_replied(_,c,t) ->
				interaction_state <-
				  `Waiting cmd;
				self # interpret_ctrl_reply c t
			    | `Transfer cmd ->
				interaction_state <-
				  `Waiting cmd
			    | _ -> assert false
		       )
	      ~is_error:(fun err ->
			   raise(FTP_protocol_error err))
	      f data_sock;
	    interaction_state <- `Transfer cmd
	| None ->
	    (* Indicates that a connection is to be opened *)
	    let addr,port =
	      match state.ftp_port with
		| `Passive(a,p) -> a,p | _ -> assert false in
	    let conn_engine =
	      Uq_engines.connector
		(`Socket(
		   `Sock_inet(Unix.SOCK_STREAM,
			      Unix.inet_addr_of_string addr,
			      port),
		   Uq_engines.default_connect_options))
		event_system in
	    Uq_engines.when_state
	      ~is_done:(function
			  | `Socket(data_sock,_) ->
			      self # setup_receiver 
				~is_done:(fun () ->
					    match interaction_state with
					      | `Transfer_replied(_,c,t) ->
						  interaction_state <-
						    `Waiting cmd;
						  self # interpret_ctrl_reply c t
					      | `Transfer cmd ->
						  interaction_state <-
						    `Waiting cmd
					      | _ -> assert false
					 )
				~is_error:(fun err ->
					     raise(FTP_protocol_error err))
				f data_sock;
			      interaction_state <- `Transfer cmd;
			  | _ -> assert false
		       )
	      ~is_error:(fun err -> 
			   raise(FTP_protocol_error err))
	      conn_engine;
	    interaction_state <- `Connecting_pasv(cmd,conn_engine);
    )

  method private setup_active_receiver cmd f =
    assert(is_active state);
    ( match data_engine with
	| Some e ->
	    (* Continue with the old connection *)
	    if e # descr_state <> `Clean then
	      proto_viol "Data connection not clean";
	    (* Create a new engine taking the connection over *)
	    let data_sock = e # descr in
	    self # setup_receiver 
	      ~is_done:(fun () ->
			  match interaction_state with
			    | `Transfer_replied(_,c,t) ->
				interaction_state <-
				  `Waiting cmd;
				self # interpret_ctrl_reply c t
			    | `Transfer cmd ->
				interaction_state <-
				  `Waiting cmd
			    | _ -> assert false
		       )
	      ~is_error:(fun err ->
			   raise(FTP_protocol_error err))
	      f data_sock;
	    interaction_state <- `Transfer cmd
	| None ->
	    (* Indicates that a connection is to be opened *)
	    let addr,port,server_sock =
	      match state.ftp_port with
		| `Active(a,p,fd) -> a,p,fd | _ -> assert false in
	    let acc_engine =
	      new Uq_engines.poll_engine 
		[ (Unixqueue.Wait_in server_sock), (-1.0) ] event_system in
	    Uq_engines.when_state
	      ~is_done:(function
			  | Unixqueue.Input_arrived(_,_) ->
			      let data_sock, _ = Unix.accept server_sock in
			      self # setup_receiver 
				~is_done:(fun () ->
					    match interaction_state with
					      | `Transfer_replied(_,c,t) ->
						  interaction_state <-
						    `Waiting cmd;
						  self # interpret_ctrl_reply c t
					      | `Transfer cmd ->
						  interaction_state <-
						    `Waiting cmd
					      | _ -> assert false
					 )
				~is_error:(fun err ->
					     raise(FTP_protocol_error err))
				f data_sock;
			      interaction_state <- `Transfer cmd;
			  | _ ->
			      assert false
		       )
	      ~is_error:(fun err -> 
			   raise(FTP_protocol_error err))
	      acc_engine;
            let acc_engine = (acc_engine :> Unixqueue.event Uq_engines.engine) in
	    interaction_state <- `Listening_actv(cmd,acc_engine);
    )


  method private setup_receiver ~is_done ~is_error f data_sock =
    let local = f state in
    let e = 
      new ftp_data_receiver
	~esys:event_system
	~mode:state.ftp_trans
	~local_receiver:local
	~descr:data_sock () in
    data_engine <- Some (e :> ftp_data_engine);
    Uq_engines.when_state 
      ~is_done
      ~is_error
      e

	

  method add_cmd ?(onreply = fun _ _ -> ()) (cmd : cmd) =
    Queue.push (cmd, onreply) queue;
    self # send_command_when_ready()


  method send_abort () = ()

  method run () = Unixqueue.run event_system

end


class type ftp_method =
object
  method connect : (string * int) option
  method execute : ftp_client_pi -> state -> unit
end


exception FTP_method_failure of int * string


class connect_method ~host ?(port = 21) () : ftp_method =
object(self)
  method connect = Some(host,port)
  method execute _ _ = ()
end


class login_method ~user ~get_password ~get_account () : ftp_method =
object(self)
  method connect = None
  method execute (c:ftp_client_pi) state =
    let ( >> ) cmd f = c # add_cmd ~onreply:f cmd in
    (`USER user) >> 
      (fun state (c,t) ->
	 match state.cmd_state with
	   | `Preliminary -> ()
	   | `Success -> ()
	   | `User_pass_seq ->
	       (`PASS (get_password())) >>
		 (fun state (c,t) ->
		    match state.cmd_state with
		      | `Preliminary -> ()
		      | `Success -> ()
		      | `Pass_acct_seq ->
			  (`ACCT (get_account())) >>
			    (fun state (c,t) ->
			       match state.cmd_state with
				 | `Preliminary -> ()
				 | `Success -> ()
				 | `Failure -> raise(FTP_method_failure (c,t))
				 | _ -> assert false
			    )
		      | `Failure -> raise(FTP_method_failure (c,t))
		      | _ -> assert false
		 )
	   | `User_acct_seq ->
	       (`ACCT (get_account())) >>
		 (fun state (c,t) ->
		    match state.cmd_state with
		      | `Preliminary -> ()
		      | `Success -> ()
		      | `Failure -> raise(FTP_method_failure (c,t))
		      | _ -> assert false
		 )
	   | `Failure ->
	       raise(FTP_method_failure (c,t))
	   | _ ->
	       assert false
      )
end


let slash_re = Netstring_pcre.regexp "/+";;


let rec is_prefix l1 l2 =
  match (l1, l2) with
    | (x1::l1'), (x2::l2') ->
	x1 = x2 && is_prefix l1' l2' 
    | [], _ ->
	true
    | (_::_), [] ->
	false


let rec without_prefix l1 l2 =
  match (l1, l2) with
    | (x1::l1'), (x2::l2') ->
	if x1 = x2 then without_prefix l1' l2' else failwith "without_prefix"
    | [], _ ->
	l2
    | (_::_), [] ->
	failwith "without_prefix"


class get_method ~file ~representation ~store () : ftp_method =
  let path = Netstring_pcre.split slash_re file in
object(self)
  method connect = None
    
  method execute (c : ftp_client_pi) state =
    let ( >> ) cmd f = c # add_cmd ~onreply:f cmd in
    let ( >>> ) cmd f = 
      cmd >>
	( fun state (c,t) ->
	    match state.cmd_state with
	      | `Preliminary -> ()
	      | `Success -> f state (c,t)
	      | `Failure -> raise(FTP_method_failure (c,t))
	      | _ -> assert false
	) in
    
    let rec walk_to_directory f state =
      let cur_dir = List.rev state.ftp_dir in
      if is_prefix cur_dir path then
	match without_prefix cur_dir path with
	  | [] ->
	      failwith "Bad filename"
	  | [ filename ] ->
	      f state filename
	  | dir :: _  ->
	      (`CWD dir) >> (walk_to_directory_check f)
      else
	`CDUP >> (walk_to_directory_check f)

    and walk_to_directory_check f state (c,t) =
      match state.cmd_state with
	| `Preliminary -> ()
	| `Success -> walk_to_directory f state
	| `Failure -> raise(FTP_method_failure (c,t))
	| _ -> assert false
    in

    walk_to_directory
      (fun state filename ->
	 (`TYPE representation) >>>
	   (fun state reply ->
	      `PASV >>>
		(fun state reply ->
		   (`RETR (filename,store)) >>> (fun _ _ -> ())
		)
	   )
      )
      state

end


class ftp_client 
        ?(event_system = Unixqueue.create_unix_event_system())
        ?(onempty = fun () -> ())
        () =
object(self)
  val mutable pi_opt = (None : ftp_client_pi option)
  val mutable queue = Queue.create()
  val mutable current = None

  method private next() =
    try
      let (ftpm, onsuccess, onerror) = Queue.take queue in (* or Queue.Empty *)
      current <- Some(ftpm,onsuccess,onerror);
      (* First check whether we have to connect again: *)
      ( match ftpm # connect with
	  | None ->
	      ( match pi_opt with
		  | None -> failwith "Missing connection"
		  | Some pi -> self # exec pi ftpm onsuccess onerror
	      )
	  | Some (host,port) ->
	      (* First stop the current connection *)
	      ( match pi_opt with
		  | None -> ()
		  | Some pi ->
		      pi # add_cmd `QUIT;
		      (* TODO: fallback method if QUIT fails *)
		      pi_opt <- None
	      );
	      let conn_engine =
		Uq_engines.connector
		  (`Socket(
		     `Sock_inet(Unix.SOCK_STREAM,
				Unix.inet_addr_of_string host,
				port),
		     Uq_engines.default_connect_options))
		  event_system in
	      Uq_engines.when_state
		~is_done:(function
			    | `Socket(sock,_) ->
				let pi = 
				  new ftp_client_pi
				    ~event_system
				    ~onempty:self#pi_got_empty
				    sock in
				pi_opt <- Some pi;
				self # exec pi ftpm onsuccess onerror
			    | _ -> assert false
			 )
		~is_error:(fun err -> raise(FTP_protocol_error err))
		conn_engine
      )
    with
	Queue.Empty ->
	  onempty();
	  ( match pi_opt with
	      | None -> ()
	      | Some pi ->
		  pi # add_cmd `QUIT;
		  (* TODO: fallback method if QUIT fails *)
		  pi_opt <- None
	  );

	  

  method private exec pi ftpm onsuccess onerror =
    let state = pi # state in
    ftpm # execute pi state

  method private pi_got_empty state =
    (* First, call the onsuccess function of the current method *)
    ( match current with
	| None -> ()
	| Some(_,onsuccess,_) -> 
	    current <- None;
	    onsuccess()
    );
    (* Go on, and do the next method *)
    self # next()
    
  method add ?(onsuccess = fun () -> ()) 
             ?(onerror = fun (_ : exn) -> ()) 
             (ftpm : ftp_method) =
    Queue.add (ftpm, onsuccess, onerror) queue;
    if current = None then self # next();


  method run () = Unixqueue.run event_system

end