# Copyright 2016 Apcera Inc. All rights reserved. defmodule Nats.Parser do defp parse_err(mesg) do {:error, "NATS: parsing error: #{mesg}", nil} end @min_lookahead 4 defp init_state(rest \\ <<>>, func \\ &verb/2, state \\ nil), do: {func, rest, state} defp cont(func, state, how_many \\ @min_lookahead, rest \\ <<>>), do: {:cont, how_many, init_state(rest, func, state)} def parse(string), do: parse(init_state, string) def parse(nil, string), do: parse(init_state, string) def parse({ func, <<>>, state}, string), do: func.(string, state) def parse({ func, buff, state}, string), do: func.(<>, state) defp verb(<<"MSG ", rest :: bits>>, _), do: args(rest, [:msg]) defp verb(<<"PUB ", rest :: bits>>, _), do: args(rest, [:pub]) defp verb(<<"SUB ", rest :: bits>>, _), do: args(rest, [:sub]) defp verb(<<"UNSUB ", rest :: bits>>, _), do: args(rest, [:unsub]) defp verb(<<"+OK\r\n", rest :: bits>>, _), do: done(rest, :ok) defp verb(<<"PING\r\n", rest :: bits>>, _), do: done(rest, :ping) defp verb(<<"PONG\r\n", rest :: bits>>, _), do: done(rest, :pong) defp verb(<<"CONNECT ", rest :: bits>>, _), do: json(rest, :connect, <<>>) defp verb(<<"INFO ", rest :: bits>>, _), do: json(rest, :info, <<>>) defp verb(<<"-ERR ", rest :: bits>>, _), do: err(rest, <<>>) defp verb(other, state) when byte_size(other) < @min_lookahead, # FIXME: jam: fail faster... do: cont(&verb/2, state, @min_lookahead, other) defp verb(buff, _) do read = binary_part(buff, 0, min(16, byte_size(buff))) parse_err("invalid protocol bytes: #{inspect(read)}") end defp err(<<>>, acc), do: cont(&err/2, acc) defp err(what = <>, acc), do: done(what, [acc, :err]) defp err(what = <>, acc), do: done(what, [acc, :err]) defp err(<>, acc), do: err(rest, <>) defp json(<<>>, acc), do: cont(&json/2, acc) defp json(what = <>, verb, acc), do: done(what, [acc, verb]) defp json(what = <>, verb, acc), do: done(what, [acc, verb]) defp json(<>, verb, acc), do: json(rest, verb, <>) defp args(<<>>, argv), do: cont(&args/2, argv) defp args(what = <>, argv), do: done(what, argv) defp args(what = <>, argv), do: done(what, argv) defp args(<>, argv), do: args(rest, argv) defp args(<>, argv), do: args(rest, argv) defp args(<>, argv), do: args_arg(rest, <>, argv) # broken out of the above (and below) for continutions defp args_arg(buf, so_far, argv) do case arg(buf, so_far) do {:ok, res, rest} -> args(rest, [res|argv]) {:cont, sofar} -> cont(&(args_arg(&1, &2, argv)), sofar) other -> other end end defp arg(<<>>, acc), do: {:cont, acc} defp arg(<>, acc), do: {:ok, acc, rest} defp arg(<>, acc), do: {:ok, acc, rest} defp arg(what = <>, acc), do: {:ok, acc, what} defp arg(what = <>, acc), do: {:ok, acc, what} defp arg(<>, acc), do: arg(rest, <>) # We're at the end of the body defp body(<<"\r\n", rest::bits>>, 0, verb, acc), do: {:ok, put_elem(verb, tuple_size(verb) - 1, acc), rest, init_state} # We're at the end of the body, but its malformed (missing `\\r\\n`) defp body(_rest, 0, verb, _acc), do: parse_err("malformed body trailer for: #{inspect(verb)}") defp body(<<>>, nleft, verb, acc), do: cont(&body_cont/2, {nleft, verb, acc}) # "We have N more bytes to read" defp body(rest, nleft, verb, acc) do rest_size = byte_size(rest) to_read = min(nleft, rest_size) # IO.puts("body -> #{inspect(nleft)} rest=#{inspect(rest)} acc=#{inspect(acc)}") <> = rest body(remainder, nleft - to_read, verb, <>) end defp body_cont(buff, {nleft, verb, acc}), do: body(buff, nleft, verb, acc) defp done(rest, verb) when is_atom(verb) do {:ok, {verb}, rest, nil} end defp done(buff, argv) when byte_size(buff) < 2, do: cont(&done/2, argv, buff) defp done(<<"\r\n", rest :: bits>>, argv) do verb = done1(List.to_tuple(Enum.reverse(argv))) case verb do {:body, want, real_verb} -> body(rest, want, real_verb, <<>>) {:error, _, _} -> verb other -> {:ok, other, rest, init_state(rest)} end end defp done(<>, _), do: parse_err("invalid trailer `#{c1}#{c2}`") defp parse_json(verb, json_str) do case :json_lexer.string(to_char_list(json_str)) do {:ok, tokens, _} -> pres = :json_parser.parse(tokens) case pres do {:ok, json } when is_map(json) -> {verb, json} {:ok, _ } -> parse_err("not a json object in #{verb}: #{inspect(json_str)}") {:error, {_, what, mesg}} -> parse_err("invalid json in #{verb} #{what}: #{mesg}: #{inspect(json_str)}") other -> parse_err("unexpected json parser result for #{verb}: #{inspect(other)}: #{inspect(json_str)}") end {:eof, _} -> parse_err("json not complete in #{verb}: #{inspect(json_str)}") {:error, {_, why, mesg}} -> parse_err("invalid json tokens in #{verb}: #{why}: #{mesg}") # safe programming ;-) other -> parse_err("unexpected json lexer result for json in #{verb}: #{inspect(other)}: #{inspect(json_str)}") end end defp done1(w = {:err, _}), do: w defp done1({:info, json}), do: parse_json(:info, json) defp done1({:connect, json}), do: parse_json(:connect, json) defp done1({:unsub, sid}), do: {:unsub, sid, nil} defp done1({:unsub, sid, maxs}) when is_binary(maxs), do: done1({:unsub, sid, parse_int(maxs)}) defp done1({:unsub, sid, maxs}) when maxs == nil or is_integer(maxs), do: {:unsub, sid, maxs} defp done1(w = {:sub, _sub, _q, _sid}), do: w defp done1({:sub, sub, sid}), do: {:sub, sub, nil, sid} defp done1({:pub, sub, size}), do: done1({:pub, sub, nil, parse_int(size)}) defp done1({:pub, sub, ret, size}) when is_binary(size), do: done1({:pub, sub, ret, parse_int(size)}) defp done1(verb = {:pub, _sub, _ret, size}) when is_integer(size), do: {:body, size, verb} defp done1({:msg, sub, sid, size}) when is_binary(size), do: done1({:msg, sub, sid, nil, parse_int(size)}) defp done1({:msg, sub, sid, ret, size}) when is_binary(size), do: done1({:msg, sub, sid, ret, parse_int(size)}) defp done1(verb = {:msg, _sub, _sid, _ret, size}) when is_integer(size), do: {:body, size, verb} defp done1(verb) do why = "" # total hack to fix up things so there is ONE error that makes sense v = elem(verb, 0) if v == :msg || v == :pub do case elem(verb, tuple_size(verb) - 1) do {:error, reason} -> why = ": #{reason}" size -> size end end # IO.puts("done1 catchall: #{inspect(verb)}: #{why}") parse_err("invalid arguments to #{elem(verb,0)}#{why}") end defp parse_int(what), do: parse_int1(what, Integer.parse(what, 10)) defp parse_int1(_orig, {result, <<>>}) when result >= 0, do: result defp parse_int1(orig, _), do: {:error, "invalid integer: #{orig}"} def to_json(false) do <<"false">> end def to_json(true) do <<"true">> end def to_json(nil) do <<"null">> end def to_json(n) when is_number(n) do <<"#{n}">> end def to_json(str) when is_binary(str) do <> <> str <> <> end def to_json(map) when is_map(map) do <> <> Enum.join(Enum.map(map, fn({k,v}) -> member_pair(k,v) end), ", ") <> <> end def to_json(array) when is_list(array) do <> <> Enum.join(Enum.map(array, fn(x) -> to_json(x) end), ", ") <> <> end defp member_pair(k,v) when is_binary(k) do to_json(k) <> <<": ">> <> to_json(v) end @endverb "\r\n" def encode(mesg) do [encode1(mesg), @endverb] end defp encode1({:ok}) do "+OK" end defp encode1({:ping}) do "PING" end defp encode1({:pong}) do "PONG" end defp encode1({:err, msg}) do ["-ERR ", msg] end defp encode1({:info, json}) do ["INFO ", to_json(json)] end defp encode1({:connect, json}) do ["CONNECT ", to_json(json)] end defp encode1({:msg, sub, sid, nil, what}) do ["MSG ", sub, " ", sid, " ", to_string(byte_size(what)), @endverb, what] end defp encode1({:msg, sub, sid, ret, what}) do ["MSG ", sub, " ", sid, " ", ret, " ", to_string(byte_size(what)), @endverb, what] end defp encode1({:pub, sub, nil, what}) do ["PUB ", sub, " ", to_string(byte_size(what)), @endverb, what] end defp encode1({:pub, sub, reply, what}) do ["PUB ", sub, " ", reply, " ", to_string(byte_size(what)), @endverb, what] end defp encode1({:sub, subject, nil, sid}) do ["SUB ", subject, " ", sid] end defp encode1({:sub, subject, queue, sid}) do ["SUB ", subject, " ", queue, " ", sid] end defp encode1({:unsub, sid, nil}), do: ["UNSUB ", sid] defp encode1({:unsub, sid, afterReceiving}), do: ["UNSUB ", sid, " ", to_string(afterReceiving)] end