valitydev/machinery-erlang
14
0
Fork 0
mirror of https://github.com/valitydev/machinery-erlang.git synced 2024-11-06 00:35:19 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

FF-137: add call by tag (#16)

Browse Source 
* update machinery

* udpate machinery backend

* update machinery mg backend

* review updates

* fixes

* add ref type

* review updates

* export ref type
This commit is contained in:
Roman Pushkov 2019-11-06 11:59:06 +03:00 committed by GitHub
parent 6e82ff5ca3
commit 1fa4e1d058
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 62 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/machinery.erl
View File

@ -11,7 +11,9 @@
-module(machinery).
-type namespace() :: atom().
-type ref() :: id() | {tag, tag()}.
-type id() :: binary().
-type tag() :: binary().
-type args(T) :: T.
-type response(T) :: T.
@ -41,6 +43,8 @@
-export_type([namespace/0]).
-export_type([id/0]).
-export_type([tag/0]).
-export_type([ref/0]).
-export_type([range/0]).
-export_type([args/1]).
-export_type([response/1]).
@ -94,6 +98,7 @@
{set_timer, timer()} |
{set_timer, timer(), range()} |
{set_timer, timer(), range(), seconds()} |
{tag, tag()} |
unset_timer |
continue |
remove.
@ -124,38 +129,38 @@ start(NS, ID, Args, Backend) ->
{Module, Opts} = machinery_utils:get_backend(Backend),
machinery_backend:start(Module, NS, ID, Args, Opts).
-spec call(namespace(), id(), args(_), backend(_)) ->
-spec call(namespace(), ref(), args(_), backend(_)) ->
{ok, response(_)} | {error, notfound}.
call(NS, ID, Args, Backend) ->
call(NS, ID, {undefined, undefined, forward}, Args, Backend).
call(NS, Ref, Args, Backend) ->
call(NS, Ref, {undefined, undefined, forward}, Args, Backend).
-spec call(namespace(), id(), range(), args(_), backend(_)) ->
-spec call(namespace(), ref(), range(), args(_), backend(_)) ->
{ok, response(_)} | {error, notfound}.
call(NS, ID, Range, Args, Backend) ->
call(NS, Ref, Range, Args, Backend) ->
{Module, Opts} = machinery_utils:get_backend(Backend),
machinery_backend:call(Module, NS, ID, Range, Args, Opts).
machinery_backend:call(Module, NS, Ref, Range, Args, Opts).
-spec repair(namespace(), id(), args(_), backend(_)) ->
-spec repair(namespace(), ref(), args(_), backend(_)) ->
ok | {error, notfound | working}.
repair(NS, ID, Args, Backend) ->
repair(NS, ID, {undefined, undefined, forward}, Args, Backend).
repair(NS, Ref, Args, Backend) ->
repair(NS, Ref, {undefined, undefined, forward}, Args, Backend).
-spec repair(namespace(), id(), range(), args(_), backend(_)) ->
-spec repair(namespace(), ref(), range(), args(_), backend(_)) ->
ok | {error, notfound | working}.
repair(NS, ID, Range, Args, Backend) ->
repair(NS, Ref, Range, Args, Backend) ->
{Module, Opts} = machinery_utils:get_backend(Backend),
machinery_backend:repair(Module, NS, ID, Range, Args, Opts).
machinery_backend:repair(Module, NS, Ref, Range, Args, Opts).
-spec get(namespace(), id(), backend(_)) ->
-spec get(namespace(), ref(), backend(_)) ->
{ok, machine(_, _)} | {error, notfound}.
get(NS, ID, Backend) ->
get(NS, ID, {undefined, undefined, forward}, Backend).
get(NS, Ref, Backend) ->
get(NS, Ref, {undefined, undefined, forward}, Backend).
-spec get(namespace(), id(), range(), backend(_)) ->
-spec get(namespace(), ref(), range(), backend(_)) ->
{ok, machine(_, _)} | {error, notfound}.
get(NS, ID, Range, Backend) ->
get(NS, Ref, Range, Backend) ->
{Module, Opts} = machinery_utils:get_backend(Backend),
machinery_backend:get(Module, NS, ID, Range, Opts).
machinery_backend:get(Module, NS, Ref, Range, Opts).
%% Internal API

19
src/machinery_backend.erl
View File

@ -13,6 +13,7 @@
-type namespace() :: machinery:namespace().
-type id() :: machinery:id().
-type ref() :: machinery:ref().
-type range() :: machinery:range().
-type args() :: machinery:args(_).
-type backend_opts() :: machinery:backend_opts(_).
@ -38,17 +39,17 @@
start(Backend, Namespace, Id, Args, Opts) ->
Backend:start(Namespace, Id, Args, Opts).
-spec call(backend(), namespace(), id(), range(), args(), backend_opts()) ->
-spec call(backend(), namespace(), ref(), range(), args(), backend_opts()) ->
{ok, machinery:response(_)} | {error, notfound}.
call(Backend, Namespace, Id, Range, Args, Opts) ->
Backend:call(Namespace, Id, Range, Args, Opts).
call(Backend, Namespace, Ref, Range, Args, Opts) ->
Backend:call(Namespace, Ref, Range, Args, Opts).
-spec repair(backend(), namespace(), id(), range(), args(), backend_opts()) ->
-spec repair(backend(), namespace(), ref(), range(), args(), backend_opts()) ->
ok | {error, notfound | working}.
repair(Backend, Namespace, Id, Range, Args, Opts) ->
Backend:repair(Namespace, Id, Range, Args, Opts).
repair(Backend, Namespace, Ref, Range, Args, Opts) ->
Backend:repair(Namespace, Ref, Range, Args, Opts).
-spec get(backend(), namespace(), id(), range(), backend_opts()) ->
-spec get(backend(), namespace(), ref(), range(), backend_opts()) ->
{ok, machinery:machine(_, _)} | {error, notfound}.
get(Backend, Namespace, Id, Range, Opts) ->
Backend:get(Namespace, Id, Range, Opts).
get(Backend, Namespace, Ref, Range, Opts) ->
Backend:get(Namespace, Ref, Range, Opts).

41
src/machinery_mg_backend.erl
View File

@ -9,6 +9,7 @@
-include_lib("mg_proto/include/mg_proto_state_processing_thrift.hrl").
-type namespace() :: machinery:namespace().
-type ref() :: machinery:ref().
-type id() :: machinery:id().
-type range() :: machinery:range().
-type args(T) :: machinery:args(T).
@ -120,12 +121,12 @@ start(NS, ID, Args, Opts) ->
error({failed, NS, ID})
end.
-spec call(namespace(), id(), range(), args(_), backend_opts()) ->
-spec call(namespace(), ref(), range(), args(_), backend_opts()) ->
{ok, response(_)} | {error, notfound}.
call(NS, ID, Range, Args, Opts) ->
call(NS, Ref, Range, Args, Opts) ->
Client = get_client(Opts),
Schema = get_schema(Opts),
Descriptor = {NS, ID, Range},
Descriptor = {NS, Ref, Range},
CallArgs = marshal({schema, Schema, {args, call}}, Args),
case machinery_mg_client:call(marshal(descriptor, Descriptor), CallArgs, Client) of
{ok, Response} ->
@ -135,15 +136,15 @@ call(NS, ID, Range, Args, Opts) ->
{exception, #mg_stateproc_NamespaceNotFound{}} ->
error({namespace_not_found, NS});
{exception, #mg_stateproc_MachineFailed{}} ->
error({failed, NS, ID})
error({failed, NS, Ref})
end.
-spec repair(namespace(), id(), range(), args(_), backend_opts()) ->
-spec repair(namespace(), ref(), range(), args(_), backend_opts()) ->
ok | {error, notfound | working}.
repair(NS, ID, Range, Args, Opts) ->
repair(NS, Ref, Range, Args, Opts) ->
Client = get_client(Opts),
Schema = get_schema(Opts),
Descriptor = {NS, ID, Range},
Descriptor = {NS, Ref, Range},
CallArgs = marshal({schema, Schema, {args, repair}}, Args),
case machinery_mg_client:repair(marshal(descriptor, Descriptor), CallArgs, Client) of
{ok, ok} ->
@ -156,12 +157,12 @@ repair(NS, ID, Range, Args, Opts) ->
error({namespace_not_found, NS})
end.
-spec get(namespace(), id(), range(), backend_opts()) ->
-spec get(namespace(), ref(), range(), backend_opts()) ->
{ok, machine(_, _)} | {error, notfound}.
get(NS, ID, Range, Opts) ->
get(NS, Ref, Range, Opts) ->
Client = get_client(Opts),
Schema = get_schema(Opts),
Descriptor = {NS, ID, Range},
Descriptor = {NS, Ref, Range},
case machinery_mg_client:get_machine(marshal(descriptor, Descriptor), Client) of
{ok, Machine} ->
{ok, unmarshal({machine, Schema}, Machine)};
@ -261,10 +262,10 @@ set_aux_state(NewState, _) ->
%% 'history_range' = marshal(range, {undefined, undefined, forward})
%% };
marshal(descriptor, {NS, ID, Range}) ->
marshal(descriptor, {NS, Ref, Range}) ->
#mg_stateproc_MachineDescriptor{
'ns' = marshal(namespace, NS),
'ref' = {'id', marshal(id, ID)},
'ref' = marshal(ref, Ref),
'range' = marshal(range, Range)
};
@ -334,9 +335,18 @@ marshal(timer, {deadline, V}) ->
marshal(namespace, V) ->
marshal(atom, V);
marshal(ref, V) when is_binary(V) ->
{id, marshal(id, V)};
marshal(ref, {tag, V}) ->
{tag, marshal(tag, V)};
marshal(id, V) ->
marshal(string, V);
marshal(tag, V) ->
marshal(string, V);
marshal(event_id, V) ->
marshal(integer, V);
@ -417,6 +427,13 @@ apply_action(continue, CA) ->
apply_action(remove, CA) ->
CA#mg_stateproc_ComplexAction{
remove = #mg_stateproc_RemoveAction{}
};
apply_action({tag, Tag}, CA) ->
CA#mg_stateproc_ComplexAction{
tag = #mg_stateproc_TagAction{
tag = marshal(tag, Tag)
}
}.
%%