Merge pull request #11667 from rabbitmq/md/khepri-projections-wrap-ets-calls

rabbit_db_*: Wrap `ets` calls to projections in `whereis/1` checks

Author: the-mikedavis

deps/rabbit/src/rabbit_db_binding.erl

@@ -53,6 +53,8 @@
 -define(MNESIA_SEMI_DURABLE_TABLE, rabbit_semi_durable_route).
 -define(MNESIA_REVERSE_TABLE, rabbit_reverse_route).
 -define(MNESIA_INDEX_TABLE, rabbit_index_route).
+-define(KHEPRI_BINDINGS_PROJECTION, rabbit_khepri_bindings).
+-define(KHEPRI_INDEX_ROUTE_PROJECTION, rabbit_khepri_index_route).
 
 %% -------------------------------------------------------------------
 %% exists().
@@ -411,7 +413,12 @@ get_all_in_mnesia() ->
       end).
 
 get_all_in_khepri() ->
-     [B || #route{binding = B} <- ets:tab2list(rabbit_khepri_bindings)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            [B || #route{binding = B} <- ets:tab2list(Table)]
+    end.
 
 -spec get_all(VHostName) -> [Binding] when
       VHostName :: vhost:name(),
@@ -437,11 +444,16 @@ get_all_in_mnesia(VHost) ->
     [B || #route{binding = B} <- rabbit_db:list_in_mnesia(?MNESIA_TABLE, Match)].
 
 get_all_in_khepri(VHost) ->
-    VHostResource = rabbit_misc:r(VHost, '_'),
-    Match = #route{binding = #binding{source      = VHostResource,
-                                      destination = VHostResource,
-                                      _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Match)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            VHostResource = rabbit_misc:r(VHost, '_'),
+            Match = #route{binding = #binding{source      = VHostResource,
+                                              destination = VHostResource,
+                                              _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Match)]
+    end.
 
 -spec get_all(Src, Dst, Reverse) -> [Binding] when
       Src :: rabbit_types:binding_source(),
@@ -469,10 +481,15 @@ get_all_in_mnesia(SrcName, DstName, Reverse) ->
     mnesia:async_dirty(Fun).
 
 get_all_in_khepri(SrcName, DstName) ->
-    MatchHead = #route{binding = #binding{source      = SrcName,
-                                          destination = DstName,
-                                          _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, MatchHead)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            MatchHead = #route{binding = #binding{source      = SrcName,
+                                                  destination = DstName,
+                                                  _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, MatchHead)]
+    end.
 
 %% -------------------------------------------------------------------
 %% get_all_for_source().
@@ -511,8 +528,13 @@ list_for_route(Route, true) ->
     end.
 
 get_all_for_source_in_khepri(Resource) ->
-    Route = #route{binding = #binding{source = Resource, _ = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Route)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            Route = #route{binding = #binding{source = Resource, _ = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Route)]
+    end.
 
 %% -------------------------------------------------------------------
 %% get_all_for_destination().
@@ -541,9 +563,14 @@ get_all_for_destination_in_mnesia(Dst) ->
     mnesia:async_dirty(Fun).
 
 get_all_for_destination_in_khepri(Destination) ->
-    Match = #route{binding = #binding{destination = Destination,
-                                      _           = '_'}},
-    [B || #route{binding = B} <- ets:match_object(rabbit_khepri_bindings, Match)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            Match = #route{binding = #binding{destination = Destination,
+                                              _           = '_'}},
+            [B || #route{binding = B} <- ets:match_object(Table, Match)]
+    end.
 
 %% -------------------------------------------------------------------
 %% fold().
@@ -617,11 +644,16 @@ match_in_mnesia(SrcName, Match) ->
                  Routes, Match(Binding)].
 
 match_in_khepri(SrcName, Match) ->
-    MatchHead = #route{binding = #binding{source      = SrcName,
-                                          _           = '_'}},
-    Routes = ets:select(rabbit_khepri_bindings, [{MatchHead, [], [['$_']]}]),
-    [Dest || [#route{binding = Binding = #binding{destination = Dest}}] <-
-                 Routes, Match(Binding)].
+    case ets:whereis(?KHEPRI_BINDINGS_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            MatchHead = #route{binding = #binding{source      = SrcName,
+                                                  _           = '_'}},
+            Routes = ets:select(Table, [{MatchHead, [], [['$_']]}]),
+            [Dest || [#route{binding = Binding = #binding{destination = Dest}}] <-
+                         Routes, Match(Binding)]
+    end.
 
 %% Routing - HOT CODE PATH
 %% -------------------------------------------------------------------
@@ -654,18 +686,26 @@ match_routing_key_in_mnesia(SrcName, RoutingKeys, UseIndex) ->
             route_in_mnesia_v1(SrcName, RoutingKeys)
     end.
 
-match_routing_key_in_khepri(Src, ['_']) ->
+match_routing_key_in_khepri(Src, RoutingKeys) ->
+    case ets:whereis(?KHEPRI_INDEX_ROUTE_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            do_match_routing_key_in_khepri(Table, Src, RoutingKeys)
+    end.
+
+do_match_routing_key_in_khepri(Table, Src, ['_']) ->
     MatchHead = #index_route{source_key  = {Src, '_'},
                              destination = '$1',
                              _           = '_'},
-    ets:select(rabbit_khepri_index_route, [{MatchHead, [], ['$1']}]);
+    ets:select(Table, [{MatchHead, [], ['$1']}]);
 
-match_routing_key_in_khepri(Src, RoutingKeys) ->
+do_match_routing_key_in_khepri(Table, Src, RoutingKeys) ->
     lists:foldl(
       fun(RK, Acc) ->
               try
                   Dst = ets:lookup_element(
-                          rabbit_khepri_index_route,
+                          Table,
                           {Src, RK},
                           #index_route.destination),
                   Dst ++ Acc

deps/rabbit/src/rabbit_db_exchange.erl

@@ -57,6 +57,7 @@
 -define(MNESIA_TABLE, rabbit_exchange).
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_exchange).
 -define(MNESIA_SERIAL_TABLE, rabbit_exchange_serial).
+-define(KHEPRI_PROJECTION, rabbit_khepri_exchange).
 
 %% -------------------------------------------------------------------
 %% get_all().
@@ -182,9 +183,14 @@ get_in_mnesia(Name) ->
     rabbit_mnesia:dirty_read({?MNESIA_TABLE, Name}).
 
 get_in_khepri(Name) ->
-    case ets:lookup(rabbit_khepri_exchange, Name) of
-        [X] -> {ok, X};
-        []  -> {error, not_found}
+    case ets:whereis(?KHEPRI_PROJECTION) of
+        undefined ->
+            {error, not_found};
+        Table ->
+            case ets:lookup(Table, Name) of
+                [X] -> {ok, X};
+                []  -> {error, not_found}
+            end
     end.
 
 %% -------------------------------------------------------------------
@@ -227,7 +233,12 @@ get_many_in_mnesia(Table, Names) when is_list(Names) ->
     lists:append([ets:lookup(Table, Name) || Name <- Names]).
 
 get_many_in_khepri(Names) when is_list(Names) ->
-    lists:append([ets:lookup(rabbit_khepri_exchange, Name) || Name <- Names]).
+    case ets:whereis(?KHEPRI_PROJECTION) of
+        undefined ->
+            [];
+        Table ->
+            lists:append([ets:lookup(Table, Name) || Name <- Names])
+    end.
 
 %% -------------------------------------------------------------------
 %% count().