← Back to branch summary

~ubuntu-branches/ubuntu/oneiric/ocsigen/oneiric

« back to all changes in this revision

Viewing changes to http/ocsigen_stream.ml

  • Committer: Bazaar Package Importer
  • Author(s): Stephane Glondu
  • Date: 2009-07-02 10:02:08 UTC
  • mfrom: (1.1.9 upstream) (4.1.3 squeeze)
  • Revision ID: james.westby@ubuntu.com-20090702100208-n158b1sqwzn0asil
Tags: 1.2.0-2
Fix build on non-native architectures

Show diffs side-by-side

added added

removed removed

Lines of Context:
http/ocsigen_stream.ml
1
 
(* Ocsigen
2
 
 * ocsigen_stream.ml Copyright (C) 2005 Vincent Balat
3
 
 * Laboratoire PPS - CNRS Universit� Paris Diderot
4
 
 *
5
 
 * This program is free software; you can redistribute it and/or modify
6
 
 * it under the terms of the GNU Lesser General Public License as published by
7
 
 * the Free Software Foundation, with linking exception;
8
 
 * either version 2.1 of the License, or (at your option) any later version.
9
 
 *
10
 
 * This program is distributed in the hope that it will be useful,
11
 
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12
 
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
 
 * GNU Lesser General Public License for more details.
14
 
 *
15
 
 * You should have received a copy of the GNU Lesser General Public License
16
 
 * along with this program; if not, write to the Free Software
17
 
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18
 
 *)
19
 
 
20
 
let (>>=) = Lwt.bind
21
 
 
22
 
exception Interrupted of exn
23
 
exception Cancelled
24
 
exception Already_read
25
 
exception Finalized
26
 
 
27
 
type 'a stream = 'a step Lwt.t Lazy.t
28
 
 
29
 
and 'a step =
30
 
  | Finished of 'a stream option (* If there is another stream following
31
 
                                    (usefull for substreams) *)
32
 
  | Cont of 'a * 'a stream       (* Current buffer, what follows *)
33
 
 
34
 
type 'a t =
35
 
  { mutable stream : 'a stream;
36
 
    mutable in_use : bool;
37
 
    mutable finalizer : unit -> unit Lwt.t }
38
 
 
39
 
let empty follow =
40
 
  match follow with
41
 
    None    -> Lwt.return (Finished None)
42
 
  | Some st -> Lwt.return (Finished (Some (Lazy.lazy_from_fun st)))
43
 
 
44
 
let cont stri f =
45
 
  Lwt.return (Cont (stri, Lazy.lazy_from_fun f))
46
 
 
47
 
let make ?finalize:(g = fun () -> Lwt.return ()) f =
48
 
  { stream = Lazy.lazy_from_fun f; in_use = false; finalizer = g }
49
 
 
50
 
let next = Lazy.force
51
 
 
52
 
let rec get_aux st =
53
 
  lazy
54
 
    (Lwt.try_bind
55
 
       (fun () -> Lazy.force st.stream)
56
 
       (fun e ->
57
 
          Lwt.return
58
 
            (match e with
59
 
               Cont (s, rem) -> st.stream <- rem; Cont (s, get_aux st)
60
 
             | _             -> e))
61
 
       (fun e ->
62
 
          st.stream <- lazy (Lwt.fail e);
63
 
          Lwt.fail (Interrupted e)))
64
 
 
65
 
let get st =
66
 
  if st.in_use then raise Already_read;
67
 
  st.in_use <- true;
68
 
  get_aux st
69
 
 
70
 
(** read the stream until the end, without decoding *)
71
 
let rec consume_aux st =
72
 
  next st >>= fun e ->
73
 
  match e with
74
 
  | Cont (_, f)        -> consume_aux f
75
 
  | Finished (Some ss) -> consume_aux ss
76
 
  | Finished None      -> Lwt.return ()
77
 
 
78
 
let cancel st =
79
 
  let st' = st.stream in
80
 
  st.stream <- lazy (Lwt.fail Cancelled);
81
 
  consume_aux st'
82
 
 
83
 
let consume st =
84
 
  consume_aux st.stream
85
 
 
86
 
let finalize st =
87
 
  let f = st.finalizer in
88
 
  st.finalizer <- (fun () -> Lwt.return ());
89
 
  f () >>= fun () ->
90
 
  st.stream <- lazy (Lwt.fail Finalized);
91
 
  Lwt.return ()
92
 
 
93
 
let add_finalizer st g =
94
 
  let f = st.finalizer in
95
 
  st.finalizer <- fun () -> f () >>= fun () -> g ()
96
 
 
97
 
(****)
98
 
 
99
 
(** String streams *)
100
 
 
101
 
open Lwt
102
 
 
103
 
exception Stream_too_small
104
 
exception Stream_error of string
105
 
exception String_too_large
106
 
 
107
 
(*XXX Quadratic!!! *)
108
 
let string_of_stream =
109
 
  let rec aux l s =
110
 
    next s >>= fun e ->
111
 
    match e with
112
 
    | Finished _ -> return ""
113
 
    | Cont (s, f) ->
114
 
        let l2 = l+String.length s in
115
 
        if l2 > Ocsigen_config.get_netbuffersize ()
116
 
        then fail String_too_large
117
 
        else
118
 
          aux l2 f >>=
119
 
             (fun r -> return (s^r))
120
 
  in aux 0
121
 
 
122
 
(*XXX Quadratic!!! *)
123
 
let string_of_streams =
124
 
  let rec aux l = function
125
 
    | Finished None -> return ""
126
 
    | Finished (Some s) -> next s >>= fun r -> aux l r
127
 
    | Cont (s, f) ->
128
 
        let l2 = l+String.length s in
129
 
        if l2 > Ocsigen_config.get_netbuffersize ()
130
 
        then fail String_too_large
131
 
        else
132
 
          next f >>= fun r ->
133
 
          aux l2 r >>= fun r ->
134
 
          return (s^r)
135
 
  in aux 0
136
 
 
137
 
let enlarge_stream = function
138
 
  | Finished a -> fail Stream_too_small
139
 
  | Cont (s, f) ->
140
 
      let long = String.length s in
141
 
      let max = Ocsigen_config.get_netbuffersize () in
142
 
      if long >= max
143
 
      then fail Ocsigen_lib.Input_is_too_large
144
 
      else
145
 
        next f >>= fun e ->
146
 
        match e with
147
 
        | Finished _ -> fail Stream_too_small
148
 
        | Cont (r, ff) ->
149
 
            let long2 = String.length r in
150
 
            let long3=long+long2 in
151
 
            let new_s = s^r in
152
 
            if long3 <= max
153
 
            then return (Cont (new_s, ff))
154
 
            else let long4 = long3 - max in
155
 
            cont (String.sub new_s 0 max)
156
 
                 (fun () ->
157
 
                    Lwt.return (Cont (String.sub new_s max long4, ff)))
158
 
 
159
 
let rec stream_want s len =
160
 
 (* returns a stream with at most len bytes read if possible *)
161
 
  match s with
162
 
  | Finished _ -> return s
163
 
  | Cont (stri, f)  -> if String.length stri >= len
164
 
  then return s
165
 
  else catch
166
 
        (fun () -> enlarge_stream s >>= (fun r -> stream_want s len))
167
 
        (function
168
 
            Stream_too_small -> return s
169
 
          | e -> fail e)
170
 
 
171
 
let current_buffer = function
172
 
  | Finished _  -> raise Stream_too_small
173
 
  | Cont (s, _) -> s
174
 
 
175
 
let rec skip s k = match s with
176
 
| Finished _ -> raise Stream_too_small
177
 
| Cont (s, f) ->
178
 
    let len = String.length s in
179
 
    if k <= len
180
 
    then return (Cont (String.sub s k (len - k), f))
181
 
    else (enlarge_stream (Cont ("", f)) >>=
182
 
          (fun s -> skip s (k - len)))
183
 
 
184
 
let substream delim s =
185
 
  let ldelim = String.length delim in
186
 
  if ldelim = 0 then fail (Stream_error "Empty delimiter")
187
 
  else
188
 
    let rdelim = Netstring_pcre.regexp_string delim in
189
 
    let rec aux =
190
 
      function
191
 
        | Finished _ -> fail Stream_too_small
192
 
        | Cont (s, f) as stre ->
193
 
            let len = String.length s in
194
 
            if len < ldelim
195
 
            then enlarge_stream stre >>= aux
196
 
            else try
197
 
              let p,_ = Netstring_pcre.search_forward rdelim s 0 in
198
 
              cont (String.sub s 0 p)
199
 
                   (fun () ->
200
 
                      empty
201
 
                           (Some (fun () -> Lwt.return (Cont (String.sub s p (len - p),
202
 
                                               f)))))
203
 
            with Not_found ->
204
 
              let pos = (len + 1 - ldelim) in
205
 
              cont (String.sub s 0 pos)
206
 
                       (fun () -> next f >>=
207
 
                         (function
208
 
                             Finished _ -> fail Stream_too_small
209
 
                           | Cont (s', f') ->
210
 
                               aux
211
 
                                 (Cont (String.sub s pos (len - pos) ^ s',
212
 
                                        f'))
213
 
                         ))
214
 
    in aux s
215
 
 
216
 
(*****************************************************************************)
217
 
 
218
 
(*VVV Is it the good place for this? *)
219
 
 
220
 
let of_file filename =
221
 
  let fd = Lwt_unix.of_unix_file_descr
222
 
      (Unix.openfile filename [Unix.O_RDONLY;Unix.O_NONBLOCK] 0o666)
223
 
  in
224
 
  let ch = Lwt_chan.in_channel_of_descr fd in
225
 
  let rec aux () =
226
 
    catch
227
 
      (fun () ->
228
 
        Lwt_chan.input_line ch >>= fun s ->
229
 
        (cont s aux))
230
 
      (function End_of_file -> empty None | e -> fail e)
231
 
  in make ~finalize:(fun () -> Lwt.return (Lwt_unix.close fd)) aux
232
 
 
233
 
let of_string s =
234
 
  make (fun () -> cont s (fun () -> empty None))