Source file netstream.ml

(* $Id$
 * ----------------------------------------------------------------------
 *
 *)


open Netchannels;;


class type in_obj_stream =
object
  inherit Netchannels.in_obj_channel
  method block_size : int
  method window : Netbuffer.t
  method want : int -> unit
  method want_another_block : unit -> unit
  method window_length : int
  method window_at_eof : bool
  method skip : int -> unit
end


class virtual input_methods init_s_netbuf =
object(self)
  val mutable s_pos = 0
  val mutable s_at_eof = false
  val s_netbuf = init_s_netbuf
  val mutable s_closed = false


  method virtual want : int -> unit

  method virtual want_another_block : unit -> unit

  method virtual window_length : int

  method virtual input : Bytes.t -> int -> int -> int


  (* The following input methods base all on [input] *)

  method really_input buf pos len =
    if s_closed then raise Netchannels.Closed_channel;
    let rec read p =
      let l = self # input buf (pos+p) (len-p) in
      let p' = p + l in
      if p' = len then
	()
      else (
	if l=0 then raise Sys_blocked_io;
	read p'
      )
    in
    self # want len;  (* may raise Buffer_underrun *)
    read 0

  method really_input_string len =
    let buf = Bytes.create len in
    self # really_input buf 0 len;
    Bytes.unsafe_to_string buf


  method input_char () =
    let s = Bytes.create 1 in
    self # really_input s 0 1;
    Bytes.get s 0


  method input_byte () =
    let s = Bytes.create 1 in
    self # really_input s 0 1;
    int_of_char (Bytes.get s 0)


  method input_line () =
    (* CHECK: Are the different end of line conventions important here? *)
    let rec find_eol() =
      try
	Netbuffer.index_from s_netbuf 0 '\n'    (* or Not_found *)
      with
	  Not_found ->
	    if not s_at_eof then begin
	      self # want_another_block();  (* may raise Buffer_underrun *)
	      find_eol()
	    end
	    else self#window_length
    in
    if s_closed then raise Netchannels.Closed_channel;
    let n = find_eol() in
    if n >= self#window_length then begin
      if n = 0 then raise End_of_file;
      self#really_input_string n
    end
    else begin
      let s = self#really_input_string n in
      ignore(self#input_char());    (* '\n' *)
      s
    end
    

  method pos_in =
    if s_closed then raise Netchannels.Closed_channel;
    s_pos


end


class input_stream ?len ?(block_size = 4096) in_ch : in_obj_stream =
object (self)
  val s_channel = (in_ch : in_obj_channel)
  val s_maxlength = len
  val s_blocksize = block_size
  val mutable s_underrun = false

  inherit input_methods (Netbuffer.create block_size)

  (* Note: This implementation must even work if [in_ch] is a pipe,
   * and raises Buffer_underrun from time to time. This may happen
   * at inconvenient situations. In this case the flag s_underrun stores 
   * whether an underrun happened, and should be reported later.
   *)

  initializer 
    try
      self # want_minimum()   (* may raise Buffer_underrun *)
    with
	Buffer_underrun ->
	  s_underrun <- true

  method private debug msg =
    prerr_endline (msg ^ ": s_pos=" ^ string_of_int s_pos ^ 
		   " s_at_eof=" ^ string_of_bool s_at_eof ^ 
		   " buflen=" ^ string_of_int (Netbuffer.length s_netbuf) ^
		   " s_closed=" ^ string_of_bool s_closed);

  method block_size = s_blocksize
    (* The block size is a static property, so never raise Closed_channel *)

  method window = 
    if s_closed then raise Netchannels.Closed_channel;
    s_netbuf

  method window_length = 
    if s_closed then raise Netchannels.Closed_channel;
    Netbuffer.length s_netbuf

  method window_at_eof = 
    if s_closed then raise Netchannels.Closed_channel;
    s_at_eof

  method want_another_block() =
    if s_closed then raise Netchannels.Closed_channel;
    if not s_at_eof then begin
      (* How much are we allowed to read? *)
      let m =
	match s_maxlength with
	    None   -> s_blocksize
	  | Some l -> min (l - s_pos - Netbuffer.length s_netbuf) s_blocksize
      in
      assert(m >= 0);
      (* Try to read m bytes: *)
      let rec read_block k =
	if k < m then
	  let n = 
	    Netbuffer.add_inplace ~len:(m-k) 
	      s_netbuf 
	      (s_channel # input)    
	        (* may raise End_of_file, Buffer_underrun *)
	  in
	  ( if n > 0 then
	      read_block (k+n)
	    else
	      raise Sys_blocked_io
	  )
	else
	  ()
      in
      try
	if m=0 then
	  (* Artificial EOF because len is reached *)
	  s_at_eof <- true
	else
	  read_block 0
      with
	  End_of_file ->
	    s_at_eof <- true
    end;
    (* self # debug "after stream#want_another_block"; *)
    (* Unix.sleep 1; *)


  method want n =
    if s_closed then raise Netchannels.Closed_channel;
    while not s_at_eof && Netbuffer.length s_netbuf < n do
      self # want_another_block()
    done


  method private want_minimum() =
    self # want s_blocksize


  method skip len =
    if s_closed then raise Netchannels.Closed_channel;
    let rec read len =
      if len > 0 then begin
	let k = min (Netbuffer.length s_netbuf) len in
	Netbuffer.delete s_netbuf 0 k;
	s_pos <- s_pos + k;
	self # want_minimum();    (* may raise Buffer_underrun *)
	if k > 0 then read (len - k)
      end
    in
    read len


  method input buf pos len =
    if s_closed then raise Netchannels.Closed_channel;
    if s_underrun then (
      self # want_minimum();  (* may raise Buffer_underrun *)
      s_underrun <- false;
    );
    (* Assertion: Either window length >= minimum, or eof *)
    let len' = min len (Netbuffer.length s_netbuf) in
    Netbuffer.blit s_netbuf 0 buf pos len';
    Netbuffer.delete s_netbuf 0 len';
    s_pos <- s_pos + len';
    ( try 
	self # want_minimum();  (* may raise Buffer_underrun *)
      with
	  Buffer_underrun ->
	    s_underrun <- true
    );
    if len'=0 && len>0 then raise End_of_file;
    len'


  method close_in () =
    if not s_closed then (
      s_channel # close_in();
      s_closed <- true;
    )


end


(*
let find_prefix s1 pos len s2 =
  (* Checks where a non-empty prefix of [s2] occurs at the end of the substring
   * of [s1] beginning at [pos] with length [len]. The function returns 
   * the position [p] of the prefix in [s1]. 
   * The function raises Not_found if it does not find a prefix.
   * POSTCONDITION:
   * - s1[p..p+n-1] = s2[0..n-1] for some biggest n, n <= String.length s2
   *   "The string s1 contains the prefix of s2 at position p, and the
   *   prefix has the maximum length n."
   * - n < String.length s2 ==> p+n = String.length s1
   *   "If the prefix is a proper prefix, it occurs at the end of s1"
   *)
  assert(String.length s2 > 0);
  let l1 = min (String.length s1) (pos+len) in
  let l2 = String.length s2 in
  let s2c0 = s2.[0] in
  let rec check_rec p k =
    k >= l2 || p+k >= l1 || (s1.[p+k] = s2.[k] && check_rec p (k+1)) in
  let rec search_rec p =
    if p >= l1 then raise Not_found;
    let p' = String.index_from s1 p s2c0 in  (* or Not_found *)
    if p' >= l1 then raise Not_found;
    if check_rec p' 0 then
      p'
    else
      search_rec (p'+1)
  in
  search_rec pos
;;
*)


class sub_stream ?len ?delimiter in_stream : in_obj_stream =
object(self)
  val s = (in_stream : in_obj_stream)
  val mutable s_winlen = 0
  val mutable s_del = None    (* initialized below *)
  val s_len = len
  val mutable s_underrun = false

  inherit input_methods (in_stream # window)

  initializer
    (match delimiter with
	 Some "" -> invalid_arg "new Netstream.sub_stream";
       | Some d  -> s_del <- Some(d, Netaux.KMP.make_pattern d)
       | None    -> s_del <- None
    );
    (match s_len with 
	 Some l -> if l<0 then invalid_arg "new Netstream.sub_stream";
       | None   -> ()
    );
    try
      self # want_minimum()
    with
	Buffer_underrun ->
	  s_underrun <- true

  method block_size = s # block_size

  method window = 
    if s_closed then raise Netchannels.Closed_channel;
    s_netbuf

  method window_length = 
    if s_closed then raise Netchannels.Closed_channel;
    s_winlen

  method window_at_eof = 
    if s_closed then raise Netchannels.Closed_channel;
    s_at_eof


  method private compute_winlen() =
    (* sets [s_winlen], [s_at_eof], and returns whether the current window
     * is "ambigous" (it is not clear if the stream does end or does not
     * end)
     *)
    let ambigous = ref false in
    let w = s#window in
    let wlen = s#window_length in
    let weof = s#window_at_eof in
    begin match s_del with
	None -> 
	  s_winlen <- wlen;
	  s_at_eof <- weof;
      | Some(d,pat) ->
	  let p = Netaux.KMP.find_pattern
		    pat ~len:wlen (Netbuffer.unsafe_buffer w) in
	  if p >= wlen then begin
	    (* Delimiter d does not occur in the buffer *)
	    s_winlen <- wlen;
	    s_at_eof <- weof;
	  end
	  else 
	    if (p + String.length d) > wlen then begin
	      (* Case: prefix is a proper prefix *)
	      ambigous := not weof;
	      s_winlen <- wlen;
	      s_at_eof <- weof;
	    end
	    else begin
	      (* Case: [d] occurs in the window *)
	      s_winlen <- p;
	      s_at_eof <- true;
	    end
    end;
    begin match s_len with
	None ->
	  ()
      | Some l ->
	  if l - s_pos < s_winlen then begin
	    ambigous := false;
	    s_winlen <- l - s_pos;
	    s_at_eof <- true;
	  end
    end;
    !ambigous


  method want_another_block() =
    if s_closed then raise Netchannels.Closed_channel;
    s # want_another_block();   (* may raise Buffer_underrun *)
    while self # compute_winlen() do
      s # want_another_block(); (* may raise Buffer_underrun *)
    done

  method want n =
    if s_closed then raise Netchannels.Closed_channel;
    while not s_at_eof && s_winlen < n do
      self # want_another_block()   (* may raise Buffer_underrun *)
    done

  method private want_minimum() =
    if self # compute_winlen() then
      self # want_another_block();   (* may raise Buffer_underrun *)
    self # want s#block_size


  method skip len =
    if s_closed then raise Netchannels.Closed_channel;
    let rec read len =
      if len > 0 then begin
	let k = min s_winlen len in
	s # skip k;              (* may raise Buffer_underrun *)
	s_pos <- s_pos + k;
	self # want_minimum();   (* may raise Buffer_underrun *)
	if k > 0 then read (len - k)
      end
    in
    read len


  method input buf pos len =
    if s_closed then raise Netchannels.Closed_channel;
    if s_underrun then (
      self # want_minimum();  (* may raise Buffer_underrun *)
      s_underrun <- false;
    );
    (* Assertion: Either window length >= minimum, or eof *)
    let len' = min len s_winlen in
    Netbuffer.blit s_netbuf 0 buf pos len';
    s # skip len';              (* never raises Buffer_underrun *)
    s_pos <- s_pos + len';
    ( try
	self # want_minimum();
      with
	  Buffer_underrun ->
	    s_underrun <- true
    );
    if len'=0 && len>0 then raise End_of_file;
    len'


  method close_in () =
    if not s_closed then (
      s # close_in();
      s_closed <- true;
    )

end


let print_in_obj_stream fmt s =
  Format.fprintf fmt
    "<NETSTREAM pos_in:%d window_length:%d eof=%b>"
    s#pos_in
    s#window_length
    s#window_at_eof
;;