fix get_range for case where only the middle of a block is requested

tested by luwak_io_tests:partial_middle_block_get_range_test/0

src/luwak_io.erl

@@ -206,9 +206,8 @@ retrieve_blocks(Riak, Blocks, ChopHead, Length) ->
 
 retrieve_blocks(_Riak, [], _, _, Acc) ->
     lists:reverse(Acc);
-retrieve_blocks(Riak, [{Name,_}], _, 0, Acc) ->
-    {ok, Block} = luwak_tree:get(Riak, Name),
-    retrieve_blocks(Riak, [], 0, 0, [luwak_block:data(Block)|Acc]);
+retrieve_blocks(_Riak, _, _, 0, Acc) ->
+    lists:reverse(Acc);
 retrieve_blocks(Riak, [{Name,_}], _, Length, Acc) ->
     {ok, Block} = luwak_tree:get(Riak, Name),
     PreData = luwak_block:data(Block),
@@ -222,4 +221,12 @@ retrieve_blocks(Riak, [{Name,_}], _, Length, Acc) ->
 retrieve_blocks(Riak, [{Name,_}|Children], ChopHead, Length, Acc) ->
     {ok, Block} = luwak_tree:get(Riak, Name),
     <<_:ChopHead/binary, Data/binary>> = luwak_block:data(Block),
-    retrieve_blocks(Riak, Children, 0, Length - byte_size(Data), [Data|Acc]).
+    case byte_size(Data) of
+        ByteSize when ByteSize =< Length ->
+            %% wanted the rest of the block
+            retrieve_blocks(Riak, Children, 0, Length - ByteSize, [Data|Acc]);
+        _ ->
+            %% wanted only a middle chunk of the block
+            <<SubData:Length/binary, _/binary>> = Data,
+            retrieve_blocks(Riak, Children, 0, 0, [SubData|Acc])
+    end.

test/luwak_io_tests.erl

@@ -44,6 +44,69 @@ multilevel_get_range_test() ->
       ?assertEqual(<<"youplease">>, iolist_to_binary(Blocks))
     end).
 
+%% @doc Makes a simple file, named Name, with BlockCount blocks, each
+%%      filled with BlockSize characters.
+%%      Ex: make_standard_range_file(R, <<"foo">>, 5, 3)
+%%      will make a file named "foo" with the contents:
+%%      aaaaabbbbbccccc
+%% @spec make_standard_range_file(riak_client(), binary(),
+%%                                integer(), integer())
+%%         -> {ok, Contents::binary(), FileHandle::luwak_file()}
+make_standard_range_file(Riak, Name, BlockSize, BlockCount) ->
+    {ok, File} = luwak_file:create(Riak, Name,
+                                   [{block_size, BlockSize}],
+                                   dict:new()),
+    Data = iolist_to_binary(
+             [ lists:duplicate(BlockSize, $a+N) ||
+                 N <- lists:seq(0, BlockCount-1) ]),
+    {ok, _, NewFile} = luwak_io:put_range(Riak, File, 0, Data),
+    {ok, Data, NewFile}.
+
+%% tests a get_range that specifies an offset+length that ends in
+%% the middle of a block
+%% Ex: aaaaaaaaaa bbbbbbbbbb cccccccccc
+%%                ^start      end^
+partial_end_block_get_range_test() ->
+    test_helper:riak_test(
+      fun(Riak) ->
+              {ok, Data, File} =  make_standard_range_file(
+                                    Riak, <<"endblockrange">>, 10, 3),
+              Read = iolist_to_binary(
+                       luwak_io:get_range(Riak, File, 10, 14)),
+              <<_:10/binary, Expected:14/binary, _/binary>> = Data,
+              ?assertEqual(Expected, Read)
+      end).
+
+%% tests a get_range that specifies an offset that lands in the
+%% middle of a block
+%% Ex: aaaaaaaaaa bbbbbbbbbb cccccccccc
+%%          ^start       end^
+partial_start_block_get_range_test() ->
+    test_helper:riak_test(
+      fun(Riak) ->
+              {ok, Data, File} = make_standard_range_file(
+                                   Riak, <<"startblockrange">>, 10, 3),
+              Read = iolist_to_binary(
+                       luwak_io:get_range(Riak, File, 5, 15)),
+              <<_:5/binary, Expected:15/binary, _/binary>> = Data,
+              ?assertEqual(Expected, Read)
+      end).
+
+%% tests a get_range that starts and ends in the same block, but
+%% is unaligned with that block
+%% Ex: aaaaaaaaaa bbbbbbbbbb cccccccccc
+%%                 ^st end^
+partial_middle_block_get_range_test() ->
+    test_helper:riak_test(
+      fun(Riak) ->
+              {ok, Data, File} = make_standard_range_file(
+                                   Riak, <<"midblockrange">>, 10, 3),
+               Read = iolist_to_binary(
+                        luwak_io:get_range(Riak, File, 11, 8)),
+               <<_:11/binary, Expected:8/binary, _/binary>> = Data,
+              ?assertEqual(Expected, Read)
+      end).
+
 eof_get_range_test() ->
   test_helper:riak_test(fun(Riak) ->
       {ok, File} = luwak_file:create(Riak, <<"file1">>, [{block_size,3},{tree_order,3}], dict:new()),