Fix leo-project/leofs/issues/146

leo-project · Mar 4, 2014 · 0b2e3a8 · 0b2e3a8
1 parent c211625
commit 0b2e3a8
Show file tree

Hide file tree

Showing 2 changed files with 73 additions and 15 deletions.
diff --git a/src/leo_storage_handler_object.erl b/src/leo_storage_handler_object.erl
@@ -321,13 +321,32 @@ delete(_,_) ->
              {ok, #metadata{}} |
              {error, any}).
 head(AddrId, Key) ->
+    case leo_redundant_manager_api:get_redundancies_by_addr_id(get, AddrId) of
+        {ok, #redundancies{nodes = Redundancies}} ->
+            head_1(Redundancies, AddrId, Key);
+        _ ->
+            {error, ?ERROR_COULD_NOT_GET_REDUNDANCY}
+    end.
+
+%% @private
+head_1([],_,_) ->
+    {error, not_found};
+head_1([#redundant_node{node = Node,
+                        available = true}|Rest], AddrId, Key) when Node == erlang:node() ->
     case leo_object_storage_api:head({AddrId, Key}) of
         {ok, MetaBin} ->
             {ok, binary_to_term(MetaBin)};
-        not_found = Cause ->
-            {error, Cause};
-        {error, Why} ->
-            {error, Why}
+        _Other ->
+            head_1(Rest, AddrId, Key)
+    end;
+head_1([#redundant_node{node = Node,
+                        available = true}|Rest], AddrId, Key) ->
+    RPCKey = rpc:async_call(Node, ?MODULE, head, [{AddrId, Key}]),
+    case rpc:nb_yield(RPCKey, ?DEF_REQ_TIMEOUT) of
+        {value, {ok, MetaBin}} ->
+            {ok, binary_to_term(MetaBin)};
+        _ ->
+            head_1(Rest, AddrId, Key)
     end.
 
 
@@ -340,16 +359,22 @@ head(AddrId, Key) ->
              ok | not_found | {error, any()}).
 copy(DestNodes, AddrId, Key) ->
     Ref = make_ref(),
-    case ?MODULE:head(AddrId, Key) of
-        {ok, #metadata{del = ?DEL_FALSE} = Metadata} ->
-            case ?MODULE:get({Ref, Key}) of
-                {ok, Ref, Metadata, Bin} ->
-                    leo_storage_ordning_reda_client:stack(DestNodes, AddrId, Key, Metadata, Bin);
-                {error, Ref, Cause} ->
-                    {error, Cause}
+
+    case leo_object_storage_api:head({AddrId, Key}) of
+        {ok, MetaBin} ->
+            case binary_to_term(MetaBin) of
+                #metadata{del = ?DEL_FALSE} = Metadata ->
+                    case ?MODULE:get({Ref, Key}) of
+                        {ok, Ref, Metadata, Bin} ->
+                            leo_storage_ordning_reda_client:stack(DestNodes, AddrId, Key, Metadata, Bin);
+                        {error, Ref, Cause} ->
+                            {error, Cause}
+                    end;
+                #metadata{del = ?DEL_TRUE} = Metadata ->
+                    leo_storage_ordning_reda_client:stack(DestNodes, AddrId, Key, Metadata, <<>>);
+                _ ->
+                    {error, invalid_data_type}
             end;
-        {ok, #metadata{del = ?DEL_TRUE} = Metadata} ->
-            leo_storage_ordning_reda_client:stack(DestNodes, AddrId, Key, Metadata, <<>>);
         Error ->
             Error
     end.

diff --git a/test/leo_storage_handler_object_tests.erl b/test/leo_storage_handler_object_tests.erl
@@ -561,13 +561,24 @@ delete_2_({_Node0, _Node1}) ->
 %%--------------------------------------------------------------------
 %% OTHER
 %%--------------------------------------------------------------------
-head_({_Node0, _Node1}) ->
+head_({Node0, Node1}) ->
     %% 1.
     meck:new(leo_object_storage_api),
     meck:expect(leo_object_storage_api, head,
                 fun(_Key) ->
                         {ok, term_to_binary(?TEST_META_0)}
                 end),
+    meck:new(leo_redundant_manager_api),
+    meck:expect(leo_redundant_manager_api, get_redundancies_by_addr_id,
+                fun(get, _AddrId) ->
+                        {ok, #redundancies{id = 0,
+                                           nodes = [#redundant_node{node = Node0,
+                                                                     available = true},
+                                                    #redundant_node{node = Node1,
+                                                                    available = true}],
+                                           n = 2, r = 1, w = 1, d = 1}}
+                end),
+
     Res0 = leo_storage_handler_object:head(0, ?TEST_KEY_0),
     ?assertEqual({ok, ?TEST_META_0}, Res0),
 
@@ -578,6 +589,17 @@ head_({_Node0, _Node1}) ->
                 fun(_Key) ->
                         not_found
                 end),
+    meck:new(leo_redundant_manager_api),
+    meck:expect(leo_redundant_manager_api, get_redundancies_by_addr_id,
+                fun(get, _AddrId) ->
+                        {ok, #redundancies{id = 0,
+                                           nodes = [#redundant_node{node = Node0,
+                                                                     available = true},
+                                                    #redundant_node{node = Node1,
+                                                                    available = true}],
+                                           n = 2, r = 1, w = 1, d = 1}}
+                end),
+
     Res1 = leo_storage_handler_object:head(0, ?TEST_KEY_0),
     ?assertEqual({error, not_found}, Res1),
 
@@ -588,8 +610,19 @@ head_({_Node0, _Node1}) ->
                 fun(_Key) ->
                         {error, []}
                 end),
+    meck:new(leo_redundant_manager_api),
+    meck:expect(leo_redundant_manager_api, get_redundancies_by_addr_id,
+                fun(get, _AddrId) ->
+                        {ok, #redundancies{id = 0,
+                                           nodes = [#redundant_node{node = Node0,
+                                                                     available = true},
+                                                    #redundant_node{node = Node1,
+                                                                    available = true}],
+                                           n = 2, r = 1, w = 1, d = 1}}
+                end),
+
     Res2 = leo_storage_handler_object:head(0, ?TEST_KEY_0),
-    ?assertEqual({error, []}, Res2),
+    ?assertEqual({error, not_found}, Res2),
     ok.