-module(line_read_test).

-include_lib("kernel/include/file.hrl").

-export([  read/1
         , read/2
         , read_std/1
         , read_port/1
         , read_line/1
         , close/1]).

read_std(Fname) -> 
  {ok, FH} = file:open(Fname, [read, raw, binary, {read_ahead, 1 bsl 20}]),
  read(FH).

read(Fname, BuffSize) ->
  {ok, FH} = open(Fname, BuffSize),
  read(FH).

read(Fname) when is_list(Fname) ->
  {ok, FH} = open(Fname, 65536),
  read(FH);
read(FH) when is_pid(FH); is_record(FH, file_descriptor) ->
  try
    read_lines(FH, 0)
  after
    garbage_collect(),
    ok = file:close(FH)
  end.

read_lines(FH, Lines) ->
  case file:read_line(FH) of
    eof -> Lines;
    {ok, _Line} ->
      read_lines(FH, Lines + 1)
  end.

read_port(Fname) ->
  OldFlag = process_flag(trap_exit, true),
  Port = open_port({spawn_executable, "/bin/cat"},
                   [{line, 65536}, binary, {args, [Fname]}]),
  try
    read_port(Port, 0)
  after
    garbage_collect(),
    process_flag(trap_exit, OldFlag),
    (catch port_close(Port))
  end.

read_port(Port, Lines) ->
  receive
    {Port, {data, {eol, _Line}}} -> read_port(Port, Lines+1);
    {Port, {data, _}} -> read_port(Port, Lines);
    {'EXIT', Port, _} -> Lines
  end.

% file_descriptor module API

open(Fname, BuffSize) when is_list(Fname) ->
  Self = self(),    
  Pid = spawn_link(fun() -> line_reader(Self, Fname, BuffSize) end),
  receive
    {Pid, opened} -> {ok, #file_descriptor{module=?MODULE, data=Pid}};
    {Pid, Error} -> {error, Error}
  end.

read_line(#file_descriptor{data=Pid}) ->
  Mref = monitor(process, Pid),
  Pid ! {self(), Mref, read_line},
  receive
    {ok, Mref, eof} -> demonitor(Mref, [flush]), eof;
    {ok, Mref, Data} -> demonitor(Mref, [flush]), {ok, Data};
    {error, Mref, Error} -> demonitor(Mref, [flush]), {error, Error};
    {'DOWN', Mref, _, _, _} -> {error, closed}
  end.

close(#file_descriptor{data=Pid}) -> Pid ! close, ok.

% file server

line_reader(Parent, Fname, BuffSize) ->
  case file:open(Fname, [read, raw, binary, {read_ahead, 65536}]) of
    {ok, FH} ->
      Parent ! {self(), opened},
      loop(Parent, {FH, <<>>, BuffSize});
    {error, Error} ->
      Parent ! {self(), Error}
  end.

loop(Parent, eof) ->
  receive
    close -> ok;
    {'EXIT', Parent, _} -> ok;
    {Pid, Ref, read_line} ->
          Pid ! {ok, Ref, eof},
      loop(Parent, eof)
  end;
loop(Parent, {FH, _, _} = H) ->
  receive
    close -> file:close(FH);
    {'EXIT', Parent, _} -> file:close(FH);
    {Pid, Ref, read_line} ->
      case read_line_int(H) of
        eof ->
          Pid ! {ok, Ref, eof},
          file:close(FH),
          loop(Parent, eof);
        {ok, Data, NewH} ->
          Pid ! {ok, Ref, Data},
          loop(Parent, NewH);
        {error, Error} ->
          Pid ! {error, Ref, Error},
          file:close(FH)
      end
  end.

read_line_int(eof) -> eof;
read_line_int({FH, Buff, BuffSize}) ->
  case split(Buff) of
    [Line, Rest] -> {ok, Line, {FH, Rest, BuffSize}};
    [Prefix] -> case read_line_int(FH, BuffSize, Prefix) of
        {error, _} = E -> E;
        {ok, <<>>, eof} -> eof;
        {ok, Line, X} -> {ok, <<Prefix/binary, Line/binary>>, X}
      end
  end.

read_line_int(FH, BuffSize, Acc) ->
  case file:read(FH, BuffSize) of
    eof -> {ok, Acc, eof};
    {error, _} = E -> E;
    {ok, Data} ->
      case split(Data) of
        [Line, Rest] ->
          {ok, <<Acc/binary, Line/binary>>, {FH, Rest, BuffSize}};
        [Prefix] ->
          read_line_int(FH, BuffSize, <<Acc/binary, Prefix/binary>>)
      end
  end.

split(Bin) ->
  case binary:match(Bin, <<$\n>>) of
    nomatch ->
      [Bin];
    {Pos, _} ->
      Split = Pos+1,
      <<Prefix:Split/binary, Rest/binary>> = Bin,
      [Prefix, Rest]
  end.