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
|
%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (the "License"); you may not use this file except in
%% compliance with the License. You may obtain a copy of the License
%% at http://www.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and
%% limitations under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developer of the Original Code is GoPivotal, Inc.
%% Copyright (c) 2007-2014 GoPivotal, Inc. All rights reserved.
%%
%% We use a gen_server simply so that during the terminate/2 call
%% (i.e., during shutdown), we can sync/flush the dets table to disk.
-module(rabbit_recovery_terms).
-behaviour(gen_server).
-export([start/0, stop/0, store/2, read/1, clear/0]).
-export([upgrade_recovery_terms/0, start_link/0]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-rabbit_upgrade({upgrade_recovery_terms, local, []}).
%%----------------------------------------------------------------------------
-ifdef(use_specs).
-spec(start() -> rabbit_types:ok_or_error(term())).
-spec(stop() -> rabbit_types:ok_or_error(term())).
-spec(store(file:filename(), term()) -> rabbit_types:ok_or_error(term())).
-spec(read(file:filename()) -> rabbit_types:ok_or_error2(term(), not_found)).
-spec(clear() -> 'ok').
-endif. % use_specs
%%----------------------------------------------------------------------------
-define(SERVER, ?MODULE).
start() -> rabbit_sup:start_child(?MODULE).
stop() -> rabbit_sup:stop_child(?MODULE).
store(DirBaseName, Terms) -> dets:insert(?MODULE, {DirBaseName, Terms}).
read(DirBaseName) ->
case dets:lookup(?MODULE, DirBaseName) of
[{_, Terms}] -> {ok, Terms};
_ -> {error, not_found}
end.
clear() ->
ok = dets:delete_all_objects(?MODULE),
flush().
%%----------------------------------------------------------------------------
upgrade_recovery_terms() ->
open_table(),
try
QueuesDir = filename:join(rabbit_mnesia:dir(), "queues"),
Dirs = case rabbit_file:list_dir(QueuesDir) of
{ok, Entries} -> Entries;
{error, _} -> []
end,
[begin
File = filename:join([QueuesDir, Dir, "clean.dot"]),
case rabbit_file:read_term_file(File) of
{ok, Terms} -> ok = store(Dir, Terms);
{error, _} -> ok
end,
file:delete(File)
end || Dir <- Dirs],
ok
after
close_table()
end.
start_link() -> gen_server:start_link(?MODULE, [], []).
%%----------------------------------------------------------------------------
init(_) ->
process_flag(trap_exit, true),
open_table(),
{ok, undefined}.
handle_call(Msg, _, State) -> {stop, {unexpected_call, Msg}, State}.
handle_cast(Msg, State) -> {stop, {unexpected_cast, Msg}, State}.
handle_info(_Info, State) -> {noreply, State}.
terminate(_Reason, _State) ->
close_table().
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
%%----------------------------------------------------------------------------
open_table() ->
File = filename:join(rabbit_mnesia:dir(), "recovery.dets"),
{ok, _} = dets:open_file(?MODULE, [{file, File},
{ram_file, true},
{auto_save, infinity}]).
flush() -> ok = dets:sync(?MODULE).
close_table() ->
ok = flush(),
ok = dets:close(?MODULE).
|
