Skip to content

Add rabbit_queue_type behaviour #2052

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 18 commits into from
Closed

Add rabbit_queue_type behaviour #2052

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
7fa38c0
Add marker rabbit_queue_type behaviour
kjnilsson Jul 2, 2019
e5dff9f
Move queue declare into rabbit_queue_type
kjnilsson Jul 3, 2019
8aa9d9f
Move queue delete into queue type implementation
kjnilsson Jul 3, 2019
1e8500b
First cut stateful queue type
kjnilsson Jul 3, 2019
7a7bf15
Queue type: dequeue/basic_get
kjnilsson Jul 15, 2019
7d34763
Move info inside queue type abstraction
kjnilsson Jul 16, 2019
f0aaa63
Move policy change into queue type interface
kjnilsson Jul 18, 2019
4a91b5d
Add purge to queue type
kjnilsson Jul 18, 2019
5277cbb
Add recovery to the queue type interface
kjnilsson Jul 22, 2019
3603048
Rename amqqueue quorum_nodes field
kjnilsson Jul 24, 2019
2e3aaf4
Fix tests and handle classic API response
dcorbacho Jul 24, 2019
277840b
Fix HA queue confirm bug
kjnilsson Jul 25, 2019
4f6ebd8
Fix issue with events of deleted queues
kjnilsson Jul 25, 2019
e489655
be defensive around state_info
kjnilsson Jul 25, 2019
550b980
fix basic.get reject bug
kjnilsson Jul 25, 2019
5b6bcec
correct default value of amqqueue getter
kjnilsson Jul 25, 2019
899beb3
wip
kjnilsson Jul 26, 2019
1465bb6
wip
kjnilsson Aug 5, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions include/amqqueue.hrl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,16 +51,16 @@
(?is_amqqueue_v1(Q) andalso
?amqqueue_v1_field_state(Q) =:= State))).

-define(amqqueue_v1_type, classic).
-define(amqqueue_v1_type, rabbit_classic_queue).

-define(amqqueue_is_classic(Q),
((?is_amqqueue_v2(Q) andalso
?amqqueue_v2_field_type(Q) =:= classic) orelse
?amqqueue_v2_field_type(Q) =:= rabbit_classic_queue) orelse
?is_amqqueue_v1(Q))).
Copy link
Collaborator

@dumbbell dumbbell Jul 29, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To follow up on Diana's comments, we can probably replace those functions & associated macros by something more generic. @lukebakken did change the initial generic functions/macros to those specific more readable versions. @lukebakken: I think you linked an article about that kind of API but I don't rember the content, what was the reason behind this already?

Copy link
Collaborator

@lukebakken lukebakken Jul 29, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My goal was to put all "knowledge" of how to determine if an amqqueue is classic into this macro. This way, when the identification of a classic queue changes, we only have to change this macro and not all the places where it is used.

Copy link
Collaborator

@lukebakken lukebakken Jul 29, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"Tell Don't Ask"


-define(amqqueue_is_quorum(Q),
(?is_amqqueue_v2(Q) andalso
?amqqueue_v2_field_type(Q) =:= quorum) orelse
?amqqueue_v2_field_type(Q) =:= rabbit_quorum_queue) orelse
false).

-define(amqqueue_has_valid_pid(Q),
Expand Down
38 changes: 19 additions & 19 deletions src/amqqueue.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,9 +57,9 @@
% policy_version
get_policy_version/1,
set_policy_version/2,
% quorum_nodes
get_quorum_nodes/1,
set_quorum_nodes/2,
% type_state
get_type_state/1,
set_type_state/2,
% recoverable_slaves
get_recoverable_slaves/1,
set_recoverable_slaves/2,
Expand Down Expand Up @@ -118,8 +118,8 @@
slave_pids_pending_shutdown = [] :: [pid()] | '_',
vhost :: rabbit_types:vhost() | undefined | '_', %% secondary index
options = #{} :: map() | '_',
type = ?amqqueue_v1_type :: atom() | '_',
quorum_nodes = [] :: [node()] | '_'
type = ?amqqueue_v1_type :: module() | '_',
type_state = #{} :: map() | '_'
}).

-type amqqueue() :: amqqueue_v1:amqqueue_v1() | amqqueue_v2().
Expand All @@ -143,7 +143,7 @@
vhost :: rabbit_types:vhost() | undefined,
options :: map(),
type :: atom(),
quorum_nodes :: [node()]
type_state :: #{}
}.

-type ra_server_id() :: {Name :: atom(), Node :: node()}.
Expand All @@ -170,7 +170,7 @@
vhost :: '_',
options :: '_',
type :: atom() | '_',
quorum_nodes :: '_'
type_state :: '_'
}.

-export_type([amqqueue/0,
Expand Down Expand Up @@ -451,7 +451,7 @@ set_gm_pids(Queue, GMPids) ->

-spec get_leader(amqqueue_v2()) -> node().

get_leader(#amqqueue{type = quorum, pid = {_, Leader}}) -> Leader.
get_leader(#amqqueue{type = rabbit_quorum_queue, pid = {_, Leader}}) -> Leader.
Copy link
Collaborator

@dumbbell dumbbell Jul 29, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should the "PID" be an opaque type as well?

Again, that "leader" thing is very quorum queues/Ra specific.


% operator_policy

Expand Down Expand Up @@ -551,18 +551,18 @@ set_recoverable_slaves(#amqqueue{} = Queue, Slaves) ->
set_recoverable_slaves(Queue, Slaves) ->
amqqueue_v1:set_recoverable_slaves(Queue, Slaves).

% quorum_nodes (new in v2)
% type_state (new in v2)

-spec get_quorum_nodes(amqqueue()) -> [node()].
-spec get_type_state(amqqueue()) -> map().
get_type_state(#amqqueue{type_state = TState}) ->
TState;
get_type_state(_) ->
#{}.

get_quorum_nodes(#amqqueue{quorum_nodes = Nodes}) -> Nodes;
get_quorum_nodes(_) -> [].

-spec set_quorum_nodes(amqqueue(), [node()]) -> amqqueue().

set_quorum_nodes(#amqqueue{} = Queue, Nodes) ->
Queue#amqqueue{quorum_nodes = Nodes};
set_quorum_nodes(Queue, _Nodes) ->
-spec set_type_state(amqqueue(), map()) -> amqqueue().
set_type_state(#amqqueue{} = Queue, TState) ->
Queue#amqqueue{type_state = TState};
set_type_state(Queue, _TState) ->
Queue.

% slave_pids
Expand Down Expand Up @@ -660,7 +660,7 @@ is_classic(Queue) ->
-spec is_quorum(amqqueue()) -> boolean().

is_quorum(Queue) ->
get_type(Queue) =:= quorum.
get_type(Queue) =:= rabbit_quorum_queue.

fields() ->
case record_version_to_use() of
Expand Down
23 changes: 12 additions & 11 deletions src/amqqueue_v1.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,9 +57,9 @@
% policy_version
get_policy_version/1,
set_policy_version/2,
% quorum_nodes
get_quorum_nodes/1,
set_quorum_nodes/2,
% type_state
get_type_state/1,
set_type_state/2,
% recoverable_slaves
get_recoverable_slaves/1,
set_recoverable_slaves/2,
Expand Down Expand Up @@ -449,16 +449,16 @@ get_recoverable_slaves(#amqqueue{recoverable_slaves = Slaves}) ->
set_recoverable_slaves(#amqqueue{} = Queue, Slaves) ->
Queue#amqqueue{recoverable_slaves = Slaves}.

% quorum_nodes (new in v2)
% type_state (new in v2)

-spec get_quorum_nodes(amqqueue()) -> no_return().
-spec get_type_state(amqqueue()) -> no_return().

get_quorum_nodes(_) -> throw({unsupported, ?record_version, get_quorum_nodes}).
get_type_state(_) -> throw({unsupported, ?record_version, get_type_state}).

-spec set_quorum_nodes(amqqueue(), [node()]) -> no_return().
-spec set_type_state(amqqueue(), [node()]) -> no_return().

set_quorum_nodes(_, _) ->
throw({unsupported, ?record_version, set_quorum_nodes}).
set_type_state(_, _) ->
throw({unsupported, ?record_version, set_type_state}).

% slave_pids

Expand Down Expand Up @@ -525,8 +525,9 @@ is_classic(Queue) ->

-spec is_quorum(amqqueue()) -> boolean().

is_quorum(Queue) ->
get_type(Queue) =:= quorum.
is_quorum(Queue) when ?is_amqqueue(Queue) ->
false.
% get_type(Queue) =:= rabbit_quorum_queue.

fields() -> fields(?record_version).

Expand Down
Loading