C[omp]ute

From: "andrew cooke" <andrew@...>

Date: Wed, 2 May 2007 22:13:06 -0400 (CLT)

I was going to write a long post about this code, but I really don't have
the time.  So it's just here in case anyone feels like reading it.  The
first section defines a protocol and supporting functions for chaining
data through different processes (the "unix pipe" idea).  The protocol is
minimal - just data or exit - but each process can have some persistent
state that is preserved across calls.

I'm particularly proud of the formatting here :o)  Took me some time to
work out what is (and isn't) possible with the syntax, scoping, etc.


forward(none, Message) -> Message;
forward(Next, Message) -> Next(Message).

new(Response, Filter, Name, Next) ->

    DoLog =
        fun(Template, Params) -> res_info(Response, Template, Params) end,


    % when the Filter is called with a {data, State, Data} tuple it
    % should return one of:
    % - {data, Data2} containing processed data
    % - {state, State2} containing new state
    % - {both, State2, Data2} containing new state and data
    % new data (Data2) are forwarded to the next filter in the chain,
    % if defined (it may be "none") even if identical to old values.
    % to avoid forwarding anything to the next filter, return state
    % alone.

    DoData =
        fun(Driver, _State, {both, State2, Data2}) ->
                forward(Next, {data, Data2}),
                Driver(Driver, State2);
           (Driver, State, {data, Data2}) ->
                forward(Next, {data, Data2}),
                Driver(Driver, State);
           (Driver, _State, {state, State2}) ->
                Driver(Driver, State2);
           (_Driver, _State, Other) ->
                DoLog("Unexpected result ~p in ~p", [Other, Name])
        end,


    % when the Filter is called with the {exit, State} tuple it should
    % return one of:
    % - {data, Data} containing final data to be forwarded
    % - exit
    % if data are returned they are passed to the next filter (if
    % defined), after which, "exit" is forwarded.

    DoExit =
        fun({data, Data}) ->
                forward(Next, {data, Data}),
                Next(exit);
           (exit) ->
                Next(exit);
           (Other) ->
                DoLog("Unexpected result ~p in ~p", [Other, Name])
        end,


    % driver loop.

    Driver =
        fun(Driver, State) ->
                receive
                    {data, Data} ->
                        DoData(Driver, State, Filter({data, State, Data}));
                    exit ->
                        DoExit(Filter({exit, State}));
                    Other ->
                        DoLog("Unexpected message ~p in ~p", [Other, Name])
                end
        end,

    Pid = spawn(fun() -> Driver(Driver, Filter(start)) end),
    fun(Message) -> Pid ! Message end.


Next, some example processes.  First, a simple one that groups the data by
lines (ended by newline character).  Note the exit strategy handles a
missing final newline.

lines(start) -> [];
lines({data, Partial, Data}) ->
    {Partial2, Lines} = collect(Data, Partial, []),
    {both, Partial2, Lines};
lines({exit, Partial}) -> {data, [lists:reverse(Partial)]}.

collect([], Partial, Lines) -> {Partial, lists:reverse(Lines)};
collect([$\n|Data], Partial, Lines) ->
    Line = lists:reverse([$\n|Partial]),
    collect(Data, [], [Line|Lines]);
collect([C|Data], Partial, Lines) ->
    collect(Data, [C|Partial], Lines).


And second, a process that "gunzips" a stream.  As far as I can tell, the
Erlang zlib library doesn't (directly) support streaming gunzip, so this
accumulates the data beforehand.

unzip(start) -> [];
unzip({data, Acc, Data}) -> {state, [Data|Acc]};
unzip({exit, Acc}) ->
    Compressed = list_to_binary(lists:reverse(Acc)),
    Decompressed = binary_to_list(zlib:gunzip(Compressed)),
    {data, Decompressed}.


I was worried that this buffering would make subsequent chained processes
inefficient (processors waiting on data), so here's something that
fragments the input to a process (and shows that wrapping these processes
as functions pays off).

shape_before(Fun, ChunkLimit) ->
    fun({data, Data}) -> fragment(Fun, ChunkLimit, Data);
       (Message) -> Fun(Message)
    end.

fragment(_Fun, _Chunklimit, []) -> ok;
fragment(Fun, Chunklimit, Data) when length(Data) > Chunklimit ->
    {Chunk, Data2} = lists:split(Chunklimit, Data),
    Fun({data, Chunk}),
    fragment(Fun, Chunklimit, Data2);
fragment(Fun, _Chunklimit, Data) ->
    Fun({data, Data}).


Finally, this is construction of a chain of processes, with fragmentation
and optional compression:

close(Response) ->
    res_info(Response, "Closing"),
    Channel = response:channel(Response),
    socket:close(channel:browser(Channel), channel:proxy(Channel)).

build_chain(Response) ->
    Echo = chain:link(Response, chain:echo(Response), "Echo"),
    Links = chain:new(Response, fun chain_links:links/1, "Links", Echo),
    Lines = chain:new(Response, fun chain_lines:lines/1, "Lines", Links),
    res_info(Response, "Encoding ~p", [response:encoding(Response)]),
    case response:encoding(Response) of
        unknown -> Lines;
        "gzip" -> chain:new(Response, fun chain_unzip:unzip/1, "Gzip",
                            chain:shape_before(Lines, 2000));
        Encoding -> {error, Encoding}
    end.

The processes are in reverse order, so this (optionally) unzips, splits
into lines, searches for hypertext limits, and echoes the result to the
screen.  As you might have guessed, I'm processing HTTP data flows.

Andrew