       iex> Enum.take(2)
       [%{:x => \"a\", :y => \"b\"}, %{:x => \"c\", :y => \"d\"}]
   """
+  
+  def decode(stream, options \\ []) do
+    with options <- options_with_defaults(options),
+         { :ok, { headers, stream } } <-
+           get_headers(options |> Keyword.get(:headers), stream, options),
+         { :ok, row_length } <-
+           get_row_length(headers || stream, options),
+         do: { :ok,
+               process_stream({ stream, row_length }, options |> Keyword.merge(headers: headers))}
+  end
 
-  def decode(stream, options \\ []) do
-    { headers, stream } = options
-                          |> Keyword.get(:headers, false)
-                          |> get_headers(stream, options)
+  defp process_stream({ stream, row_length }, options) do
+    stream
+    |> aggregate(options |> Keyword.get(:multiline_escape))
+    |> Stream.with_index
+    |> ParallelStream.map(&(process_line({ &1, row_length }, options)),
+                          options)
+  end
+  
+  defp options_with_defaults(options) do
+    num_pipes = options |> Keyword.get(:num_pipes, Defaults.num_workers)
+    options
+    |> Keyword.merge(num_pipes: num_pipes,
+                     num_workers: options |> Keyword.get(:num_workers, num_pipes),
+                     multiline_escape: options |> Keyword.get(:multiline_escape, true),
+                     headers: options |> Keyword.get(:headers, false))
+  end
 
-    row_length = case headers do
-      false -> stream |> get_row_length(options)
-      _ -> headers |> Enum.count
-    end
+  defp process_line({ { line, index }, row_length }, options) do
+    with { :ok, lex, _ } <- Lexer.lex({ line, index }, options),
+         { :ok, parsed, _ } <- Parser.parse({ lex, index }, options),
+         { :ok } <- check_row_length({ parsed, index }, row_length),
+         do: build_row(parsed, options |> Keyword.get(:headers))
+  end
 
-    num_workers = options
-                  |> Keyword.get(:num_workers, options
-                  |> Keyword.get(:num_pipes, Defaults.num_workers))
-
-    multiline_escape = options
-                        |> Keyword.get(:multiline_escape, true)
-
-
-    stream
-    |> aggregate(multiline_escape)
-    |> Stream.with_index
-    |> ParallelStream.map(fn { line, index } ->
-      { line, index }
-      |> Lexer.lex(options)
-      |> lex_to_parse
-      |> Parser.parse(options)
-      |> check_row_length(row_length)
-      |> build_row(headers)
-    end, options |> Keyword.merge(num_workers: num_workers))
-    |> handle_errors
-  end
   defp aggregate(stream, true) do
     stream |> LineAggregator.aggregate
   end

      |> Enum.take(1)
      |> List.first

    { first_line, 0 }
      |> Lexer.lex(options)
      |> lex_to_parse
      |> Parser.parse(options)
      |> handle_error_for_result!
      |> build_row(nil)
  end

  defp handle_errors(stream) do

    stream
  end

  defp check_row_length(_, false) do
    { :ok }
  end
  defp check_row_length({ data, index }, row_length) do
    actual_length = data |> Enum.count

    case actual_length do
      ^row_length -> { :ok }
      _ -> { :error, RowLengthError, "Encountered a row with length #{actual_length} instead of #{row_length}", index }
    end
  end

  defp build_row(data, headers) when is_list(headers) do
    { :ok, headers |> Enum.zip(data) |> Enum.into(%{}) }
  end
  defp build_row(data, _) do
    { :ok, data }
  end

  defp get_headers(headers, stream, _) when is_list(headers) do
    { :ok, { headers, stream } }
  end
  defp get_headers(headers, stream, options) when headers do
    with { :ok, headers} <- get_first_row(stream, options),
    do: { :ok, { headers, stream |> Stream.drop(1) } }
  end
  defp get_headers(_, stream, _) do
    { :ok, { false, stream } }
  end

  defp get_row_length(%Stream{} = stream, options) do
    with { :ok, row } <- get_first_row(stream, options),
    do: get_row_length(row)
  end
  defp get_row_length(row) do
    { :ok, Enum.count(row) }
  end

  defp get_first_row(stream, options) do
