chore(utils_stream): simplfy the csv reader implementation

apps/emqx_utils/src/emqx_utils_stream.erl

@@ -169,14 +169,9 @@ ets(Cont, ContF) ->
 %% The .csv binary is assumed to be in UTF-8 encoding and to have a header row.
 -spec csv(binary()) -> stream(map()).
 csv(Bin) when is_binary(Bin) ->
-    CSVData = csv_data(Bin),
     Reader = fun _Iter(Headers, Lines) ->
         case csv_read_line(Lines) of
-            {ok, Line, Rest} ->
+            {Fields, Rest} ->
-                %% XXX: not support ' ' for a field?
-                Fields = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [
-                    global, trim_all
-                ]),
                 case length(Fields) == length(Headers) of
                     true ->
                         User = maps:from_list(lists:zip(Headers, Fields)),
@@ -188,27 +183,17 @@ csv(Bin) when is_binary(Bin) ->
                 []
         end
     end,
-    case get_csv_header(CSVData) of
+    HeadersAndLines = binary:split(Bin, [<<"\r">>, <<"\n">>], [global, trim_all]),
-        {ok, CSVHeaders, CSVLines} ->
+    case csv_read_line(HeadersAndLines) of
+        {CSVHeaders, CSVLines} ->
             fun() -> Reader(CSVHeaders, CSVLines) end;
-        error ->
+        eof ->
             empty()
     end.
 
-csv_data(Data) ->
+csv_read_line([Line | Lines]) ->
-    Lines = binary:split(Data, [<<"\r">>, <<"\n">>], [global, trim_all]),
+    %% XXX: not support ' ' for the field value
-    {csv_data, Lines}.
+    Fields = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
-
+    {Fields, Lines};
-get_csv_header(CSV) ->
+csv_read_line([]) ->
-    case csv_read_line(CSV) of
-        {ok, Line, NewCSV} ->
-            Seq = binary:split(Line, [<<",">>, <<" ">>, <<"\n">>], [global, trim_all]),
-            {ok, Seq, NewCSV};
-        eof ->
-            error
-    end.
-
-csv_read_line({csv_data, [Line | Lines]}) ->
-    {ok, Line, {csv_data, Lines}};
-csv_read_line({csv_data, []}) ->
     eof.