diff --git a/apps/efka/include/message_pb.hrl b/apps/efka/include/message_pb.hrl index 30cd5ce..15dc9c2 100644 --- a/apps/efka/include/message_pb.hrl +++ b/apps/efka/include/message_pb.hrl @@ -65,7 +65,8 @@ -define('SERVICE_PARAMS_PB_H', true). -record(service_params, {service_id = <<>> :: unicode:chardata() | undefined, % = 1, optional - params = <<>> :: unicode:chardata() | undefined % = 2, optional + params = <<>> :: unicode:chardata() | undefined, % = 2, optional + timeout = 0 :: non_neg_integer() | undefined % = 3, optional, 32 bits }). -endif. @@ -73,7 +74,8 @@ -define('SERVICE_METRICS_PB_H', true). -record(service_metrics, {service_id = <<>> :: unicode:chardata() | undefined, % = 1, optional - metrics = <<>> :: unicode:chardata() | undefined % = 2, optional + metrics = <<>> :: unicode:chardata() | undefined, % = 2, optional + timeout = 0 :: non_neg_integer() | undefined % = 3, optional, 32 bits }). -endif. diff --git a/apps/efka/src/efka_agent.erl b/apps/efka/src/efka_agent.erl index 8eb771b..3086500 100644 --- a/apps/efka/src/efka_agent.erl +++ b/apps/efka/src/efka_agent.erl @@ -296,7 +296,7 @@ handle_info({server_push_message, PacketId, <>}, State = #state{transport_pid = TransportPid, inflight = Inflight}) -> - #service_params{service_id = ServiceId, params = Params} = message_pb:decode_msg(ParamsBin, service_params), + #service_params{service_id = ServiceId, params = Params, timeout = Timeout} = message_pb:decode_msg(ParamsBin, service_params), case efka_micro_service:get_pid(ServiceId) of undefined -> @@ -309,13 +309,17 @@ handle_info({server_push_message, PacketId, < Ref = make_ref(), efka_micro_service:push_params(ServicePid, Ref, Params), + + %% 处理超时逻辑 + erlang:start_timer(Timeout * 1000, self(), {request_timeout, Ref}), + {noreply, State#state{inflight = maps:put(Ref, PacketId, Inflight)}} end, {noreply, State}; %% 采集项目 handle_info({server_push_message, PacketId, <>}, State = #state{transport_pid = TransportPid, inflight = Inflight}) -> - #service_metrics{service_id = ServiceId, metrics = Metrics} = message_pb:decode_msg(MetricsBin, service_metrics), + #service_metrics{service_id = ServiceId, metrics = Metrics, timeout = Timeout} = message_pb:decode_msg(MetricsBin, service_metrics), case efka_micro_service:get_pid(ServiceId) of undefined -> @@ -332,25 +336,31 @@ handle_info({server_push_message, PacketId, < +handle_info({ems_reply, Ref, EmsReply}, State = #state{inflight = Inflight}) -> case maps:take(Ref, Inflight) of error -> {noreply, State}; {PacketId, NInflight} -> - case Reply of - ok -> - Reply = #efka_response { - code = 1, - message = <<"">> - }, - efka_transport:response(TransportPid, PacketId, message_pb:encode_msg(Reply)); - {error, Reason} -> - Reply = #efka_response{ - code = 0, - message = Reason - }, - efka_transport:response(TransportPid, PacketId, message_pb:encode_msg(Reply)) - end, + Reply = case EmsReply of + ok -> + #efka_response{code = 1, message = <<"">>}; + {error, Reason} -> + #efka_response{code = 0, message = Reason} + end, + safe_response(PacketId, message_pb:encode_msg(Reply), State), + + {noreply, State#state{inflight = NInflight}} + end; + +%% 请求超时逻辑处理 +handle_info({timeout, _, {request_timeout, Ref}}, State = #state{inflight = Inflight}) -> + case maps:take(Ref, Inflight) of + error -> + {noreply, State}; + {PacketId, NInflight} -> + Reply = #efka_response{code = 0, message = <<"reqeust timeout">>}, + safe_response(PacketId, message_pb:encode_msg(Reply), State), + {noreply, State#state{inflight = NInflight}} end; @@ -413,4 +423,16 @@ code_change(_OldVsn, State = #state{}, _Extra) -> %%%=================================================================== %%% Internal functions -%%%=================================================================== \ No newline at end of file +%%%=================================================================== + +%% 安全回复 +-spec safe_response(PacketId :: integer(), Reply :: binary(), State :: #state{}) -> no_return(). +safe_response(PacketId, Reply, #state{status = ?STATE_ACTIVATED, transport_pid = TransportPid}) when is_integer(PacketId), is_binary(Reply), is_pid(TransportPid) -> + case is_process_alive(TransportPid) of + true -> + efka_transport:response(TransportPid, PacketId, Reply); + false -> + ok + end; +safe_response(_PacketId, _Reply, #state{}) -> + ok. \ No newline at end of file diff --git a/apps/efka/src/proto/message_pb.erl b/apps/efka/src/proto/message_pb.erl index 8c4ed85..95c3265 100644 --- a/apps/efka/src/proto/message_pb.erl +++ b/apps/efka/src/proto/message_pb.erl @@ -327,7 +327,7 @@ encode_msg_topic_message(#topic_message{topic = F1, content = F2}, Bin, TrUserDa encode_msg_service_params(Msg, TrUserData) -> encode_msg_service_params(Msg, <<>>, TrUserData). -encode_msg_service_params(#service_params{service_id = F1, params = F2}, Bin, TrUserData) -> +encode_msg_service_params(#service_params{service_id = F1, params = F2, timeout = F3}, Bin, TrUserData) -> B1 = if F1 == undefined -> Bin; true -> begin @@ -338,13 +338,22 @@ encode_msg_service_params(#service_params{service_id = F1, params = F2}, Bin, Tr end end end, - if F2 == undefined -> B1; + B2 = if F2 == undefined -> B1; + true -> + begin + TrF2 = id(F2, TrUserData), + case is_empty_string(TrF2) of + true -> B1; + false -> e_type_string(TrF2, <>, TrUserData) + end + end + end, + if F3 == undefined -> B2; true -> begin - TrF2 = id(F2, TrUserData), - case is_empty_string(TrF2) of - true -> B1; - false -> e_type_string(TrF2, <>, TrUserData) + TrF3 = id(F3, TrUserData), + if TrF3 =:= 0 -> B2; + true -> e_varint(TrF3, <>, TrUserData) end end end. @@ -352,7 +361,7 @@ encode_msg_service_params(#service_params{service_id = F1, params = F2}, Bin, Tr encode_msg_service_metrics(Msg, TrUserData) -> encode_msg_service_metrics(Msg, <<>>, TrUserData). -encode_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2}, Bin, TrUserData) -> +encode_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2, timeout = F3}, Bin, TrUserData) -> B1 = if F1 == undefined -> Bin; true -> begin @@ -363,13 +372,22 @@ encode_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2}, Bin, end end end, - if F2 == undefined -> B1; + B2 = if F2 == undefined -> B1; + true -> + begin + TrF2 = id(F2, TrUserData), + case is_empty_string(TrF2) of + true -> B1; + false -> e_type_string(TrF2, <>, TrUserData) + end + end + end, + if F3 == undefined -> B2; true -> begin - TrF2 = id(F2, TrUserData), - case is_empty_string(TrF2) of - true -> B1; - false -> e_type_string(TrF2, <>, TrUserData) + TrF3 = id(F3, TrUserData), + if TrF3 =:= 0 -> B2; + true -> e_varint(TrF3, <>, TrUserData) end end end. @@ -1208,107 +1226,121 @@ skip_32_topic_message(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) skip_64_topic_message(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_topic_message(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). -decode_msg_service_params(Bin, TrUserData) -> dfp_read_field_def_service_params(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData). +decode_msg_service_params(Bin, TrUserData) -> dfp_read_field_def_service_params(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(0, TrUserData), TrUserData). -dfp_read_field_def_service_params(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_service_params_service_id(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -dfp_read_field_def_service_params(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_service_params_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -dfp_read_field_def_service_params(<<>>, 0, 0, _, F@_1, F@_2, _) -> #service_params{service_id = F@_1, params = F@_2}; -dfp_read_field_def_service_params(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_service_params(Other, Z1, Z2, F, F@_1, F@_2, TrUserData). +dfp_read_field_def_service_params(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_params_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_params(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_params_params(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_params(<<24, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_params_timeout(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_params(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #service_params{service_id = F@_1, params = F@_2, timeout = F@_3}; +dfp_read_field_def_service_params(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dg_read_field_def_service_params(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). -dg_read_field_def_service_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_service_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -dg_read_field_def_service_params(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) -> +dg_read_field_def_service_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 32 - 7 -> dg_read_field_def_service_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +dg_read_field_def_service_params(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, TrUserData) -> Key = X bsl N + Acc, case Key of - 10 -> d_field_service_params_service_id(Rest, 0, 0, 0, F@_1, F@_2, TrUserData); - 18 -> d_field_service_params_params(Rest, 0, 0, 0, F@_1, F@_2, TrUserData); + 10 -> d_field_service_params_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); + 18 -> d_field_service_params_params(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); + 24 -> d_field_service_params_timeout(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); _ -> case Key band 7 of - 0 -> skip_varint_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 1 -> skip_64_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 2 -> skip_length_delimited_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 3 -> skip_group_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 5 -> skip_32_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData) + 0 -> skip_varint_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 1 -> skip_64_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 2 -> skip_length_delimited_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 3 -> skip_group_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 5 -> skip_32_service_params(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData) end end; -dg_read_field_def_service_params(<<>>, 0, 0, _, F@_1, F@_2, _) -> #service_params{service_id = F@_1, params = F@_2}. +dg_read_field_def_service_params(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #service_params{service_id = F@_1, params = F@_2, timeout = F@_3}. -d_field_service_params_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_service_params_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -d_field_service_params_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) -> +d_field_service_params_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_params_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_params_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, TrUserData) -> {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_service_params(RestF, 0, 0, F, NewFValue, F@_2, TrUserData). + dfp_read_field_def_service_params(RestF, 0, 0, F, NewFValue, F@_2, F@_3, TrUserData). -d_field_service_params_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_service_params_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -d_field_service_params_params(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, TrUserData) -> +d_field_service_params_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_params_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_params_params(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, TrUserData) -> {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_service_params(RestF, 0, 0, F, F@_1, NewFValue, TrUserData). + dfp_read_field_def_service_params(RestF, 0, 0, F, F@_1, NewFValue, F@_3, TrUserData). -skip_varint_service_params(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_service_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -skip_varint_service_params(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +d_field_service_params_timeout(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_params_timeout(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_params_timeout(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, TrUserData) -> + {NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest}, + dfp_read_field_def_service_params(RestF, 0, 0, F, F@_1, F@_2, NewFValue, TrUserData). -skip_length_delimited_service_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_service_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -skip_length_delimited_service_params(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) -> +skip_varint_service_params(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> skip_varint_service_params(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +skip_varint_service_params(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). + +skip_length_delimited_service_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> skip_length_delimited_service_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +skip_length_delimited_service_params(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) -> Length = X bsl N + Acc, <<_:Length/binary, Rest2/binary>> = Rest, - dfp_read_field_def_service_params(Rest2, 0, 0, F, F@_1, F@_2, TrUserData). + dfp_read_field_def_service_params(Rest2, 0, 0, F, F@_1, F@_2, F@_3, TrUserData). -skip_group_service_params(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) -> +skip_group_service_params(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, TrUserData) -> {_, Rest} = read_group(Bin, FNum), - dfp_read_field_def_service_params(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData). + dfp_read_field_def_service_params(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, TrUserData). -skip_32_service_params(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +skip_32_service_params(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). -skip_64_service_params(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +skip_64_service_params(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_params(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). -decode_msg_service_metrics(Bin, TrUserData) -> dfp_read_field_def_service_metrics(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData). +decode_msg_service_metrics(Bin, TrUserData) -> dfp_read_field_def_service_metrics(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(0, TrUserData), TrUserData). -dfp_read_field_def_service_metrics(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_service_metrics_service_id(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -dfp_read_field_def_service_metrics(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_service_metrics_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -dfp_read_field_def_service_metrics(<<>>, 0, 0, _, F@_1, F@_2, _) -> #service_metrics{service_id = F@_1, metrics = F@_2}; -dfp_read_field_def_service_metrics(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_service_metrics(Other, Z1, Z2, F, F@_1, F@_2, TrUserData). +dfp_read_field_def_service_metrics(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_metrics_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_metrics(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_metrics_metrics(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_metrics(<<24, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_service_metrics_timeout(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +dfp_read_field_def_service_metrics(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #service_metrics{service_id = F@_1, metrics = F@_2, timeout = F@_3}; +dfp_read_field_def_service_metrics(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dg_read_field_def_service_metrics(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). -dg_read_field_def_service_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_service_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -dg_read_field_def_service_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) -> +dg_read_field_def_service_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 32 - 7 -> dg_read_field_def_service_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +dg_read_field_def_service_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, TrUserData) -> Key = X bsl N + Acc, case Key of - 10 -> d_field_service_metrics_service_id(Rest, 0, 0, 0, F@_1, F@_2, TrUserData); - 18 -> d_field_service_metrics_metrics(Rest, 0, 0, 0, F@_1, F@_2, TrUserData); + 10 -> d_field_service_metrics_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); + 18 -> d_field_service_metrics_metrics(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); + 24 -> d_field_service_metrics_timeout(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData); _ -> case Key band 7 of - 0 -> skip_varint_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 1 -> skip_64_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 2 -> skip_length_delimited_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 3 -> skip_group_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData); - 5 -> skip_32_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData) + 0 -> skip_varint_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 1 -> skip_64_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 2 -> skip_length_delimited_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 3 -> skip_group_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData); + 5 -> skip_32_service_metrics(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData) end end; -dg_read_field_def_service_metrics(<<>>, 0, 0, _, F@_1, F@_2, _) -> #service_metrics{service_id = F@_1, metrics = F@_2}. +dg_read_field_def_service_metrics(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #service_metrics{service_id = F@_1, metrics = F@_2, timeout = F@_3}. -d_field_service_metrics_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_service_metrics_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -d_field_service_metrics_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) -> +d_field_service_metrics_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_metrics_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_metrics_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, TrUserData) -> {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_service_metrics(RestF, 0, 0, F, NewFValue, F@_2, TrUserData). + dfp_read_field_def_service_metrics(RestF, 0, 0, F, NewFValue, F@_2, F@_3, TrUserData). -d_field_service_metrics_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_service_metrics_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -d_field_service_metrics_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, TrUserData) -> +d_field_service_metrics_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_metrics_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_metrics_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, TrUserData) -> {NewFValue, RestF} = begin Len = X bsl N + Acc, <> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end, - dfp_read_field_def_service_metrics(RestF, 0, 0, F, F@_1, NewFValue, TrUserData). + dfp_read_field_def_service_metrics(RestF, 0, 0, F, F@_1, NewFValue, F@_3, TrUserData). -skip_varint_service_metrics(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData); -skip_varint_service_metrics(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +d_field_service_metrics_timeout(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_service_metrics_timeout(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +d_field_service_metrics_timeout(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, TrUserData) -> + {NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest}, + dfp_read_field_def_service_metrics(RestF, 0, 0, F, F@_1, F@_2, NewFValue, TrUserData). -skip_length_delimited_service_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_service_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData); -skip_length_delimited_service_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) -> +skip_varint_service_metrics(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> skip_varint_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData); +skip_varint_service_metrics(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). + +skip_length_delimited_service_metrics(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> skip_length_delimited_service_metrics(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData); +skip_length_delimited_service_metrics(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) -> Length = X bsl N + Acc, <<_:Length/binary, Rest2/binary>> = Rest, - dfp_read_field_def_service_metrics(Rest2, 0, 0, F, F@_1, F@_2, TrUserData). + dfp_read_field_def_service_metrics(Rest2, 0, 0, F, F@_1, F@_2, F@_3, TrUserData). -skip_group_service_metrics(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) -> +skip_group_service_metrics(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, TrUserData) -> {_, Rest} = read_group(Bin, FNum), - dfp_read_field_def_service_metrics(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData). + dfp_read_field_def_service_metrics(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, TrUserData). -skip_32_service_metrics(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +skip_32_service_metrics(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). -skip_64_service_metrics(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData). +skip_64_service_metrics(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_service_metrics(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData). decode_msg_data(Bin, TrUserData) -> dfp_read_field_def_data(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData). @@ -2003,7 +2035,7 @@ merge_msg_topic_message(#topic_message{topic = PFtopic, content = PFcontent}, #t end}. -compile({nowarn_unused_function,merge_msg_service_params/3}). -merge_msg_service_params(#service_params{service_id = PFservice_id, params = PFparams}, #service_params{service_id = NFservice_id, params = NFparams}, _) -> +merge_msg_service_params(#service_params{service_id = PFservice_id, params = PFparams, timeout = PFtimeout}, #service_params{service_id = NFservice_id, params = NFparams, timeout = NFtimeout}, _) -> #service_params{service_id = if NFservice_id =:= undefined -> PFservice_id; true -> NFservice_id @@ -2011,10 +2043,14 @@ merge_msg_service_params(#service_params{service_id = PFservice_id, params = PFp params = if NFparams =:= undefined -> PFparams; true -> NFparams + end, + timeout = + if NFtimeout =:= undefined -> PFtimeout; + true -> NFtimeout end}. -compile({nowarn_unused_function,merge_msg_service_metrics/3}). -merge_msg_service_metrics(#service_metrics{service_id = PFservice_id, metrics = PFmetrics}, #service_metrics{service_id = NFservice_id, metrics = NFmetrics}, _) -> +merge_msg_service_metrics(#service_metrics{service_id = PFservice_id, metrics = PFmetrics, timeout = PFtimeout}, #service_metrics{service_id = NFservice_id, metrics = NFmetrics, timeout = NFtimeout}, _) -> #service_metrics{service_id = if NFservice_id =:= undefined -> PFservice_id; true -> NFservice_id @@ -2022,6 +2058,10 @@ merge_msg_service_metrics(#service_metrics{service_id = PFservice_id, metrics = metrics = if NFmetrics =:= undefined -> PFmetrics; true -> NFmetrics + end, + timeout = + if NFtimeout =:= undefined -> PFtimeout; + true -> NFtimeout end}. -compile({nowarn_unused_function,merge_msg_data/3}). @@ -2288,25 +2328,31 @@ v_msg_topic_message(X, Path, _TrUserData) -> mk_type_error({expected_msg, topic_ -compile({nowarn_unused_function,v_msg_service_params/3}). -dialyzer({nowarn_function,v_msg_service_params/3}). -v_msg_service_params(#service_params{service_id = F1, params = F2}, Path, TrUserData) -> +v_msg_service_params(#service_params{service_id = F1, params = F2, timeout = F3}, Path, TrUserData) -> if F1 == undefined -> ok; true -> v_type_string(F1, [service_id | Path], TrUserData) end, if F2 == undefined -> ok; true -> v_type_string(F2, [params | Path], TrUserData) end, + if F3 == undefined -> ok; + true -> v_type_uint32(F3, [timeout | Path], TrUserData) + end, ok; v_msg_service_params(X, Path, _TrUserData) -> mk_type_error({expected_msg, service_params}, X, Path). -compile({nowarn_unused_function,v_msg_service_metrics/3}). -dialyzer({nowarn_function,v_msg_service_metrics/3}). -v_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2}, Path, TrUserData) -> +v_msg_service_metrics(#service_metrics{service_id = F1, metrics = F2, timeout = F3}, Path, TrUserData) -> if F1 == undefined -> ok; true -> v_type_string(F1, [service_id | Path], TrUserData) end, if F2 == undefined -> ok; true -> v_type_string(F2, [metrics | Path], TrUserData) end, + if F3 == undefined -> ok; + true -> v_type_uint32(F3, [timeout | Path], TrUserData) + end, ok; v_msg_service_metrics(X, Path, _TrUserData) -> mk_type_error({expected_msg, service_metrics}, X, Path). @@ -2540,8 +2586,14 @@ get_msg_defs() -> #field{name = result, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, #field{name = message, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]}, {{msg, topic_message}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}, - {{msg, service_params}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}, - {{msg, service_metrics}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}, + {{msg, service_params}, + [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, + #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, + #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]}, + {{msg, service_metrics}, + [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, + #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, + #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]}, {{msg, data}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}, {{msg, ping}, [#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, @@ -2621,8 +2673,14 @@ find_msg_def(efka_response) -> #field{name = result, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, #field{name = message, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]; find_msg_def(topic_message) -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]; -find_msg_def(service_params) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]; -find_msg_def(service_metrics) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]; +find_msg_def(service_params) -> + [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, + #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, + #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]; +find_msg_def(service_metrics) -> + [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, + #field{name = metrics, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}, + #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]; find_msg_def(data) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]; find_msg_def(ping) -> [#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, diff --git a/message_pb.proto b/message_pb.proto index ae5dca8..6e265aa 100644 --- a/message_pb.proto +++ b/message_pb.proto @@ -45,12 +45,14 @@ message TopicMessage { message ServiceParams { string service_id = 1; string params = 2; + uint32 timeout = 3; } // 服务采集项 message ServiceMetrics { string service_id = 1; string metrics = 2; + uint32 timeout = 3; } // 数据传输