Erlang/Elixir gRPC 与 Protobuf：从编码原理到 grpcbox 实战

syntax = "proto3";

package user.v1;

import "google/protobuf/timestamp.proto";

service UserService {
  rpc GetUser(GetUserRequest) returns (User);
  rpc ListUsers(ListUsersRequest) returns (stream User);
  rpc UploadUsers(stream User) returns (UploadSummary);
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message GetUserRequest { int64 user_id = 1; }

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  repeated string roles = 4;
  google.protobuf.Timestamp created_at = 5;
  map<string, string> labels = 6;
  optional string nickname = 7;
}

message UploadSummary { int32 accepted = 1; int32 rejected = 2; }

%% rebar.config
{plugins, [rebar3_gpb_plugin]}.

{gpb_opts, [
    {i, "priv/proto"},                  %% 导入路径
    {module_name_suffix, "_pb"},        %% 生成 user_pb 而非 user
    {strings, as_binary},               %% string 映射为 binary（推荐）
    {maps, true},                       %% 消息映射为 maps 而非 record
    {type_specs, true}                  %% 生成 -spec 供 Dialyzer 使用
]}.

Msg = #{id => 42, name => <<"Alice">>, email => <<"a@b.c">>, roles => [<<"admin">>]},
Bin = user_pb:encode_msg(Msg, 'user.v1.User'),
Decoded = user_pb:decode_msg(iolist_to_binary(Bin), 'user.v1.User').

defmodule MyApp.Proto.UserMapper do
  alias User.V1.User

  def to_proto(%MyApp.Accounts.User{} = u) do
    %User{id: u.id, name: u.name, email: u.email, roles: u.roles,
          created_at: to_ts(u.inserted_at), labels: u.labels || %{}}
  end

  defp to_ts(%DateTime{} = dt),
    do: Google.Protobuf.Timestamp.new(seconds: DateTime.to_unix(dt))
end

-module(user_service).
-behaviour(grpcbox_service).

-export([get_user/2, list_users/2]).

%% 一元 RPC：请求 -> 响应
get_user(GetUserReq, Stream) ->
    #{user_id := Id} = GetUserReq,
    case user_db:find(Id) of
        {ok, User} -> {ok, to_pb(User), Stream};
        error      -> {grpcbox:status(not_found, <<"user not found">>), Stream}
    end.

%% 服务端流式：一次请求，多次响应
list_users(#{page_size := Size}, Stream) ->
    Users = user_db:list(Size),
    lists:foldl(fun(U, S) -> grpcbox:send_response(to_pb(U), S) end, Stream, Users),
    {ok, Stream}.

%% sys.config
{grpcbox, [
    {servers, [
        #{grpc_opts => #{
              service_protos => [user_pb],
              services => #{'user.v1.UserService' => user_service}
          },
          listen_opts => #{port => 8080, ip => {0, 0, 0, 0}},
          transport_opts => #{max_connections => 10000, idle_timeout => 30000}}
    ]}
]}.

upload_users(Stream, _InitState) ->
    upload_loop(Stream, 0, 0).

upload_loop(Stream, Accepted, Rejected) ->
    case grpcbox:recv_request(Stream) of
        {ok, User} ->
            case validate(User) of
                ok      -> upload_loop(Stream, Accepted + 1, Rejected);
                invalid -> upload_loop(Stream, Accepted, Rejected + 1)
            end;
        eof ->
            {ok, #{accepted => Accepted, rejected => Rejected}, Stream}
    end.

{ok, Channel} = grpcbox_channel:start_link(user_channel, #{
    endpoints => [{http, "user-svc.internal", 8080, []}],
    pool_size => 4
}),

{ok, Reply} = grpcbox_client:unary(
    Channel, <<"/user.v1.UserService/GetUser">>, #{user_id => 42},
    #{timeout => 5000}),
Name = maps:get(name, Reply).

{ok, Stream} = grpcbox_client:stream(
    Channel, <<"/user.v1.UserService/ListUsers">>, #{page_size => 100},
    #{timeout => 30000}),

collect(Stream).
collect(Stream) ->
    case grpcbox_client:recv(Stream) of
        {ok, Msg} -> [Msg | collect(Stream)];
        eof       -> []
    end.

%% 客户端发送元数据
MD = #{<<"authorization">> => <<"Bearer ", Token/binary>>,
       <<"x-request-id">> => RequestId},
grpcbox_client:unary(Channel, Path, Req, #{metadata => MD, timeout => 5000}).

%% 服务端读取元数据
get_user(Req, Stream) ->
    case maps:get(<<"authorization">>, grpcbox:metadata(Stream), undefined) of
        undefined -> {grpcbox:status(unauthenticated, <<"missing token">>), Stream};
        Token     -> handle_with_token(Token, Req, Stream)
    end.

-module(auth_interceptor).
-behaviour(grpcbox_interceptor).

-export([unary/5, stream/4]).

unary(_Path, Req, Stream, MD, _Opts) ->
    case verify(MD) of
        {ok, Claims} -> grpcbox:continue(Req, Stream, MD#{claims => Claims});
        {error, R}   -> grpcbox:abort(Stream, #{code => unauthenticated, message => R})
    end.

stream(_Path, Stream, MD, _Opts) ->
    case verify(MD) of
        {ok, _}    -> grpcbox:continue(Stream, MD);
        {error, R} -> grpcbox:abort(Stream, #{code => unauthenticated, message => R})
    end.

{ok, Channel} = grpcbox_channel:start_link(user_channel, #{
    endpoints => [{http, "user-svc-1.internal", 8080, []},
                  {http, "user-svc-2.internal", 8080, []}],
    pool_size => 8,
    load_balancing => round_robin
}),

stream(Path, Stream, Metadata, _Opts) ->
    ParentCtx = otel_propagator:text_map_extract(Metadata),
    Ctx = otel_tracer:start_span(Path, #{parent => ParentCtx}),
    grpcbox:continue(Stream, Metadata#{otel_ctx => Ctx}).