Current section

Files

Jump to
grpc lib grpc message.ex
Raw

lib/grpc/message.ex

defmodule GRPC.Message do
use Bitwise, only_operators: true
@max_message_length 1 <<< (32 - 1)
alias GRPC.RPCError
@moduledoc false
# Transform data between encoded protobuf and HTTP/2 body of gRPC.
#
# gRPC body format is:
#
# # http://www.grpc.io/docs/guides/wire.html
# Delimited-Message -> Compressed-Flag Message-Length Message
# Compressed-Flag -> 0 / 1 # encoded as 1 byte unsigned integer
# Message-Length -> {length of Message} # encoded as 4 byte unsigned integer
# Message -> *{binary octet}
@doc """
Transform protobuf data to gRPC body
## Examples
iex> message = <<1, 2, 3, 4, 5, 6, 7, 8>>
iex> GRPC.Message.to_data(message)
{:ok, <<0, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7, 8>>, 13}
iex> message = <<1, 2, 3, 4, 5, 6, 7, 8, 9>>
iex> GRPC.Message.to_data(message, %{max_message_length: 8})
{:error, "Encoded message is too large (9 bytes)"}
"""
@spec to_data(iodata, map | Keyword.t()) ::
{:ok, binary, non_neg_integer} | {:error, String.t()}
def to_data(message, opts \\ %{}) do
compressor = opts[:compressor]
{compress_flag, message} =
if compressor do
{1, compressor.compress(message)}
else
{0, message}
end
length = byte_size(message)
max_length = opts[:max_message_length] || @max_message_length
if length > max_length do
{:error, "Encoded message is too large (#{length} bytes)"}
else
result = <<compress_flag, length::size(4)-unit(8), message::binary>>
{:ok, result, length + 5}
end
end
@doc """
Transform gRPC body to protobuf data
## Examples
iex> GRPC.Message.from_data(<<0, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7, 8>>)
<<1, 2, 3, 4, 5, 6, 7, 8>>
"""
@spec from_data(binary) :: binary
def from_data(data) do
<<_flag::unsigned-integer-size(8), _length::bytes-size(4), message::binary>> = data
message
end
@doc """
Transform gRPC body to protobuf data with compressing
## Examples
iex> GRPC.Message.from_data(%{compressor: nil}, <<0, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7, 8>>)
{:ok, <<1, 2, 3, 4, 5, 6, 7, 8>>}
"""
@spec from_data(map, binary) :: {:ok, binary} | {:error, GRPC.RPCError.t()}
def from_data(%{compressor: nil}, data) do
case data do
<<0, _length::bytes-size(4), message::binary>> ->
{:ok, message}
<<1, _length::bytes-size(4), _::binary>> ->
{:error,
RPCError.exception(
status: :internal,
message: "Compressed flag is set, but not specified in headers."
)}
_ ->
{:error, RPCError.exception(status: :invalid_argument, message: "Message is malformed.")}
end
end
def from_data(%{compressor: compressor}, data) do
case data do
<<1, _length::bytes-size(4), message::binary>> ->
{:ok, compressor.decompress(message)}
<<0, _length::bytes-size(4), message::binary>> ->
{:ok, message}
_ ->
{:error, RPCError.exception(status: :invalid_argument, message: "Message is malformed.")}
end
end
def from_frame(bin), do: from_frame(bin, [])
def from_frame(<<>>, acc), do: Enum.reverse(acc)
def from_frame(
<<_flag::unsigned-integer-size(8), length::32, msg::bytes-size(length), rest::binary>>,
acc
) do
from_frame(rest, [msg | acc])
end
def complete?(
<<_flag::unsigned-integer-size(8), length::unsigned-integer-size(32), message::binary>>
) do
length == byte_size(message)
end
def complete?(_), do: false
def message_length(data) do
<<_flag::unsigned-integer-size(8), length::unsigned-integer-size(32), _message::binary>> =
data
length
end
@doc """
Get message data from data buffer
## Examples
iex> GRPC.Message.get_message(<<0, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7, 8, 0, 0, 0>>)
{{0, <<1, 2, 3, 4, 5, 6, 7, 8>>}, <<0, 0, 0>>}
iex> GRPC.Message.get_message(<<1, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7, 8>>)
{{1, <<1, 2, 3, 4, 5, 6, 7, 8>>}, <<>>}
iex> GRPC.Message.get_message(<<0, 0, 0, 0, 8, 1, 2, 3, 4, 5, 6, 7>>)
false
"""
def get_message(
<<flag::unsigned-integer-size(8), length::unsigned-integer-size(32),
message::bytes-size(length), rest::binary>>
) do
{{flag, message}, rest}
end
def get_message(_) do
false
end
end