stream.erl

%%
%%   Copyright 2012 - 2013 Dmitry Kolesnikov, All Rights Reserved
%%
%%   Licensed under the Apache License, Version 2.0 (the "License");
%%   you may not use this file except in compliance with the License.
%%   You may obtain a copy of the License at
%%
%%       http://www.apache.org/licenses/LICENSE-2.0
%%
%%   Unless required by applicable law or agreed to in writing, software
%%   distributed under the License is distributed on an "AS IS" BASIS,
%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%   See the License for the specific language governing permissions and
%%   limitations under the License.
%%
%% @doc
%%   streams or lazy lists are a sequential data structure that contains 
%%   on demand computed elements. Erlang implementation derived from
%%   scheme stream interface @see http://srfi.schemers.org/srfi-41/srfi-41.html
-module(stream).
-behaviour(traversable).
-behaviour(foldable).

-include("datum.hrl").

%%
%% stream primitives - abstract data type 
-export([
   new/0,
   new/1,
   new/2,

   %%
   %% traversable
   head/1,
   tail/1,
   build/1,
   list/1, 
   list/2,
   is_empty/1,
   drop/2,
   dropwhile/2,
   filter/2,
   foreach/2,
   map/2,
   split/2,
   splitwhile/2,
   take/2,
   takewhile/2,

   %%
   %% foldable
   fold/3,
   foldl/3,
   foldr/3,
   unfold/2,

   %%
   %% stream
   '++'/2,
   '++'/1,
   flat/1,
   scan/2,
   scan/3,
   zip/1,
   zip/2,
   zipwith/2,
   zipwith/3,
   reverse/1,
   cycle/1
]).


%%
%% inline stream primitives
-compile({inline,[new/0, new/1, new/2, head/1, tail/1]}).

%%%------------------------------------------------------------------
%%%
%%% stream primitives
%%%
%%%------------------------------------------------------------------

%%
%% creates a newly allocated stream containing stream head and promise.
%% the promise is recursive, it returns new stream pair when evaluated.
-spec new() -> datum:stream(_).
-spec new(_) -> datum:stream(_).
-spec new(_, function()) -> datum:stream(_).

new() ->
   ?None.

new(Head) ->
   new(Head, fun stream:new/0).

new(Head, #stream{} = Stream) ->
   #stream{head = Head, tail = fun() -> Stream end};
new(Head, Fun)
 when is_function(Fun, 0) ->
   #stream{head = Head, tail = Fun};
new(Head, {_M, _F, _A} = Gen) ->
   #stream{head = Head, tail = Gen};
new(Head, ?None) ->
   new(Head).


%%%------------------------------------------------------------------
%%%
%%% traversable
%%%
%%%------------------------------------------------------------------

%%
%% take collection and return head element of collection
%%
-spec head(datum:traversable(_)) -> datum:option(_).

head(?None) ->
   ?None;
head(#stream{head = Head}) ->
   Head.

%%
%% force stream promise and return new stream (evaluates tail of stream).
-spec tail(datum:traversable(_)) -> datum:traversable(_).

tail(?None) ->
   ?None;
tail(#stream{tail = {M, F, A}}) ->
   M:F(A);
tail(#stream{tail = Fun}) ->
   Fun().

%%
%% build a new collection from Erlang list
%%
-spec build([_] | integer()) -> datum:traversable(_).

build([]) ->
   new();
build([Head|Tail]) ->
   new(Head, fun() -> build(Tail) end);
build(X)
 when is_integer(X) ->
   new(X, fun() -> build(X + 1) end).


%%
%% returns a newly-allocated list containing stream elements
-spec list(datum:stream()) -> list().

list(?None) ->
   [];
list(#stream{} = Stream) ->
   [stream:head(Stream) | list(stream:tail(Stream))].

%%
%% returns a newly-allocated list containing stream elements
-spec list(integer(), datum:stream()) -> list().

list(N, Stream) ->
   list(stream:take(N, Stream)).


%%
%% return true if collection is empty 
%%
-spec is_empty(datum:traversable(_)) -> true | false.

is_empty(?None) ->
   true;
is_empty(#stream{}) ->
   false.


%%
%% returns the suffix of the input stream that starts at the next element after
%% the first n elements.
-spec drop(integer(), datum:stream(_)) -> datum:stream(_).

drop(0, #stream{} = Stream) ->
   Stream;
drop(N, #stream{} = Stream) ->
   drop(N - 1, tail(Stream));
drop(_, ?None) ->
   ?None.

%%
%% drops elements from stream while predicate returns true and returns remaining
%% stream suffix.
-spec dropwhile(datum:predicate(_), datum:stream(_)) -> datum:stream(_).      

dropwhile(Pred, #stream{} = Stream) ->
   case Pred(head(Stream)) of
      true  -> 
         dropwhile(Pred, tail(Stream)); 
      false -> 
         Stream
   end;
dropwhile(_, ?None) ->
   ?None.

%%
%% returns a newly-allocated stream that contains only those elements x of the 
%% input stream for which predicate is true.
-spec filter(datum:predicate(_), datum:stream(_)) -> datum:stream(_).

filter(Pred, #stream{head = Head} = Stream) ->
   case Pred(Head) of
      true -> 
         new(Head, fun() -> filter(Pred, tail(Stream)) end);
      false ->
         filter(Pred, tail(Stream))
   end;
filter(_, ?None) ->
   ?None.

%%
%% applies a function to each stream element for its side-effects; 
%% it returns nothing. 
-spec foreach(function(), datum:stream()) -> ok.

foreach(Fun, #stream{} = Stream) ->
   _ = Fun(head(Stream)),
   foreach(Fun, tail(Stream));
foreach(_, ?None) ->
   ok.


%%
%% create a new stream by apply a function to each element of input stream. 
-spec map(fun((_) -> _), datum:stream(_)) -> datum:stream(_).

map(Fun, #stream{} = Stream) ->
   new(Fun(head(Stream)), fun() -> map(Fun, tail(Stream)) end);
map(_, ?None) ->
   ?None.



%%
%% partitions stream into two streams. The split behaves as if it is defined as 
%% consequent take(N, Stream), drop(N, Stream). 
-spec split(integer(), datum:traversable(_)) -> {datum:traversable(_), datum:traversable(_)}.

split(N, Stream) ->
   split(N, [], Stream).

split(0, Acc, Stream) ->
   {stream:build(lists:reverse(Acc)), Stream};
   
split(N, Acc, #stream{} = Stream) ->
   split(N - 1, [head(Stream)|Acc], tail(Stream));

split(_, Acc, ?None) ->
   {stream:build(lists:reverse(Acc)), ?None}.

%%
%% partitions stream into two streams according to predicate.
%% The splitwith/2 behaves as if it is defined as consequent 
%% takewhile(Pred, Stream), dropwhile(Pred, Stream)
-spec splitwhile(datum:predicate(_), datum:traversable(_)) -> {datum:traversable(_), datum:traversable(_)}.

splitwhile(Pred, Stream) ->
   splitwhile(Pred, [], Stream).

splitwhile(Pred, Acc, #stream{} = Stream) ->
   case Pred(head(Stream)) of
      true  ->
         splitwhile(Pred, [head(Stream)|Acc], tail(Stream));
      false ->
         {stream:build(lists:reverse(Acc)), Stream}
     end;

splitwhile(_, Acc, ?None) ->
   {stream:build(lists:reverse(Acc)), ?None}.


%%
%% returns a newly-allocated stream containing the first n elements of 
%% the input stream. 
-spec take(integer(), datum:traversable(_)) -> datum:traversable(_).

take(0, _) ->
   new();
take(N, #stream{} = Stream) ->
   new(head(Stream), fun() -> take(N - 1, tail(Stream)) end);
take(_, ?None) ->
   ?None.

%%
%% returns a newly-allocated stream that contains those elements from stream 
%% while predicate returns true.
-spec takewhile(datum:predicate(_), datum:traversable(_)) -> datum:traversable(_).

takewhile(Pred, #stream{} = Stream) ->
   case Pred(head(Stream)) of
      true  -> 
         new(head(Stream), fun() -> takewhile(Pred, tail(Stream)) end);
      false ->
         new()
     end;
takewhile(_, ?None) ->
   ?None.

%%%------------------------------------------------------------------
%%%
%%% foldable
%%%
%%%------------------------------------------------------------------

%%
%% applies a function to stream head and accumulator to compute a new accumulator,
%% then applies the function to the new accumulator and the next element of stream to 
%% compute a succeeding accumulator, and so on, the final accumulated value is returned
%% when the end of the stream is reached. Stream must be finite.
-spec fold(datum:monoid(_), _, datum:foldable(_)) -> _.

fold(Fun, Acc, Stream) ->
   foldl(Fun, Acc, Stream).

%%
%% Left-associative fold of a structure
%%
-spec foldl(datum:monoid(_), _, datum:foldable(_)) -> _.

foldl(Fun, Acc, #stream{} = Stream) ->
   foldl(Fun, Fun(head(Stream), Acc), tail(Stream));
foldl(_, Acc, ?None) ->
   Acc.


%%
%% Right-associative fold of a structure
%%
-spec foldr(datum:monoid(_), _, datum:foldable(_)) -> _.

foldr(Fun, Acc, #stream{} = Stream) ->
   lists:foldr(Fun, Acc, stream:list(Stream)).

%% 
%% The fundamental recursive structure constructor, 
%% it applies a function to each previous seed element in turn
%% to determine the next element.
%%
-spec unfold(fun((_) -> _), _) -> datum:foldable(_).

unfold(Fun, Seed)
 when is_function(Fun, 1) ->
   case Fun(Seed) of
      {Head, Next} ->
         new(Head, fun() -> unfold(Fun, Next) end);
      _ ->
         new()
   end.


%%%------------------------------------------------------------------
%%%
%%% stream 
%%%
%%%------------------------------------------------------------------

%%
%% concatenate streams, returns newly-allocated stream composed of elements
%% copied from input streams (in order of input). 
-spec '++'(datum:stream(_), datum:stream(_)) -> datum:stream(_).

'++'(?None, StreamB) ->
   StreamB;
'++'(StreamA, ?None) ->
   StreamA;
'++'(StreamA, StreamB) ->
   new(head(StreamA), fun() -> '++'(tail(StreamA), StreamB) end).

%%
%% concatenate streams, returns newly-allocated stream composed of elements
%% copied from input streams (in order of input). 
-spec '++'([datum:stream(_)]) -> datum:stream(_).

'++'([A,B|T]) ->
   '++'(['++'(A,B)|T]);
'++'([T]) ->
   T.

%%
%% flat stream of streams
-spec flat(datum:stream(_)) -> datum:stream(_).

flat(#stream{head = ?None} = Stream) ->
   flat(tail(Stream));

flat(#stream{head = #stream{} = Head} = Stream) ->
   new(head(Head), fun() -> flat(Stream#stream{head = tail(Head)}) end);

flat(Stream) ->
   Stream.


%%
%% accumulates the partial folds of an input stream into a newly-allocated stream.
%% the output stream is accumulator followed by partial fold.
-spec scan(function(), datum:stream(_)) -> datum:stream(_).
-spec scan(function(), _, datum:stream(_)) -> datum:stream(_).

scan(Fun, #stream{} = Stream) ->
   scan(Fun, head(Stream), tail(Stream)).

scan(Fun, Acc0, #stream{} = Stream) ->
   new(Acc0, fun() -> scan(Fun, Fun(head(Stream), Acc0), tail(Stream)) end);
scan(_, Acc0, ?None) ->
   new(Acc0).


%%
%% takes one or more input streams and returns a newly-allocated stream 
%% in which each element is a list of the corresponding elements of the input 
%% streams. The output stream is as long as the shortest input stream.
-spec zip([datum:stream(_)]) -> datum:stream(_).
-spec zip(datum:stream(_), datum:stream(_)) -> datum:stream(_).

zip(Streams) ->
   case [head(X) || X <- Streams, X =/= ?None] of
      Head when length(Head) =:= length(Streams) ->
         new(Head, fun() -> zip([tail(X) || X <- Streams]) end);
      _ ->
         new()
   end.

zip(A, B) ->
   zip([A, B]).

%%
%% takes one or more input streams and returns a newly-allocated stream,  
%% each element produced by composition function that map list of input heads to
%% new head. The output stream is as long as the longest input stream.
-spec zipwith(function(), [datum:stream(_)]) -> datum:stream(_).
-spec zipwith(function(), datum:stream(_), datum:stream(_)) -> datum:stream(_).

zipwith(_, []) ->
   new();
zipwith(Fun, Streams) ->
   zipwith1(Fun, [head(X) || X <- Streams, X =/= ?None], Streams).

zipwith1(_Fun, [], _Streams) ->
   new();
zipwith1(Fun, Head, Streams) ->
   case Fun(Head) of
      [] ->
         new();
      Hx ->
         new(Hx, fun() -> zipwith(Fun, [tail(X) || X <- Streams, X =/= ?None]) end)
   end.
   
zipwith(Fun, A, B) ->
   zipwith(Fun, [A, B]).


%%%------------------------------------------------------------------
%%%
%%% stream utility
%%%
%%%------------------------------------------------------------------

%%
%% reverse order of elements in stream.
-spec reverse(datum:stream(_)) -> datum:stream(_).

reverse(Stream) ->
   reverse(Stream, new()).

reverse(?None, Acc) ->
   Acc;
reverse(#stream{} = Stream, Acc) ->
   reverse(tail(Stream), new(head(Stream), Acc)).

%%
%% takes list of elements and returns a newly-allocated stream composed of 
%% list elements, repeating them in succession forever.
-spec cycle(list()) -> datum:stream().

cycle(List) -> 
   cycle([], List).

cycle([Head|Tail], List) ->
   new(Head, fun() -> cycle(Tail, List) end);
cycle([], [Head|Tail]=List) ->
   new(Head, fun() -> cycle(Tail, List) end).





%%%------------------------------------------------------------------
%%%
%%% private
%%%
%%%------------------------------------------------------------------