-
Notifications
You must be signed in to change notification settings - Fork 20
/
luwak_file.erl
163 lines (149 loc) · 6.36 KB
/
luwak_file.erl
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
-module(luwak_file).
-export([create/3,
create/4,
set_attributes/3,
get_attributes/1,
exists/2,
delete/2,
get/2,
get_property/2,
update_root/3,
update_checksum/3,
name/1,
length/2]).
-include_lib("luwak.hrl").
%% @spec create(Riak :: riak(), Name :: binary(), Attributes :: dict())
%% -> {ok, File :: luwak_file()}
%% @doc Create a luwak file handle with the given name and attributes. Will
%% overwrite an existing file of the same name.
%% @equiv create(Riak, Name, [], Attributes)
create(Riak, Name, Attributes) when is_binary(Name) ->
create(Riak, Name, [], Attributes).
%% @spec create(Riak :: riak(), Name :: binary(), Properties :: proplist(),
%% Attributes :: dict())
%% -> {ok, File :: luwak_file()}
%% @doc Create a luwak file handle with the given name and attributes.
%% Recognized properties:
%% {block_size, int()} - The maximum size of an individual data chunk
%% in bytes. Default is 1000000.
%% {tree_order, int()} - The maximum number of children for an
%% individual tree node. Default is 250.
%% {checksumming, boolean()} - This controls whether or not checksumming
%% will occur. Checksumming will leave a
%% checksum of the most recent write request
%% as a property of the filehandle.
create(Riak, Name, Properties, Attributes) when is_binary(Name) ->
DefaultBlockSize = app_helper:get_env(luwak,
default_block_size,
?BLOCK_DEFAULT),
BlockSize = proplists:get_value(block_size, Properties, DefaultBlockSize),
Order = proplists:get_value(tree_order, Properties, ?ORDER_DEFAULT),
Checksumming = proplists:get_value(checksumming, Properties, false),
if
Order < 2 -> throw("tree_order cannot be less than 2");
BlockSize < 1 -> throw("block_size cannot be less than 1");
true -> ok
end,
Value = [
{attributes, Attributes},
{block_size, BlockSize},
{created, now()},
{modified, now()},
{checksumming, Checksumming},
{checksum, undefined},
{tree_order, Order},
{ancestors, []},
{root, undefined}
],
Obj = riak_object:new(?O_BUCKET, Name, Value),
Riak:put(Obj, 2, 2, ?TIMEOUT_DEFAULT, [{returnbody, true}]).
%% @spec set_attributes(Riak :: riak(), Obj :: luwak_file(),
%% Attributes :: dict())
%% -> {ok, NewFile}
%% @doc Sets the new attributes, saves them, and returns a new file handle.
set_attributes(Riak, Obj, Attributes) ->
Value = lists:keyreplace(attributes, 1, riak_object:get_value(Obj),
{attributes, Attributes}),
Obj2 = riak_object:update_value(Obj, Value),
Riak:put(Obj2, 2, 2, ?TIMEOUT_DEFAULT, [{returnbody, true}]).
%% @spec get_attributes(Obj :: luwak_file()) -> dict()
%% @doc Gets the attribute dictionary from the file handle.
get_attributes(Obj) ->
proplists:get_value(attributes, riak_object:get_value(Obj)).
%% @spec exists(Riak :: riak(), Name :: binary())
%% -> {ok, true} | {ok, false} | {error, Reason}
%% @doc Checks for the existence of the named file.
exists(Riak, Name) ->
case Riak:get(?O_BUCKET, Name, 2) of
{ok, _Obj} -> {ok, true};
{error, notfound} -> {ok, false};
Err -> Err
end.
%% @spec length(Riak :: riak(), File :: luwak_file()) -> Length
%% @doc returns the length in bytes of the file.
length(Riak, File) ->
case get_property(File, root) of
undefined ->
0;
RootName ->
{ok, Node} = luwak_tree:get(Riak, RootName),
case Node of
#n{children=Children} ->
luwak_tree_utils:blocklist_length(Children);
Block ->
byte_size(luwak_block:data(Block))
end
end.
%% @spec delete(Riak :: riak(), Name :: binary()) -> ok | {error, Reason}
%% @doc deletes the named file from luwak. This is a fast operation since
%% the blocks and tree for that file remain untouched. A GC operation
%% (not yet implemented) will be required to clean them up properly.
delete(Riak, Name) ->
Riak:delete(?O_BUCKET, Name, 2).
%% @spec get(Riak :: riak(), Name :: binary()) -> {ok, File} | {error, Reason}
%% @doc returns a filehandle for the named file.
get(Riak, Name) ->
Riak:get(?O_BUCKET, Name, 2).
%% @spec get_property(Obj :: luwak_file(), PropName :: atom()) -> Property
%% @doc retrieves the named property from the filehandle.
get_property(Obj, type) ->
case riak_object:get_value(Obj) of
List when is_list(List) -> proplists:get_value(type, List);
#n{} -> node
end;
get_property(Obj, links) ->
case riak_object:get_value(Obj) of
List when is_list(List) -> proplists:get_value(links, List);
#n{children=Children} -> Children
end;
get_property(Obj, PropName) ->
case riak_object:get_value(Obj) of
List when is_list(List) -> proplists:get_value(PropName, List);
_ -> undefined
end.
%% @private
update_root(Riak, Obj, NewRoot) ->
ObjVal1 = riak_object:get_value(Obj),
OldRoot = proplists:get_value(root, ObjVal1),
Ancestors = proplists:get_value(ancestors, ObjVal1),
ObjVal2 = lists:keyreplace(ancestors, 1, ObjVal1,
{ancestors, [OldRoot|Ancestors]}),
ObjVal3 = lists:keyreplace(root, 1, ObjVal2, {root, NewRoot}),
Obj2 = riak_object:update_value(Obj, ObjVal3),
Riak:put(Obj2, 2, 2, ?TIMEOUT_DEFAULT, [{returnbody, true}]).
%% @private
update_checksum(Riak, Obj, ChecksumFun) ->
case get_property(Obj, checksumming) of
true ->
ObjVal1 = riak_object:get_value(Obj),
ObjVal2 = lists:keyreplace(checksum, 1, ObjVal1,
{checksum, {sha1, ChecksumFun()}}),
Obj2 = riak_object:update_value(Obj, ObjVal2),
Riak:put(Obj2, 2, 2, ?TIMEOUT_DEFAULT, [{returnbody, true}]);
_ ->
{ok, Obj}
end.
%% @spec name(Obj :: luwak_file()) -> binary()
%% @doc returns the name of the given file handle.
name(Obj) ->
riak_object:key(Obj).