Stream Expressions

Streams are very powerful and concise once the necessary tools (stream builders, combinators, and other utilities) are in place. However, the code behind these tools can be a chore to write, at times resembling state machines more than elegant functional expressions. The Camlp4 preprocessor adds syntax support for stream and stream parser expressions, raising the level of abstraction a bit higher and simplifying many common stream-oriented tasks.

Interactive Use

To enable stream expression support in the toplevel, you can type the following:

# #load "dynlink.cma";;
# #load "camlp4o.cma";;

Or, if you are using findlib:

# #use "topfind";;
# #camlp4o;;

Stream Literals

Stream expressions are enclosed by "[<" and ">]" brackets, and using them is a lot like using lists. The simplest stream possible is the empty stream:

[< >]

Literal values in stream expressions are prefixied by single-quotes:

[< '1; '2; '(1 + 2) >]  (* Equivalent to Stream.of_list [1; 2; 3] *)

This is to distinguish them from streams, which are automatically concatenated:

[< '1; '2; more_numbers; '99 >]

In the above example, the stream will produce the integers 1 and 2, followed by all of the values generated by the more_numbers stream, and once more_numbers as been exhausted, it will produce the integer 99.

Recursive Definition

Streams can be defined with recursive functions, providing a straightforward and familiar mechanism to produce infinite sequences. The following defines an never-ending stream of 1s:

let ones =
  let rec aux () =
    [< '1; aux () >] in
  aux ()

Note the auxiliary function, aux, which is called recursively to form a loop. This is necessarily different from infinite lists, which can be defined with "let rec" without a helper function:

# let rec ones = 1 :: ones;;
val ones : int list =
  [1; 1; 1; 1; 1; ...]

The stream expression syntax is not able to interpret recursive values like this, and attempts to do this will result in a syntax error:

# let rec ones = [< '1; ones >];;
Error: This kind of expression is not allowed
as right-hand side of `let rec'

This is only a minor inconvenience, since most streams will be built from one or more parameters. For example, here is the const_stream from the Streams chapter, redefined using stream expression syntax:

let rec const_stream k = [< 'k; const_stream k >]

Similarly, this simple one-liner is all it takes to produce a counter:

let rec count_stream i = [< 'i; count_stream (i + 1) >]

Stream expressions can be built in phases, since one expression can easily include another. The following Fibonacci sequence generator builds an outer stream containing the first two numbers (zero and one), and calls a recursive auxiliary function to produce the rest. Since Fibonacci numbers get large quickly, we'll use the Num module for big integer support:

# let fib =
    let rec aux a b =
      let sum = Num.add_num a b in
      [< 'sum; aux b sum >] in
    let zero, one = Num.Int 0, Num.Int 1 in
    [< 'zero; 'one; aux zero one >];;

Now we can peek at the first 50 numbers in the sequence:

# Stream.npeek 50 fib;;
- : Num.num list =
[<num 0>; <num 1>; <num 1>; <num 2>; <num 3>; <num 5>; <num 8>; <num 13>;
 <num 21>; <num 34>; <num 55>; <num 89>; <num 144>; <num 233>; <num 377>;
 <num 610>; <num 987>; <num 1597>; <num 2584>; <num 4181>; <num 6765>;
 <num 10946>; <num 17711>; <num 28657>; <num 46368>; <num 75025>;
 <num 121393>; <num 196418>; <num 317811>; <num 514229>; <num 832040>;
 <num 1346269>; <num 2178309>; <num 3524578>; <num 5702887>; <num 9227465>;
 <num 14930352>; <num 24157817>; <num 39088169>; <num 63245986>;
 <num 102334155>; <num 165580141>; <num 267914296>; <num 433494437>;
 <num 701408733>; <num 1134903170>; <num 1836311903>; <num 2971215073>;
 <num 4807526976>; <num 7778742049>]

Example: Directory walker

As a more practical example, we can define a recursive directory walker that avoids loading the entire directory tree into memory:

let rec walk dir =
  let items =
    try
      Array.map
        (fun fn ->
           let path = Filename.concat dir fn in
           try
             if Sys.is_directory path
             then `Dir path
             else `File path
           with e -> `Error (path, e))
        (Sys.readdir dir)
    with e -> [| `Error (dir, e) |] in
  Array.fold_right
    (fun item rest ->
       match item with
         | `Dir path -> [< 'item; walk path; rest >]
         | _ -> [< 'item; rest >])
    items
    [< >]

This function works by first assembling an array of paths for the specified base directory. Each path is wrapped in a variant type that keeps track of which path is a file and which is a directory. The array is then folded into a stream, expanding each subdirectory by recursively calling walk again. Errors are included in the variant so that exceptions can be handled or ignored as needed.

The expanding of subdirectories

[< 'item; walk path; rest >]

illustrates a convenient feature of stream expressions: any number of sub-streams can appear in any order, and they will be lazily evaluated as needed. walk path and rest both evaluate to streams that are concatenated with item at the front. The results are flattened into a single stream, just as if we had used something like stream_concat, defined in the Streams chapter.

With little effort, we can now print the names of all the directories and files underneath "/var/log":

let () =
  Stream.iter
    (function
       | `Dir path ->
           Printf.printf "dir: %s%!\n" path
       | `File path ->
           Printf.printf "file: %s%!\n" path
       | `Error (path, e) ->
           Printf.printf "error: %s (%s)%!\n" path
             (Printexc.to_string e))
    (walk "/var/log")

Stream Expressions