-module(glazer). -moduledoc """ Fast JSON encoding and decoding using the glaze C++ library. By default JSON `null` is represented as the atom `null`. To change it application-wide, set the `null` env key in your config: ``` {glazer, [{null, nil}]}. ``` See also [https://github.com/stephenberry/glaze] """. -export([json_decode/1, json_decode/2, json_try_decode/1, json_try_decode/2, json_encode/1, json_encode/2, json_minify/1, json_prettify/1, json_query/2, json_query/3, encode_integer/1, decode_integer/1, try_decode_integer/1, json_scan/1, json_scan/2, json_stream_decoder/0, json_stream_decoder/1, json_stream_feed/2, json_stream_eof/1, yaml_decode/1, yaml_decode/2, yaml_try_decode/1, yaml_try_decode/2, yaml_encode/1, yaml_encode/2, csv_decode/1, csv_decode/2, csv_try_decode/1, csv_try_decode/2, csv_encode/1, csv_encode/2, csv_stream_decoder/0, csv_stream_decoder/1, csv_stream_feed/2, csv_stream_eof/1]). -on_load(init/0). -define(LIBNAME, glazer). -define(NOT_LOADED_ERROR, erlang:nif_error({not_loaded, [{module, ?MODULE}, {line, ?LINE}]})). -type decode_opt() :: object_as_tuple | use_nil | {null_term, atom()} | {keys, atom | existing_atom | binary} | dedupe_keys. -doc """ Decode options: - `object_as_tuple` - decode JSON objects as `{[{K, V}]}` proplists rather than maps - `use_nil` - use the atom `nil` for JSON null - `{null_term, Atom}` - use `Atom` for JSON null - `{keys, atom}` - decode object keys as atoms - `{keys, existing_atom}` - decode keys as existing atoms, fall back to binary - `{keys, binary}` - decode keys as binaries (default) - `dedupe_keys` - with `object_as_tuple`, eliminate duplicate object keys from the resulting proplist, keeping the last occurrence's value (and position). Has no effect when objects are decoded as maps (the default) or with `{keys, atom | existing_atom}`: a JSON object with duplicate keys is always deduped (last value wins) when decoded to a map, since maps cannot represent duplicate keys. """. -type decode_opts() :: [decode_opt()]. -type encode_opt() :: pretty | uescape | force_utf8 | use_nil | {null_term, atom()}. -doc """ Encode options: - `pretty` - pretty-print the JSON output - `uescape` - escape non-ASCII characters as \\uXXXX sequences - `force_utf8` - fix invalid UTF-8 sequences before encoding - `use_nil` - encode the atom `nil` as JSON `null` - `{null_term, Atom}` - encode `Atom` as JSON `null` """. -type encode_opts() :: [encode_opt()]. -type yaml_decode_opt() :: use_nil | {null_term, atom()} | {keys, atom | existing_atom | binary} | yaml_1_1_bools. -doc """ YAML decode options: - `use_nil` - use the atom `nil` for YAML `null`/`~`/empty values - `{null_term, Atom}` - use `Atom` for YAML `null`/`~`/empty values - `{keys, atom}` - decode mapping keys as atoms - `{keys, existing_atom}` - decode mapping keys as existing atoms, fall back to binary - `{keys, binary}` - decode mapping keys as binaries (default) - `yaml_1_1_bools` - additionally treat `yes`/`no`/`on`/`off` (and case variants) as booleans, per the YAML 1.1 core schema. By default (YAML 1.2 core schema) only `true`/`false` are recognized as booleans. """. -type yaml_decode_opts() :: [yaml_decode_opt()]. -type yaml_encode_opt() :: use_nil | {null_term, atom()}. -doc """ YAML encode options: - `use_nil` - treat the atom `nil` as YAML `null` - `{null_term, Atom}` - treat `Atom` as YAML `null` """. -type yaml_encode_opts() :: [yaml_encode_opt()]. -type csv_decode_opt() :: {delimiter, char()} | headers | {keys, atom | existing_atom | binary}. -doc """ CSV decode options: - `{delimiter, Char}` - field delimiter (default `$,`) - `headers` - treat the first row as column names and decode each subsequent row as a map keyed by those names, instead of returning every row as a list of fields - `{keys, atom}` - with `headers`, decode column names as atoms - `{keys, existing_atom}` - with `headers`, decode column names as existing atoms, falling back to binaries for unknown atoms - `{keys, binary}` - with `headers`, decode column names as binaries (default) """. -type csv_decode_opts() :: [csv_decode_opt()]. -type csv_encode_opt() :: {delimiter, char()} | headers | {line_ending, lf | crlf}. -doc """ CSV encode options: - `{delimiter, Char}` - field delimiter (default `$,`) - `headers` - input is a list of maps; the first map's keys become the header row, and subsequent maps are encoded as rows in that column order (missing keys produce empty fields) - `{line_ending, lf | crlf}` - line terminator (default `crlf`, per RFC 4180) """. -type csv_encode_opts() :: [csv_encode_opt()]. -type json_query_reason() :: enomem | jq_not_available | jq_decode_error | {jq_compile_error, binary()} | invalid_input | binary(). -export_type([decode_opts/0, encode_opts/0, yaml_decode_opts/0, yaml_encode_opts/0, csv_decode_opts/0, csv_encode_opts/0, json_query_reason/0]). -type scan_state() :: tuple(). -record(json_stream_decoder, { opts = [] :: decode_opts(), buffer = <<>> :: binary(), state = undefined :: scan_state() | undefined }). -opaque json_stream_decoder() :: #json_stream_decoder{}. -export_type([json_stream_decoder/0]). -type csv_scan_state() :: {non_neg_integer(), boolean()}. -record(csv_stream_decoder, { opts = [] :: csv_decode_opts(), buffer = <<>> :: binary(), header = undefined :: binary() | undefined, state = {0, false} :: csv_scan_state() }). -opaque csv_stream_decoder() :: #csv_stream_decoder{}. -export_type([csv_stream_decoder/0]). -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). -endif. init() -> NullVal = application:get_env(?LIBNAME, null, null), is_atom(NullVal) orelse erlang:error("glazer: option 'null' must be an atom"), SoName = case code:priv_dir(?LIBNAME) of {error, bad_name} -> case code:which(?MODULE) of Filename when is_list(Filename) -> Dir = filename:dirname(filename:dirname(Filename)), filename:join([Dir, "priv", ?LIBNAME]); _ -> filename:join("../priv", ?LIBNAME) end; Dir -> filename:join(Dir, ?LIBNAME) end, erlang:load_nif(SoName, [{null, NullVal}]). -doc """ Decode a JSON binary or iolist to an Erlang term. JSON objects are returned as maps (default). Raises `{parse_error, Msg}` on invalid input. """. -spec json_decode(binary() | iolist()) -> term(). json_decode(Input) -> case json_try_decode(Input) of {ok, Term} -> Term; {error, Reason} -> error({parse_error, Reason}) end. -doc """ Decode a JSON binary or iolist to an Erlang term with options. Raises `{parse_error, Reason}` on invalid input. """. -spec json_decode(binary() | iolist(), decode_opts()) -> term(). json_decode(Input, Opts) -> case json_try_decode(Input, Opts) of {ok, Term} -> Term; {error, Reason} -> error({parse_error, Reason}) end. -doc """ Decode a JSON binary or iolist, returning `{ok, Term}` or `{error, Reason}` instead of raising. """. -spec json_try_decode(binary() | iolist()) -> {ok, term()} | {error, binary()}. json_try_decode(_Input) -> ?NOT_LOADED_ERROR. -doc """ Decode a JSON binary or iolist with options, returning `{ok, Term}` or `{error, Reason}` instead of raising. """. -spec json_try_decode(binary() | iolist(), decode_opts()) -> {ok, term()} | {error, binary()}. json_try_decode(_Input, _Opts) -> ?NOT_LOADED_ERROR. -doc """ Decode a YAML binary or iolist to an Erlang term. YAML mappings are returned as maps (default). Raises `{parse_error, Reason}` on invalid input. """. -spec yaml_decode(binary() | iolist()) -> term(). yaml_decode(Input) -> case yaml_try_decode(Input) of {ok, Term} -> Term; {error, Reason} -> error(Reason) end. -doc """ Decode a YAML binary or iolist to an Erlang term with options. Raises `{parse_error, Msg}` on invalid input. """. -spec yaml_decode(binary() | iolist(), yaml_decode_opts()) -> term(). yaml_decode(Input, Opts) -> case yaml_try_decode(Input, Opts) of {ok, Term} -> Term; {error, Reason} -> error({parse_error, Reason}) end. -doc """ Decode a YAML binary or iolist, returning `{ok, Term}` or `{error, Msg}` instead of raising. """. -spec yaml_try_decode(binary() | iolist()) -> {ok, term()} | {error, binary()}. yaml_try_decode(_Input) -> ?NOT_LOADED_ERROR. -doc """ Decode a YAML binary or iolist with options, returning `{ok, Term}` or `{error, Msg}` instead of raising. """. -spec yaml_try_decode(binary() | iolist(), yaml_decode_opts()) -> {ok, term()} | {error, binary()}. yaml_try_decode(_Input, _Opts) -> ?NOT_LOADED_ERROR. -doc """ Encode an Erlang term to a YAML binary in block style (2-space indentation, sequences at the same indentation as the mapping key that owns them). """. -spec yaml_encode(term()) -> binary(). yaml_encode(Data) -> yaml_encode(Data, []). -doc "Encode an Erlang term to a YAML binary in block style with options.". -spec yaml_encode(term(), yaml_encode_opts()) -> binary(). yaml_encode(_Data, _Opts) -> ?NOT_LOADED_ERROR. -doc """ Decode a CSV binary or iolist to a list of rows. By default each row is a list of binary fields. With the `headers` option, the first row is used as column names and each subsequent row is decoded as a map. Raises `unterminated_quoted_field` or `duplicate_header` on invalid input. """. -spec csv_decode(binary() | iolist()) -> [[binary()]] | [#{binary() => binary()}]. csv_decode(Input) -> csv_decode(Input, []). -doc """ Decode a CSV binary or iolist to a list of rows, with options. Raises `Reason::atom()` (`unterminated_quoted_field` or `duplicate_header`) on invalid input. """. -spec csv_decode(binary() | iolist(), csv_decode_opts()) -> [[binary()]] | [map()]. csv_decode(Input, Opts) -> case csv_try_decode(Input, Opts) of {ok, Rows} -> Rows; {error, Reason} -> error(Reason) end. -doc """ Decode a CSV binary or iolist, returning `{ok, Rows}` or `{error, Reason}` instead of raising, where `Reason` is `unterminated_quoted_field` or `duplicate_header`. """. -spec csv_try_decode(binary() | iolist()) -> {ok, [[binary()]]} | {error, atom()}. csv_try_decode(Input) -> csv_try_decode(Input, []). -doc """ Decode a CSV binary or iolist with options, returning `{ok, Rows}` or `{error, Reason}` instead of raising, where `Reason` is `unterminated_quoted_field` or `duplicate_header`. """. -spec csv_try_decode(binary() | iolist(), csv_decode_opts()) -> {ok, [[binary()]] | [map()]} | {error, atom()}. csv_try_decode(_Input, _Opts) -> ?NOT_LOADED_ERROR. -doc """ Encode a list of rows to a CSV binary. Each row is a list of fields (binaries, atoms, integers, or floats). Fields containing the delimiter, a double quote, or a line break are quoted per RFC 4180, with embedded quotes doubled. """. -spec csv_encode([[term()]] | [map()]) -> binary(). csv_encode(Data) -> csv_encode(Data, []). -doc """ Encode a list of rows to a CSV binary, with options. With the `headers` option, `Data` is a list of maps: the first map's keys become the header row (in iteration order), and each map is encoded as a row in that column order. """. -spec csv_encode([[term()]] | [map()], csv_encode_opts()) -> binary(). csv_encode(_Data, _Opts) -> ?NOT_LOADED_ERROR. %%%---------------------------------------------------------------------------- %%% CSV streaming / incremental decode %%%---------------------------------------------------------------------------- -doc """ Create a new incremental decoder for feeding CSV in chunks (e.g. from a socket or file), useful when the whole input isn't available up front. Each complete row is decoded as soon as its terminating line break is seen, via [`csv_decode/2`](`csv_decode/2`) on that single row. Only the *row boundary detection* is incremental — a small byte-scanner tracks whether the cursor is inside a quoted field across chunks, so that `\n`/`\r\n` inside quoted fields doesn't end a row. With the `headers` option, the first complete row is captured as the header and used to decode every subsequent row as a map; no row is emitted for the header itself. ## Example ```erlang 1> D0 = glazer:csv_stream_decoder(), 2> {Rows1, D1} = glazer:csv_stream_feed(D0, <<"a,b\n1,2\n3,">>), 3> Rows1. [[<<"a">>,<<"b">>],[<<"1">>,<<"2">>]] 4> {Rows2, D2} = glazer:csv_stream_feed(D1, <<"4\n">>), 5> Rows2. [[<<"3">>,<<"4">>]] 6> glazer:csv_stream_eof(D2). {ok, []} ``` """. -spec csv_stream_decoder() -> csv_stream_decoder(). csv_stream_decoder() -> csv_stream_decoder([]). -doc """ Create a new incremental CSV decoder, passing `Opts` through to every [`csv_decode/2`](`csv_decode/2`) call. """. -spec csv_stream_decoder(csv_decode_opts()) -> csv_stream_decoder(). csv_stream_decoder(Opts) when is_list(Opts) -> #csv_stream_decoder{opts = Opts}. -doc """ Feed a chunk of bytes into the decoder, returning any complete CSV rows found so far (in order) along with the updated decoder. Raises the same exceptions as [`csv_decode/2`](`csv_decode/2`) if a row that the scanner deemed complete fails to decode. ## Example ```erlang loop(Socket, D0) -> case gen_tcp:recv(Socket, 0) of {ok, Chunk} -> {Rows, D1} = glazer:csv_stream_feed(D0, Chunk), handle_rows(Rows), loop(Socket, D1); {error, closed} -> case glazer:csv_stream_eof(D0) of {ok, Trailing} -> handle_rows(Trailing); {error, Reason} -> handle_truncated_stream(Reason) end end. ``` """. -spec csv_stream_feed(csv_stream_decoder(), binary() | iolist()) -> {[[binary()]] | [map()], csv_stream_decoder()}. csv_stream_feed(#csv_stream_decoder{buffer = Buf} = D, Chunk) -> NewBuf = iolist_to_binary([Buf, Chunk]), csv_stream_drain(D#csv_stream_decoder{buffer = NewBuf}, []). csv_stream_drain(#csv_stream_decoder{buffer = Buf, state = St} = D, Acc) -> case csv_scan_row(Buf, St) of {complete, End, RestStart} -> <> = Buf, {Row, D1} = csv_stream_decode_row(D, RowBin), D2 = D1#csv_stream_decoder{buffer = Rest, state = {0, false}}, case Row of skip -> csv_stream_drain(D2, Acc); _ -> csv_stream_drain(D2, [Row | Acc]) end; {incomplete, NewSt} -> {lists:reverse(Acc), D#csv_stream_decoder{state = NewSt}} end. %% Decode a single complete row (without its line terminator). Returns `skip` %% for a blank line (matching csv_decode/2, which skips blank lines), or for %% the header row when the `headers` option is set (it's captured but not %% emitted as a row). csv_stream_decode_row(#csv_stream_decoder{opts = Opts, header = undefined} = D, RowBin) -> case is_blank(RowBin) of true -> {skip, D}; false -> case proplists:get_bool(headers, Opts) of true -> {skip, D#csv_stream_decoder{header = RowBin}}; false -> [Row] = csv_decode(RowBin, Opts -- [headers]), {Row, D} end end; csv_stream_decode_row(#csv_stream_decoder{opts = Opts, header = Header} = D, RowBin) -> case is_blank(RowBin) of true -> {skip, D}; false -> [Row] = csv_decode(<
>, Opts), {Row, D} end. -doc """ Signal end-of-stream: decode any remaining buffered bytes as a final row (useful when the input doesn't end with a trailing line break). Returns `{ok, Rows}` with zero or one trailing row, or `{error, Reason}` if the remaining bytes don't form a valid row. ## Example ```erlang 1> D0 = glazer:csv_stream_decoder(), 2> {Rows1, D1} = glazer:csv_stream_feed(D0, <<"a,b\n1,2">>), 3> Rows1. [[<<"a">>,<<"b">>]] 4> glazer:csv_stream_eof(D1). {ok, [[<<"1">>,<<"2">>]]} ``` """. -spec csv_stream_eof(csv_stream_decoder()) -> {ok, [[binary()]] | [map()]} | {error, term()}. csv_stream_eof(#csv_stream_decoder{buffer = Buf} = D) -> case is_blank(Buf) of true -> {ok, []}; false -> try csv_stream_decode_row(D, Buf) of {skip, _D1} -> {ok, []}; {Row, _D1} -> {ok, [Row]} catch error:Reason -> {error, Reason} end end. %% Resumable scan for the next CSV row terminator (`\n` or `\r\n`) outside of %% quoted fields, starting from `State = {Pos, InQuotes}`. %% %% Returns `{complete, End, RestStart}` where `binary:part(Bin, 0, End)` is %% the row (without its terminator) and `RestStart` is the offset of the %% first byte after the terminator, or `{incomplete, NewState}` if `Bin` %% doesn't yet contain a complete row. -spec csv_scan_row(binary(), csv_scan_state()) -> {complete, non_neg_integer(), non_neg_integer()} | {incomplete, csv_scan_state()}. csv_scan_row(Bin, {Pos, InQuotes}) -> csv_scan_row(Bin, byte_size(Bin), Pos, InQuotes). csv_scan_row(Bin, Size, Pos, InQuotes) when Pos < Size -> case binary:at(Bin, Pos) of $" -> csv_scan_row(Bin, Size, Pos + 1, not InQuotes); $\n when not InQuotes -> {complete, Pos, Pos + 1}; $\r when not InQuotes, Pos + 1 < Size -> case binary:at(Bin, Pos + 1) of $\n -> {complete, Pos, Pos + 2}; _ -> csv_scan_row(Bin, Size, Pos + 1, InQuotes) end; $\r when not InQuotes -> %% lone `\r` at the end of the buffer: might be the start of `\r\n`, %% so wait for more data before deciding {incomplete, {Pos, InQuotes}}; _ -> csv_scan_row(Bin, Size, Pos + 1, InQuotes) end; csv_scan_row(_Bin, _Size, Pos, InQuotes) -> {incomplete, {Pos, InQuotes}}. -doc "Encode an Erlang term to a JSON binary.". -spec json_encode(term()) -> binary(). json_encode(Data) -> json_encode(Data, []). -doc "Encode an Erlang term to a JSON binary with options.". -spec json_encode(term(), encode_opts()) -> binary(). json_encode(_Data, _Opts) -> ?NOT_LOADED_ERROR. -doc "Minify a JSON binary or iolist, removing all unnecessary whitespace.". -spec json_minify(binary() | iolist()) -> binary(). json_minify(_Input) -> ?NOT_LOADED_ERROR. -doc "Pretty-print a JSON binary or iolist with two-space indentation.". -spec json_prettify(binary() | iolist()) -> binary(). json_prettify(_Input) -> ?NOT_LOADED_ERROR. -doc """ Run a [jq](https://jqlang.org/) `Filter` program against a JSON binary or iolist `Input`, returning one Erlang term per value produced by the filter (in the order they are emitted by jq). Requires `glazer` to have been built against `libjq`; if `libjq` was not available at build time, this returns `{error, jq_not_available}`. A runtime error raised by the filter itself (e.g. via jq's `error/0,1`) is returned as `{error, Msg}` where `Msg` is the binary message produced by jq. ``` 1> glazer:json_query(<<"{\\"a\\":[1,2,3]}">>, <<".a[]">>). {ok,[1,2,3]} 2> glazer:json_query(<<"{\\"a\\":1}">>, <<".b">>). {ok,[null]} 3> glazer:json_query(<<"not json">>, <<".">>). {error, invalid_input} ``` """. -spec json_query(binary() | iolist(), binary() | iolist()) -> {ok, [term()]} | {error, json_query_reason()}. json_query(_Input, _Filter) -> ?NOT_LOADED_ERROR. -doc """ Like `json_query/2`, but decodes each result term using `DecodeOpts` (see `json_decode/2`). """. -spec json_query(binary() | iolist(), binary() | iolist(), decode_opts()) -> {ok, [term()]} | {error, json_query_reason()}. json_query(_Input, _Filter, _DecodeOpts) -> ?NOT_LOADED_ERROR. -doc """ Encode an integer to its JSON string representation. Raises `badarg` if `Int` is not an integer. """. -spec encode_integer(integer()) -> binary(). encode_integer(_Int) -> ?NOT_LOADED_ERROR. -doc """ Decode a JSON number string to an integer. Raises `invalid_number_format` on invalid input. """. -spec decode_integer(binary() | iolist()) -> integer(). decode_integer(NumberString) -> case try_decode_integer(NumberString) of {ok, Int} -> Int; {error, Reason} -> error(Reason) end. -doc """ Decode a JSON number string to an integer, returning `{ok, Int}` or `{error, invalid_number_format}` instead of raising. """. -spec try_decode_integer(binary() | iolist()) -> {ok, integer()} | {error, invalid_number_format}. try_decode_integer(_NumberString) -> ?NOT_LOADED_ERROR. -doc """ Locate the end of the next complete top-level JSON value in `Bin`, without decoding it. Returns: - `{complete, EndOffset}` - a complete value spans `binary:part(Bin, 0, EndOffset)`; the rest of `Bin` (if any) is left over for the next call - `{incomplete, ScanState}` - `Bin` doesn't yet contain a complete value; feed more data via `json_scan/2` once it's available, passing the *entire unconsumed remainder* (this `Bin`, with new bytes appended) plus `ScanState` This is the low-level primitive behind [`json_stream_feed/2`](`json_stream_feed/2`); most callers should use the `stream_*` API instead. ## Example Slicing off complete values from a buffer of concatenated JSON: ```erlang 1> Buf0 = <<"{\"a\":1} {\"b\":2}">>, 2> {complete, End1} = glazer:json_scan(Buf0). {complete, 7} 3> <> = Buf0, 4> Val1. <<"{\"a\":1}">> 5> Buf1. <<" {\"b\":2}">> 6> {complete, End2} = glazer:json_scan(Buf1). {complete, 8} ``` Resuming a scan once more bytes arrive: ```erlang 1> {incomplete, S0} = glazer:json_scan(<<"{\"a\":">>). {incomplete, {6,1,false,false,true,false}} 2> glazer:json_scan(<<"{\"a\":1}">>, S0). {complete, 7} ``` """. -spec json_scan(binary() | iolist()) -> {complete, non_neg_integer()} | {incomplete, scan_state()}. json_scan(_Bin) -> ?NOT_LOADED_ERROR. -doc """ Resume scanning `Bin` (the unconsumed remainder plus newly-appended bytes) from `ScanState`. """. -spec json_scan(binary() | iolist(), scan_state()) -> {complete, non_neg_integer()} | {incomplete, scan_state()}. json_scan(_Bin, _ScanState) -> ?NOT_LOADED_ERROR. %%%---------------------------------------------------------------------------- %%% Streaming / incremental decode %%%---------------------------------------------------------------------------- -doc """ Create a new incremental decoder for feeding JSON in chunks (e.g. from a socket or file), useful when a complete document isn't available up front or when a stream contains a sequence of concatenated/whitespace-separated JSON values (e.g. newline-delimited JSON). Decoding itself is **not** incremental — each complete top-level value is still decoded in a single pass via [`json_decode/2`](`json_decode/2`) using the library's fast whole-buffer decoder. Only the *boundary detection* (finding where one value ends and the next begins) is incremental, via a small byte-scanner that tracks nesting/string state across chunks. ## Example ```erlang 1> D0 = glazer:json_stream_decoder(), 2> {Vals1, D1} = glazer:json_stream_feed(D0, <<"{\"a\":1} {\"b\":">>), 3> Vals1. [#{<<"a">> => 1}] 4> {Vals2, _D2} = glazer:json_stream_feed(D1, <<"2}">>), 5> Vals2. [#{<<"b">> => 2}] ``` """. -spec json_stream_decoder() -> json_stream_decoder(). json_stream_decoder() -> json_stream_decoder([]). -doc """ Create a new incremental decoder, passing `Opts` through to every [`json_decode/2`](`json_decode/2`) call. """. -spec json_stream_decoder(decode_opts()) -> json_stream_decoder(). json_stream_decoder(Opts) when is_list(Opts) -> #json_stream_decoder{opts = Opts}. -doc """ Feed a chunk of bytes into the decoder, returning any complete JSON values found so far (in order) along with the updated decoder. Raises the same exceptions as [`json_decode/2`](`json_decode/2`) (e.g. `Reason`) if a value that the scanner deemed complete fails to decode. ## Example Call `json_stream_feed/2` for each chunk received from the source while more data may still arrive, and [`json_stream_eof/1`](`json_stream_eof/1`) once the source is exhausted to flush any trailing value: ```erlang loop(Socket, D0) -> case gen_tcp:recv(Socket, 0) of {ok, Chunk} -> {Vals, D1} = glazer:json_stream_feed(D0, Chunk), handle_values(Vals), loop(Socket, D1); {error, closed} -> case glazer:json_stream_eof(D0) of {ok, Trailing} -> handle_values(Trailing); {error, Reason} -> handle_truncated_stream(Reason) end end. ``` The same decoder fits naturally into a `gen_server` driving an active-mode socket: keep the `json_stream_decoder()` in the process state, feed it from `handle_info({tcp, ...})`, and flush it on `{tcp_closed, ...}`: ```erlang -module(json_conn). -behaviour(gen_server). -export([start_link/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2]). -record(state, {socket, decoder}). start_link(Socket) -> gen_server:start_link(?MODULE, Socket, []). init(Socket) -> inet:setopts(Socket, [{active, once}]), {ok, #state{socket = Socket, decoder = glazer:json_stream_decoder()}}. handle_info({tcp, Socket, Data}, #state{socket = Socket, decoder = D0} = State) -> {Vals, D1} = glazer:json_stream_feed(D0, Data), lists:foreach(fun handle_value/1, Vals), inet:setopts(Socket, [{active, once}]), {noreply, State#state{decoder = D1}}; handle_info({tcp_closed, Socket}, #state{socket = Socket, decoder = D0} = State) -> case glazer:json_stream_eof(D0) of {ok, Trailing} -> lists:foreach(fun handle_value/1, Trailing); {error, Reason} -> handle_truncated_stream(Reason) end, {stop, normal, State}; handle_info({tcp_error, Socket, Reason}, #state{socket = Socket} = State) -> {stop, Reason, State}. handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(_Request, State) -> {noreply, State}. handle_value(Val) -> io:format("received: ~p~n", [Val]). ``` """. -spec json_stream_feed(json_stream_decoder(), binary() | iolist()) -> {[term()], json_stream_decoder()}. json_stream_feed(#json_stream_decoder{buffer = Buf} = D, Chunk) -> NewBuf = iolist_to_binary([Buf, Chunk]), json_stream_drain(D#json_stream_decoder{buffer = NewBuf, state = undefined}, []). json_stream_drain(#json_stream_decoder{buffer = Buf, opts = Opts} = D, Acc) -> case json_scan(Buf) of {complete, End} -> <> = Buf, Term = json_decode(ValueBin, Opts), json_stream_drain(D#json_stream_decoder{buffer = Rest, state = undefined}, [Term | Acc]); {incomplete, NewSt} -> {lists:reverse(Acc), D#json_stream_decoder{state = NewSt}} end. -doc """ Signal end-of-stream: decode any remaining buffered bytes as a final value (useful for a trailing bare scalar, e.g. a lone number or `true`/`null`, which the scanner can't otherwise distinguish from a value that's still being written to mid-chunk). Returns `{ok, [Term]}` with zero or one trailing value, or `{error, Reason}` if the remaining bytes don't form a complete value. ## Example ```erlang 1> D0 = glazer:json_stream_decoder(), 2> {Vals1, D1} = glazer:json_stream_feed(D0, <<"123">>), 3> Vals1. [] 4> glazer:json_stream_eof(D1). {ok, [123]} ``` A stream that ends mid-value (e.g. a dropped connection) yields an error instead of silently dropping the partial data: ```erlang 1> D0 = glazer:json_stream_decoder(), 2> {Vals1, D1} = glazer:json_stream_feed(D0, <<"{\"a\":1, \"b\":">>), 3> Vals1. [] 4> glazer:json_stream_eof(D1). {error, _Reason} ``` """. -spec json_stream_eof(json_stream_decoder()) -> {ok, [term()]} | {error, term()}. json_stream_eof(#json_stream_decoder{buffer = Buf, opts = Opts}) -> case is_blank(Buf) of true -> {ok, []}; false -> try json_decode(Buf, Opts) of Term -> {ok, [Term]} catch error:Reason -> {error, Reason} end end. %% True if `Bin` is empty or contains only JSON whitespace (space, tab, CR, LF). is_blank(Bin) -> lists:all(fun(B) -> B =:= $\s orelse B =:= $\t orelse B =:= $\r orelse B =:= $\n end, binary_to_list(Bin)). %%%---------------------------------------------------------------------------- %%% Tests %%%---------------------------------------------------------------------------- -ifdef(EUNIT). encode_test_() -> [ ?_assertEqual(<<"null">>, json_encode(null)), ?_assertEqual(<<"null">>, json_encode(nil, [use_nil])), ?_assertEqual(<<"true">>, json_encode(true)), ?_assertEqual(<<"false">>, json_encode(false)), ?_assertEqual(<<"1">>, json_encode(1)), ?_assertEqual(<<"1.5">>, json_encode(1.5)), ?_assertEqual(<<"\"hello\"">>, json_encode(<<"hello">>)), ?_assertEqual(<<"[1,2,3]">>, json_encode([1, 2, 3])), ?_assertEqual(<<"{}">>, json_encode(#{})), ?_assertMatch(<<"{", _/binary>>, json_encode(#{<<"a">> => 1})) ]. decode_test_() -> [ ?_assertEqual(null, json_decode(<<"null">>)), ?_assertEqual(nil, json_decode(<<"null">>, [use_nil])), ?_assertEqual(true, json_decode(<<"true">>)), ?_assertEqual(false, json_decode(<<"false">>)), ?_assertEqual(1, json_decode(<<"1">>)), ?_assertEqual(1.5, json_decode(<<"1.5">>)), ?_assertEqual(<<"hello">>, json_decode(<<"\"hello\"">>)), ?_assertEqual([1, 2, 3], json_decode(<<"[1,2,3]">>)), ?_assertEqual(#{<<"a">> => 1}, json_decode(<<"{\"a\":1}">>)), ?_assertEqual({[{<<"a">>, 1}]}, json_decode(<<"{\"a\":1}">>, [object_as_tuple])), ?_assertEqual(null, json_decode(<<"null">>, [{null_term, null}])), ?_assertEqual(my_null, json_decode(<<"null">>, [{null_term, my_null}])) ]. roundtrip_test_() -> Vals = [null, true, false, 0, 1, -1, 1.5, <<"hello">>, [], [1, 2, 3], #{<<"a">> => 1, <<"b">> => [1, 2]}, #{<<"nested">> => #{<<"x">> => true}}], [?_assertEqual(V, json_decode(json_encode(V))) || V <- Vals]. minify_test_() -> [ ?_assertEqual(<<"[1,2,3]">>, json_minify(<<"[ 1, 2, 3 ]">>)), ?_assertEqual(<<"{\"a\":1}">>, json_minify(<<" { \"a\" : 1 } ">>)) ]. prettify_test_() -> [ ?_assertMatch(<<"[\n", _/binary>>, json_prettify(<<"[1,2,3]">>)), ?_assertMatch(<<"{\n", _/binary>>, json_prettify(<<"{\"a\":1}">>)) ]. json_query_test_() -> case json_query(<<"{\"a\":1}">>, <<".a">>) of {error, jq_not_available} -> %% libjq was not available at build time — skip. []; _ -> [ ?_assertEqual({ok, [1, 2, 3]}, json_query(<<"{\"a\":[1,2,3]}">>, <<".a[]">>)), ?_assertEqual({ok, [null]}, json_query(<<"{\"a\":1}">>, <<".b">>)), ?_assertEqual({ok, [#{a => #{b => 2}}]}, json_query(<<"{\"a\":{\"b\":2}}">>, <<".">>, [{keys, atom}])), ?_assertEqual({error, invalid_input}, json_query(<<"not json">>, <<".">>)), ?_assertMatch({error, {jq_compile_error, _}}, json_query(<<"{\"a\":1}">>, <<"bad syntax (((">>)), %% multiple emitted values, in order ?_assertEqual({ok, [1, 2, 3]}, json_query(<<"[1,2,3]">>, <<".[]">>)), %% no emitted values at all ?_assertEqual({ok, []}, json_query(<<"[1,2,3]">>, <<".[] | select(. > 10)">>)), %% iolist input/filter ?_assertEqual({ok, [2]}, json_query([<<"{\"a\":">>, <<"2}">>], [<<".">>, <<"a">>])), %% scalar, string, boolean, and nested results ?_assertEqual({ok, [<<"x">>, true, [1, 2]]}, json_query(<<"{\"a\":\"x\",\"b\":true,\"c\":[1,2]}">>, <<".a, .b, .c">>)), %% runtime error raised by the filter itself ?_assertMatch({error, Msg} when is_binary(Msg), json_query(<<"{\"a\":1}">>, <<".a | error(\"boom\")">>)), %% decode options threaded through to each result ?_assertEqual({ok, [#{<<"b">> => null}]}, json_query(<<"{\"a\":{\"b\":null}}">>, <<".a">>)), ?_assertEqual({ok, [#{<<"b">> => nil}]}, json_query(<<"{\"a\":{\"b\":null}}">>, <<".a">>, [{null_term, nil}])) ] end. keys_test_() -> [ ?_assertEqual(#{<<"a">> => 1}, json_decode(<<"{\"a\":1}">>)), ?_assertEqual(#{<<"a">> => 1}, json_decode(<<"{\"a\":1}">>, [{keys, binary}])), ?_assertEqual(#{a => 1}, json_decode(<<"{\"a\":1}">>, [{keys, atom}])), ?_assertEqual(#{a => 1}, json_decode(<<"{\"a\":1}">>, [{keys, existing_atom}])), %% existing_atom falls back to a binary for keys with no matching atom ?_assertEqual(#{<<"no_such_atom_in_glazer_test_suite_xyz">> => 1}, json_decode(<<"{\"no_such_atom_in_glazer_test_suite_xyz\":1}">>, [{keys, existing_atom}])) ]. uescape_test_() -> [ %% U+00E9 (é), UTF-8: 0xC3 0xA9 ?_assertEqual(<<"\"\\u00e9\"">>, json_encode(<<16#C3, 16#A9>>, [uescape])), %% Without uescape, UTF-8 bytes pass through unescaped ?_assertEqual(<<"\"", 16#C3, 16#A9, "\"">>, json_encode(<<16#C3, 16#A9>>)), %% U+1F600 (emoji, outside the BMP) encodes as a surrogate pair ?_assertEqual(<<"\"\\ud83d\\ude00\"">>, json_encode(<<16#F0,16#9F,16#98,16#80>>, [uescape])), %% Round-trips back to the original UTF-8 binary ?_assertEqual(<<16#C3, 16#A9>>, json_decode(json_encode(<<16#C3, 16#A9>>, [uescape]))) ]. force_utf8_test_() -> [ %% Invalid byte sequences are sanitized to U+FFFD (UTF-8: EF BF BD) ?_assertEqual(<<"\"", 16#EF, 16#BF, 16#BD, 16#EF, 16#BF, 16#BD, "a\"">>, json_encode(<<16#FF, 16#FE, $a>>, [force_utf8])), %% Without force_utf8, invalid bytes pass through verbatim ?_assertEqual(<<"\"", 16#FF, 16#FE, "a\"">>, json_encode(<<16#FF, 16#FE, $a>>)), %% Valid UTF-8 is left untouched ?_assertEqual(<<"\"", 16#C3, 16#A9, "\"">>, json_encode(<<16#C3, 16#A9>>, [force_utf8])) ]. pretty_test_() -> [ ?_assertEqual(<<"{\n \"a\": 1\n}">>, json_encode(#{<<"a">> => 1}, [pretty])), ?_assertEqual(<<"[\n 1,\n 2\n]">>, json_encode([1, 2], [pretty])), ?_assertEqual(#{<<"a">> => 1}, json_decode(json_encode(#{<<"a">> => 1}, [pretty]))) ]. null_term_encode_test_() -> [ ?_assertEqual(<<"null">>, json_encode(null)), ?_assertEqual(<<"null">>, json_encode(nil, [{null_term, nil}])), ?_assertEqual(<<"null">>, json_encode(undefined, [{null_term, undefined}])), ?_assertEqual(<<"null">>, json_encode(null, [{null_term, undefined}])) ]. object_as_tuple_test_() -> [ ?_assertEqual({[]}, json_decode(<<"{}">>, [object_as_tuple])), ?_assertEqual({[{<<"a">>, 1}, {<<"b">>, 2}]}, json_decode(<<"{\"a\":1,\"b\":2}">>, [object_as_tuple])), ?_assertEqual({[{<<"a">>, {[{<<"b">>, true}]}}]}, json_decode(<<"{\"a\":{\"b\":true}}">>, [object_as_tuple])), ?_assertEqual({[{<<"a">>, 1}]}, json_decode(json_encode({[{<<"a">>, 1}]}), [object_as_tuple])) ]. numbers_test_() -> [ ?_assertEqual(0, json_decode(<<"0">>)), ?_assertEqual(-1, json_decode(<<"-1">>)), ?_assertEqual(-1.5, json_decode(<<"-1.5">>)), ?_assertEqual(1.0e10, json_decode(<<"1.0e10">>)), ?_assertEqual(1.0e-10, json_decode(<<"1.0e-10">>)), ?_assertEqual(<<"-1">>, json_encode(-1)), ?_assertEqual(<<"0">>, json_encode(0)) ]. iolist_input_test_() -> [ ?_assertEqual(#{<<"a">> => 1}, json_decode([<<"{\"a\":">>, <<"1}">>])), ?_assertEqual(#{<<"a">> => 1}, json_decode([<<"{">>, [<<"\"a\":1">>], <<"}">>])), ?_assertEqual(<<"{\"a\":1}">>, json_minify([<<"{ \"a\"">>, <<": 1 }">>])) ]. decode_error_test_() -> [ ?_assertError(_, json_decode(<<"">>)), ?_assertError(_, json_decode(<<"{\"a\":}">>)), ?_assertError(_, json_decode(<<"{\"a\":1">>)), ?_assertError(_, json_decode(<<"[1, 2">>)), ?_assertError(_, json_decode(<<"not json">>)) ]. bigint_test_() -> Big = 123456789012345678901234567890, Neg = -Big, [ ?_assertEqual(<<"123456789012345678901234567890">>, encode_integer(Big)), ?_assertEqual(<<"-123456789012345678901234567890">>, encode_integer(Neg)), ?_assertError(badarg, encode_integer(<<"not an integer">>)), ?_assertEqual({ok, Big}, try_decode_integer(<<"123456789012345678901234567890">>)), ?_assertEqual({ok, Neg}, try_decode_integer(<<"-123456789012345678901234567890">>)), ?_assertEqual({ok, 123}, try_decode_integer(<<"123">>)), ?_assertEqual(Big, decode_integer(<<"123456789012345678901234567890">>)), ?_assertEqual(123, decode_integer(<<"123">>)), ?_assertError(invalid_number_format, decode_integer(<<"not a number">>)), ?_assertEqual({error, invalid_number_format}, try_decode_integer(<<"not a number">>)), ?_assertEqual(Big, json_decode(<<"123456789012345678901234567890">>)), ?_assertEqual(<<"123456789012345678901234567890">>, json_encode(Big)), ?_assertEqual(Big, json_decode(json_encode(Big))) ]. scan_test_() -> [ ?_assertEqual({complete, 7}, json_scan(<<"{\"a\":1}">>)), ?_assertEqual({complete, 7}, json_scan(<<"{\"a\":1} {\"b\":2}">>)), ?_assertEqual({complete, 13}, json_scan(<<"[1,2,[3,4],5]rest">>)), ?_assertMatch({incomplete, _}, json_scan(<<"{\"a\":">>)), ?_assertMatch({incomplete, _}, json_scan(<<"123">>)), %% resume across a chunk boundary: caller passes the whole buffer + new %% bytes along with the previously-returned state ?_test(begin Part1 = <<"{\"a\":">>, Part2 = <<"1}">>, {incomplete, S1} = json_scan(Part1), ?assertEqual({complete, 7}, json_scan(<>, S1)) end), %% an escape sequence straddling the chunk boundary is tracked correctly ?_test(begin Chunk1 = <<"{\"k\":\"ab\\">>, Chunk2 = <<"\"cd\"}">>, {incomplete, S2} = json_scan(Chunk1), Whole = <>, ?assertEqual({complete, byte_size(Whole)}, json_scan(Whole, S2)) end) ]. json_stream_decoder_test_() -> [ %% values split across feed/2 calls ?_test(begin D0 = json_stream_decoder(), {[#{<<"a">> := 1}], D1} = json_stream_feed(D0, <<"{\"a\":1} {\"b\":">>), {[#{<<"b">> := 2}], _D2} = json_stream_feed(D1, <<"2}">>) end), %% byte-at-a-time JSON feeding decodes every line ?_test(begin Doc = <<"{\"x\":1}\n{\"y\":[1,2,3]}\n{\"z\":\"hi\"}\n">>, {Vals, DLast} = lists:foldl( fun(B, {Acc, D}) -> {V, D2} = json_stream_feed(D, <>), {Acc ++ V, D2} end, {[], json_stream_decoder()}, binary_to_list(Doc)), {ok, []} = json_stream_eof(DLast), ?assertEqual([#{<<"x">> => 1}, #{<<"y">> => [1, 2, 3]}, #{<<"z">> => <<"hi">>}], Vals) end), %% a trailing bare scalar is only resolved at end-of-stream ?_test(begin D0 = json_stream_decoder(), {[], D1} = json_stream_feed(D0, <<" 42">>), ?assertEqual({ok, [42]}, json_stream_eof(D1)) end), %% trailing whitespace at EOF yields no extra value ?_test(begin D0 = json_stream_decoder(), {[#{<<"a">> := 1}], D1} = json_stream_feed(D0, <<"{\"a\":1}\n">>), ?assertEqual({ok, []}, json_stream_eof(D1)) end), %% decode options are threaded through to every decoded value ?_test(begin D0 = json_stream_decoder([{keys, atom}]), {[#{a := 1}], _D1} = json_stream_feed(D0, <<"{\"a\":1}">>) end), %% malformed trailing bytes surface as an error from json_stream_eof/1 ?_test(begin D0 = json_stream_decoder(), {[], D1} = json_stream_feed(D0, <<"{\"a\":">>), ?assertMatch({error, _}, json_stream_eof(D1)) end) ]. csv_stream_decoder_test_() -> [ %% rows split across feed/2 calls ?_test(begin D0 = csv_stream_decoder(), {[[<<"a">>, <<"b">>], [<<"1">>, <<"2">>]], D1} = csv_stream_feed(D0, <<"a,b\n1,2\n3,">>), {[[<<"3">>, <<"4">>]], _D2} = csv_stream_feed(D1, <<"4\n">>), ok end), %% byte-at-a-time feeding decodes every row ?_test(begin Doc = <<"a,b\n1,2\n3,4\n">>, {Rows, DLast} = lists:foldl( fun(B, {Acc, D}) -> {R, D2} = csv_stream_feed(D, <>), {Acc ++ R, D2} end, {[], csv_stream_decoder()}, binary_to_list(Doc)), ?assertEqual({ok, []}, csv_stream_eof(DLast)), ?assertEqual([[<<"a">>, <<"b">>], [<<"1">>, <<"2">>], [<<"3">>, <<"4">>]], Rows) end), %% a trailing row with no terminator is only resolved at end-of-stream ?_test(begin D0 = csv_stream_decoder(), {[[<<"a">>, <<"b">>]], D1} = csv_stream_feed(D0, <<"a,b\n1,2">>), ?assertEqual({ok, [[<<"1">>, <<"2">>]]}, csv_stream_eof(D1)) end), %% trailing newline at EOF yields no extra row ?_test(begin D0 = csv_stream_decoder(), {[[<<"a">>, <<"b">>]], D1} = csv_stream_feed(D0, <<"a,b\n">>), ?assertEqual({ok, []}, csv_stream_eof(D1)) end), %% blank lines are skipped, matching csv_decode/2 ?_test(begin D0 = csv_stream_decoder(), {Rows, D1} = csv_stream_feed(D0, <<"a,b\n\n1,2\n">>), ?assertEqual([[<<"a">>, <<"b">>], [<<"1">>, <<"2">>]], Rows), ?assertEqual({ok, []}, csv_stream_eof(D1)) end), %% a quoted field containing an embedded newline doesn't end the row ?_test(begin D0 = csv_stream_decoder(), {Rows, D1} = csv_stream_feed(D0, <<"a,\"b\nc\"\nd,e\n">>), ?assertEqual([[<<"a">>, <<"b\nc">>], [<<"d">>, <<"e">>]], Rows), ?assertEqual({ok, []}, csv_stream_eof(D1)) end), %% a quoted field containing a doubled (escaped) quote doesn't end the row ?_test(begin D0 = csv_stream_decoder(), {Rows, D1} = csv_stream_feed(D0, <<"a,\"b\"\"c\"\n">>), ?assertEqual([[<<"a">>, <<"b\"c">>]], Rows), ?assertEqual({ok, []}, csv_stream_eof(D1)) end), %% the headers option decodes subsequent rows as maps, with no row %% emitted for the header itself ?_test(begin D0 = csv_stream_decoder([headers]), {Rows, D1} = csv_stream_feed(D0, <<"a,b\n1,2\n3,4\n">>), ?assertEqual([#{<<"a">> => <<"1">>, <<"b">> => <<"2">>}, #{<<"a">> => <<"3">>, <<"b">> => <<"4">>}], Rows), ?assertEqual({ok, []}, csv_stream_eof(D1)) end), %% decode options are threaded through to every decoded row ?_test(begin D0 = csv_stream_decoder([headers, {keys, atom}]), {[#{a := <<"1">>, b := <<"2">>}], _D1} = csv_stream_feed(D0, <<"a,b\n1,2\n">>) end), %% malformed trailing bytes surface as an error from csv_stream_eof/1 ?_test(begin D0 = csv_stream_decoder(), {[], D1} = csv_stream_feed(D0, <<"\"unterminated">>), ?assertMatch({error, unterminated_quoted_field}, csv_stream_eof(D1)) end) ]. -endif.