New pb (#11)

bieniusa · web-flow · commit c54faa83d069 · 2019-07-05T10:00:45.000+09:00
* Adapting to interface changes of new pb codec
* Cleaning up dependencies and lint
diff --git a/rebar.config b/rebar.config
@@ -43,12 +43,11 @@
                    #{regex => "^[a-z]([a-z0-9]*_?)*(_SUITE)?$",
                      ignore => []}
                  },
-                 % Can be added back if antidote_crdt_bcounter:localPermissions is renamed
-                 %{
-                 %  elvis_style,
-                 %  function_naming_convention,
-                 %  #{regex => "^([a-z][a-z0-9]*_?)*$"}
-                 %},
+                 {
+                  elvis_style,
+                  function_naming_convention,
+                  #{regex => "^([a-z][a-z0-9]*_?)*$"}
+                 },
                  {elvis_style, state_record_and_type},
                  {elvis_style, no_spec_with_records}
                 ]
diff --git a/rebar.lock b/rebar.lock
@@ -1,6 +1,6 @@
 {"1.1.0",
-[{<<"antidote_pb_codec">>,{pkg,<<"antidote_pb_codec">>,<<"0.0.5">>},0}]}.
+[{<<"antidote_pb_codec">>,{pkg,<<"antidote_pb_codec">>,<<"0.1.0">>},0}]}.
 [
 {pkg_hash,[
- {<<"antidote_pb_codec">>, <<"139F291D7E4971DE3920E51D09BD10931AF957656ED6F7D9935B3EA059D88167">>}]}
+ {<<"antidote_pb_codec">>, <<"D8CC2D69BD25B3961ADECE20954FCB66719AA1DF1D027CF58389C8B7B6EFC739">>}]}
 ].
diff --git a/src/antidotec_pb.erl b/src/antidotec_pb.erl
@@ -32,61 +32,68 @@
 
 -define(TIMEOUT, 10000).
 
--spec start_transaction(Pid::term(), TimeStamp::term())
+-spec start_transaction(Pid::pid(), TimeStamp::binary() | ignore)
         -> {ok, {interactive, term()} | {static, {term(), term()}}} | {error, term()}.
 start_transaction(Pid, TimeStamp) ->
     start_transaction(Pid, TimeStamp, []).
 
--spec start_transaction(Pid::term(), TimeStamp::term(), TxnProperties::term())
-        -> {ok, {interactive, term()} | {static, {term(), term()}}} | {error, term()}.
+-spec start_transaction(Pid::pid(), TimeStamp::binary() | ignore, TxnProperties::term())
+        -> {ok, {interactive, binary()} | {static, {binary(), term()}}} | {error, term()}.
 start_transaction(Pid, TimeStamp, TxnProperties) ->
+    EncTimestamp = case TimeStamp of
+                ignore -> term_to_binary(ignore);
+                Binary -> Binary
+            end,
     case is_static(TxnProperties) of
         true ->
-            {ok, {static, {TimeStamp, TxnProperties}}};
+            {ok, {static, {EncTimestamp, TxnProperties}}};
         false ->
-            EncMsg = antidote_pb_codec:encode(start_transaction,
-                                              {TimeStamp, TxnProperties}),
+            EncMsg = antidote_pb_codec:encode_request({start_transaction, EncTimestamp, TxnProperties}),
             Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
             case Result of
                 {error, timeout} ->
                     {error, timeout};
                 _ ->
                     case antidote_pb_codec:decode_response(Result) of
-                        {start_transaction, TxId} ->
+                        {start_transaction_response, {ok, TxId}} ->
                             {ok, {interactive, TxId}};
-                        {error, Reason} ->
+                        {start_transaction_response, {error, Reason}} ->
+                            {error, Reason};
+                        {error_response, Reason} ->
                             {error, Reason};
                         Other ->
                             {error, Other}
                     end
             end
     end.
 
--spec abort_transaction(Pid::term(), TxId::term()) -> ok | {error, term()}.
+-spec abort_transaction(Pid::pid(), {interactive, TxId::binary()}) -> ok | {error, term()}.
 abort_transaction(Pid, {interactive, TxId}) ->
-    EncMsg = antidote_pb_codec:encode(abort_transaction, TxId),
+    EncMsg = antidote_pb_codec:encode_request({abort_transaction, TxId}),
     Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec:decode_response(Result) of
-                {opresponse, ok} -> ok;
-                {error, Reason} -> {error, Reason};
+                {operation_response, ok} -> ok;
+                {operation_response, {error, Reason}} -> {error, Reason};
+                {error_response, Reason} -> {error, Reason};
                 Other -> {error, Other}
             end
     end.
 
--spec commit_transaction(Pid::term(), TxId::{interactive, term()} | {static, term()}) ->
-                                {ok, term()} | {error, term()}.
+-spec commit_transaction(Pid::pid(), TxId::{interactive, binary()} | {static, binary()}) ->
+                                {ok, binary()} | {error, term()}.
 commit_transaction(Pid, {interactive, TxId}) ->
-    EncMsg = antidote_pb_codec:encode(commit_transaction, TxId),
+    EncMsg = antidote_pb_codec:encode_request({commit_transaction, TxId}),
     Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec:decode_response(Result) of
-                {commit_transaction, CommitTimeStamp} -> {ok, CommitTimeStamp};
-                {error, Reason} -> {error, Reason};
+                {commit_response, {ok, CommitTimeStamp}} -> {ok, CommitTimeStamp};
+                {commit_response, {error, Reason}} -> {error, Reason};
+                {error_response, Reason} -> {error, Reason};
                 Other -> {error, Other}
             end
     end;
@@ -96,38 +103,40 @@ commit_transaction(Pid, {static, _TxId}) ->
              {ok, CommitTime}
     end.
 
--spec update_objects(Pid::term(), Updates::[{term(), term(), term()}], TxId::term()) -> ok | {error, term()}.
+-spec update_objects(Pid::pid(), Updates::[{term(), term(), term()}], {interactive | static, TxId::binary()}) -> ok | {error, term()}.
 update_objects(Pid, Updates, {interactive, TxId}) ->
-    EncMsg = antidote_pb_codec: encode(update_objects, {Updates, TxId}),
+    EncMsg = antidote_pb_codec: encode_request({update_objects, Updates, TxId}),
     Result = antidotec_pb_socket: call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec: decode_response(Result) of
-                {opresponse, ok} -> ok;
-                {error, Reason} -> {error, Reason};
+                {operation_response, ok} -> ok;
+                {operation_response, {error, Reason}} -> {error, Reason};
+                {error_response, Reason} -> {error, Reason};
                 Other -> {error, Other}
             end
     end;
 
 update_objects(Pid, Updates, {static, TxId}) ->
     {Clock, Properties} = TxId,
-    EncMsg = antidote_pb_codec:encode(static_update_objects,
-                                      {Clock, Properties, Updates}),
+    EncMsg = antidote_pb_codec:encode_request({static_update_objects, Clock, Properties, Updates}),
     Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec:decode_response(Result) of
-                {commit_transaction, CommitTimeStamp} ->
+                {commit_response, {ok, CommitTimeStamp}} ->
                     antidotec_pb_socket:store_commit_time(Pid, CommitTimeStamp),
                     ok;
-                {error, Reason} -> {error, Reason};
+                {commit_response, {error, Reason}} ->
+                    {error, Reason};
+                {error_response, Reason} -> {error, Reason};
                 Other -> {error, Other}
             end
     end.
 
--spec read_objects(Pid::term(), Objects::[term()], TxId::term()) -> {ok, [term()]}  | {error, term()}.
+-spec read_objects(Pid::pid(), Objects::[term()], {interactive | static, TxId::binary()}) -> {ok, [term()]}  | {error, term()}.
 read_objects(Pid, Objects, Transaction) ->
     case read_values(Pid, Objects, Transaction) of
         {ok, Values} ->
@@ -141,33 +150,34 @@ read_objects(Pid, Objects, Transaction) ->
             Other
     end.
 
--spec read_values(Pid::term(), Objects::[term()], TxId::term()) -> {ok, [term()]}  | {error, term()}.
+-spec read_values(Pid::pid(), Objects::[term()], {interactive | static, TxId::binary()}) -> {ok, [term()]}  | {error, term()}.
 read_values(Pid, Objects, {interactive, TxId}) ->
-    EncMsg = antidote_pb_codec:encode(read_objects, {Objects, TxId}),
+    EncMsg = antidote_pb_codec:encode_request({read_objects, Objects, TxId}),
     Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec:decode_response(Result) of
-                {read_objects, Values} ->
+                {read_objects_response, {ok, Values}} ->
                     {ok, Values};
-                {error, Reason} -> {error, Reason};
+                {read_objects_response, {error, Reason}} ->
+                    {error, Reason};
+                {error_response, Reason} -> {error, Reason};
                 Other -> {error, Other}
             end
     end;
 read_values(Pid, Objects, {static, TxId}) ->
     {Clock, Properties} = TxId,
-    EncMsg = antidote_pb_codec:encode(static_read_objects,
-                                      {Clock, Properties, Objects}),
+    EncMsg = antidote_pb_codec:encode_request({static_read_objects, Clock, Properties, Objects}),
     Result = antidotec_pb_socket:call_infinity(Pid, {req, EncMsg, ?TIMEOUT}),
     case Result of
         {error, timeout} -> {error, timeout};
         _ ->
             case antidote_pb_codec:decode_response(Result) of
-                {static_read_objects_resp, Values, CommitTimeStamp} ->
+                {static_read_objects_response, {Values, CommitTimeStamp}} ->
                     antidotec_pb_socket:store_commit_time(Pid, CommitTimeStamp),
                     {ok, Values};
-                {error, Reason} -> {error, Reason}
+                {error_response, Reason} -> {error, Reason}
             end
     end.
 
diff --git a/src/antidotec_pb_socket.erl b/src/antidotec_pb_socket.erl
@@ -30,10 +30,8 @@
 -type address() :: string() | atom() | inet:ip_address().
 %% The TCP port number of the Riak node's Protocol Buffers interface
 -type portnum() :: non_neg_integer().
--type msg_id() :: non_neg_integer().
--type rpb_req() :: {tunneled, msg_id(), binary()} | atom() | tuple().
 
--record(request, {ref :: reference(), msg :: rpb_req(), from, timeout :: timeout(),
+-record(request, {ref :: reference(), from, timeout :: timeout(),
                   tref :: reference() | undefined }).
 
 -record(state, {
@@ -111,7 +109,17 @@ get_last_commit_time(Pid) ->
 
 %% @private
 handle_call({req, Msg, Timeout}, From, State) ->
-    {noreply, send_request(new_request(Msg, From, Timeout), State)};
+    Ref = make_ref(),
+    Req = #request{ref = Ref, from = From, timeout = Timeout,
+             tref = create_req_timer(Timeout, Ref)},
+    NewState = case gen_tcp:send(State#state.sock, Msg) of
+        ok ->
+            maybe_reply({noreply,  State#state{active = Req}});
+        {error, Reason} ->
+            logger:warning("Socket error while sending request: ~p.", [Reason]),
+            gen_tcp:close(State#state.sock)
+    end,
+    {noreply, NewState};
 
 handle_call({store_commit_time, TimeStamp}, _From, State) ->
     {reply, ok, State#state{last_commit_time = TimeStamp}};
@@ -126,10 +134,8 @@ handle_call(stop, _From, State) ->
 %% @private
 %% @todo handle timeout
 handle_info({_Proto, Sock, Data}, State=#state{active = (Active = #request{})}) ->
-    <<MsgCode:8, MsgData/binary>> = Data,
-    Response = antidote_pb_codec:decode_msg(MsgCode, MsgData),
     cancel_req_timer(Active#request.tref),
-    _ = send_caller(Response, Active),
+    _ = send_caller(Data, Active),
     NewState = State#state{active = undefined},
     ok = inet:setopts(Sock, [{active, once}]),
     {noreply, NewState};
@@ -190,13 +196,6 @@ disconnect(State) ->
     NewState = State#state{sock = undefined, active = undefined},
     {stop, disconnected, NewState}.
 
-
-%% @private
-new_request(Msg, From, Timeout) ->
-    Ref = make_ref(),
-    #request{ref = Ref, msg = Msg, from = From, timeout = Timeout,
-             tref = create_req_timer(Timeout, Ref)}.
-
 %% @private
 %% Create a request timer if desired, otherwise return undefined.
 create_req_timer(infinity, _Ref) ->
@@ -206,22 +205,6 @@ create_req_timer(undefined, _Ref) ->
 create_req_timer(Msecs, Ref) ->
     erlang:send_after(Msecs, self(), {req_timeout, Ref}).
 
-%% Send a request to the server and prepare the state for the response
-%% @private
-send_request(Request0, State) when State#state.active =:= undefined  ->
-    {Request, Pkt} = encode_request_message(Request0),
-    case gen_tcp:send(State#state.sock, Pkt) of
-        ok ->
-            maybe_reply({noreply,  State#state{active = Request}});
-        {error, Reason} ->
-            logger:warning("Socket error while sending request: ~p.", [Reason]),
-            gen_tcp:close(State#state.sock)
-    end.
-
-%% Unencoded Request (the normal PB client path)
-encode_request_message(#request{msg=Msg}=Req) ->
-    EncMsg = antidote_pb_codec:encode_msg(Msg),
-    {Req, EncMsg}.
 
 %% maybe_reply({reply, Reply, State = #state{active = Request}}) ->
 %%   NewRequest = send_caller(Reply, Request),

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{"1.1.0",`
`2`		`-[{<<"antidote_pb_codec">>,{pkg,<<"antidote_pb_codec">>,<<"0.0.5">>},0}]}.`
	`2`	`+[{<<"antidote_pb_codec">>,{pkg,<<"antidote_pb_codec">>,<<"0.1.0">>},0}]}.`
`3`	`3`	`[`
`4`	`4`	`{pkg_hash,[`
`5`		`- {<<"antidote_pb_codec">>, <<"139F291D7E4971DE3920E51D09BD10931AF957656ED6F7D9935B3EA059D88167">>}]}`
	`5`	`+ {<<"antidote_pb_codec">>, <<"D8CC2D69BD25B3961ADECE20954FCB66719AA1DF1D027CF58389C8B7B6EFC739">>}]}`
`6`	`6`	`].`