How to make lazy seq from socket input?

[timc]

Can someone suggest how to make the packets received on a socket into
a lazy sequence?
The application I'm making is a log file parser. A (Java) program is
producing log output using log4j, the output can go to file(s) or be
sent to a socket. So, the program has this outline:

(defn fileLines [fileNameSeq]
"Return a lazy sequence of lines from the given lazy sequence of file
names.
Return nil at end of file." ...)

(defn socketLines [ipAddr portN]
"Return a lazy sequence of lines from a server socket at the given
place.
Return nil if socket error." ...)

(defn parseLine [line] ... )

(defn parseLines [lineSeq]
"Parse the lines from lineSeq."
(loop [line (first lineSeq) tail (rest lineSeq)]
(when line
(parseLine line)
(recur (first tail) (rest tail)))))

I think I know how to do fileLines, but not socketLines. (Each
received packet should contain one line as it would have been written
to a file, I think). My problem is not how to manage a server socket,
but how to make an ISeq -- i.e. what exactly is the contract that ISeq
defines (the source code has no comments).

Thanks in advance.

[Alex Osborne]

timc wrote:

> I think I know how to do fileLines, but not socketLines. (Each
> received packet should contain one line as it would have been written
> to a file, I think).

Something like this?

(use 'clojure.contrib.duck-streams)

(read-lines (.getInputStream socket))

> My problem is not how to manage a server socket,
> but how to make an ISeq -- i.e. what exactly is the contract that ISeq
> defines (the source code has no comments).

You wouldn't usually implement ISeq directly in Clojure code, you'd use
the lazy-seq function. A simple example from clojure.core:

(defn repeat
"Returns a lazy (infinite!, or length n if supplied) sequence of xs."
([x] (lazy-seq (cons x (repeat x))))
([n x] (take n (repeat x))))

It's lazy because lazy-seq doesn't evaluate it's body until you ask for
it (by calling 'seq', 'first' or 'next' on it). As you access the seq a
linked list is generated. So initially there's just a LazySeq object:

<LazySeq>

Then the body of lazy-seq is evaluated creating a cons cell of the value
x and (repeat x) which evaluates to another LazySeq:

first next
+---+---+
| x | |----> <LazySeq>
+---+---+

And so on it repeats:

first next first next
+---+---+ +---+---+
| x | |----> | x | |----> <LazySeq>
+---+---+ +---+---+

So we could make a simple function that returns a sequence of lines by
calling .readLine over and over on an object:

(defn readline-seq [rdr]
(lazy-seq (cons (.readLine rdr) (readline-seq rdr))))

However, we also want to stop at the end of the file, so we put in a
check for the line being null:

(defn readline-seq [rdr]
(lazy-seq
(when-let [line (.readLine rdr)]
(cons line (readline-seq rdr)))))

This is exactly how core/line-seq is defined, and
duck-streams/read-lines is the same except it also takes care of
adapting whatever you pass in (like a File or InputStream) to a
BufferedReader (which provides the readLine method).

As for ISeq, take a look at Range.java for a simplish example. You just
need to implement the two methods, first() which returns the first thing
in the seq and next() which returns a new seq (ie another instance of
your class) representing the next part of the seq (so seq.next().first()
is the second item, seq.next().next().first() is the third item and so on).

Hope that explanation was clear enough.

[Meikel Brandmeyer]

Hi,

if you have a stream, you can basically do:

(defn stream-seq
[stream]
(take-while #(<= 0 %) (repeatedly #(.read stream))))

This will give a character at a time. From there you can build the
lines and turn it into a seq of lines. Or can you can to the lower
level and use lazy-seq directly.

(defn stream-line-seq
[stream]
(lazy-seq
(loop [line []]
(let [ch (.read stream)]
(cond
(= \newline ch) (cons (apply str line) (stream-line-seq
stream))
(<= 0 ch) (recur (conj line ch))
:else (when (< 0 (count line)) (cons (apply str line)
nil)))))))

Probably not the best implementation, but well... Of course lines is
only an example. This can be used for any type of record.

Beware of laziness in combination with resources like streams!
(premature close etc.)

Hope this helps.

Sincerely
Meikel

[timc]

Thanks for the help.

[Tom Faulhaber]

One thing to keep in mind, when using sockets, is that TCP does not
guarantee to keep packetization across the network. That is, just
because you're writing lines, doesn't mean that read will return
lines. TCP can put writes together or break them into parts or some
combination of both.

So if you need a seq of lines, you need to look at the individual
bytes and find the line breaks. But you probably won't want to read
each byte separately, use a BufferedReader or some such.

HTH,

Tom