Skip to content

Commit

Permalink
Update example for streaming lines
Browse files Browse the repository at this point in the history
Update the example for using `download_file` to create a
line-orientented stream, conforming to the code used in nimble_csv which
accounts for a multi-line final chunk.
  • Loading branch information
fastjames authored and bernardd committed Jun 30, 2022
1 parent 962945b commit dd8e106
Showing 1 changed file with 24 additions and 10 deletions.
34 changes: 24 additions & 10 deletions lib/ex_aws/s3.ex
Original file line number Diff line number Diff line change
Expand Up @@ -684,24 +684,38 @@ defmodule ExAws.S3 do
# Uncomment if you need to gunzip (and add dependency :stream_gzip)
# |> StreamGzip.gunzip()
|> Stream.chunk_while("", &chunk_fun/2, &after_fun/1)
|> Stream.concat()
end
def chunk_fun(chunk, acc) do
split_chunk(acc, chunk) || split_chunk(chunk, "")
to_try = acc <> chunk
{elements, acc} = chunk_by_newline(to_try, "\n", [], {0, byte_size(to_try)})
{:cont, elements, acc}
end
defp split_chunk("", _append_remaining), do: nil
defp split_chunk(string, append_remaining) do
case String.split(string, "\\n", parts: 2) do
[l] ->
{:cont, l}
[l, remaining] ->
{:cont, l, remaining <> append_remaining}
defp chunk_by_newline(_string, _newline, elements, {_offset, 0}) do
{Enum.reverse(elements), ""}
end
defp chunk_by_newline(string, newline, elements, {offset, length}) do
case :binary.match(string, newline, scope: {offset, length}) do
{newline_offset, newline_length} ->
difference = newline_length + newline_offset - offset
element = binary_part(string, offset, difference)
chunk_by_newline(
string,
newline,
[element | elements],
{newline_offset + newline_length, length - difference}
)
:nomatch ->
{Enum.reverse(elements, binary_part(string, offset, length))}
end
end
def after_fun(""), do: {:cont, ""}
def after_fun(acc), do: {:cont, acc, ""}
defp to_line_stream_after_fun(""), do: {:cont, []}
defp to_line_stream_after_fun(acc), do: {:cont, [acc], []}
"""
@spec download_file(bucket :: binary, path :: binary, dest :: :memory | binary) ::
__MODULE__.Download.t()
Expand Down

0 comments on commit dd8e106

Please sign in to comment.