     stream
   end
 
-  defp lex_to_parse({ :ok, tokens, index }) do
-    { tokens, index }
+  defp check_row_length(_, false) do
+    { :ok }
   end
-  defp lex_to_parse(result) do
-    result
-  end
-
-  defp check_row_length({ :ok, data, index }, row_length) do
+  defp check_row_length({ data, index }, row_length) do
     actual_length = data |> Enum.count
 
     case actual_length do
-      ^row_length -> { :ok, data, index }
+      ^row_length -> { :ok }
       _ -> { :error, RowLengthError, "Encountered a row with length #{actual_length} instead of #{row_length}", index }
     end
   end
-  defp check_row_length(error, _) do
-    error
+
+  defp build_row(data, headers) when is_list(headers) do
+    { :ok, headers |> Enum.zip(data) |> Enum.into(%{}) }
   end
-
-  defp build_row({ :ok, data, _ }, headers) when is_list(headers) do
-    headers |> Enum.zip(data) |> Enum.into(%{})
-  end
-  defp build_row({ :ok, data, _ }, _) do
-    data
-  end
-  defp build_row(error, _) do
-    error
+  defp build_row(data, _) do
+    { :ok, data }
   end
 
   defp get_headers(headers, stream, _) when is_list(headers) do
-    { headers, stream }
+    { :ok, { headers, stream } }
   end
   defp get_headers(headers, stream, options) when headers do
-    headers = stream
-              |> get_first_row(options)
-
-    { headers, stream |> Stream.drop(1) }
+    with { :ok, headers} <- get_first_row(stream, options),
+    do: { :ok, { headers, stream |> Stream.drop(1) } }
   end
   defp get_headers(_, stream, _) do
-    { false, stream }
+    { :ok, { false, stream } }
   end
 
-  defp get_row_length(stream, options) do
-    stream
-    |> get_first_row(options)
-    |> Enum.count
+  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

      iex> Enum.take(2)
      [%{:x => \"a\", :y => \"b\"}, %{:x => \"c\", :y => \"d\"}]
  """

  def decode(stream, options \\ []) do
    { headers, stream } = options
                          |> Keyword.get(:headers, false)
                          |> get_headers(stream, options)

    row_length = case headers do
      false -> stream |> get_row_length(options)
      _ -> headers |> Enum.count
    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

    process_line({ { first_line, 0 }, false }, options)
  end

  defp handle_errors(stream) do
