summaryrefslogtreecommitdiff
path: root/src/rabbit_channel.erl
diff options
context:
space:
mode:
Diffstat (limited to 'src/rabbit_channel.erl')
-rw-r--r--src/rabbit_channel.erl178
1 files changed, 41 insertions, 137 deletions
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index ec1d1fba..a9278898 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -37,7 +37,7 @@
transaction_id, tx_participants, next_tag,
uncommitted_ack_q, unacked_message_q,
username, virtual_host,
- most_recently_declared_queue, consumer_mapping, next_ticket}).
+ most_recently_declared_queue, consumer_mapping}).
%%----------------------------------------------------------------------------
@@ -94,8 +94,7 @@ init(ProxyPid, [ReaderPid, WriterPid, Username, VHost]) ->
username = Username,
virtual_host = VHost,
most_recently_declared_queue = <<>>,
- consumer_mapping = dict:new(),
- next_ticket = 101}.
+ consumer_mapping = dict:new()}.
handle_message({method, Method, Content}, State) ->
case (catch handle_method(Method, Content, State)) of
@@ -140,7 +139,6 @@ handle_message(Other, State) ->
terminate(Reason, State = #ch{writer_pid = WriterPid}) ->
Res = notify_queues(internal_rollback(State)),
- ok = rabbit_realm:leave_realms(self()),
case Reason of
normal -> ok = Res;
_ -> ok
@@ -195,14 +193,6 @@ die_precondition_failed(Fmt, Params) ->
rabbit_misc:protocol_error({false, 406, <<"PRECONDITION_FAILED">>},
Fmt, Params).
-check_ticket(TicketNumber, FieldIndex, Name, #ch{ username = Username}) ->
- rabbit_ticket:check_ticket(TicketNumber, FieldIndex, Name, Username).
-
-lookup_ticket(TicketNumber, FieldIndex,
- #ch{ username = Username, virtual_host = VHostPath }) ->
- rabbit_ticket:lookup_ticket(TicketNumber, FieldIndex,
- Username, VHostPath).
-
%% check that an exchange/queue name does not contain the reserved
%% "amq." prefix.
%%
@@ -235,57 +225,19 @@ handle_method(_Method, _, #ch{state = starting}) ->
handle_method(#'channel.close'{}, _, State = #ch{writer_pid = WriterPid}) ->
ok = notify_queues(internal_rollback(State)),
- ok = rabbit_realm:leave_realms(self()),
ok = rabbit_writer:send_command(WriterPid, #'channel.close_ok'{}),
ok = rabbit_writer:shutdown(WriterPid),
stop;
-handle_method(#'access.request'{realm = RealmNameBin,
- exclusive = Exclusive,
- passive = Passive,
- active = Active,
- write = Write,
- read = Read},
- _, State = #ch{username = Username,
- virtual_host = VHostPath,
- next_ticket = NextTicket}) ->
- RealmName = rabbit_misc:r(VHostPath, realm, RealmNameBin),
- Ticket = #ticket{realm_name = RealmName,
- passive_flag = Passive,
- active_flag = Active,
- write_flag = Write,
- read_flag = Read},
- case rabbit_realm:access_request(Username, Exclusive, Ticket) of
- ok ->
- rabbit_ticket:record_ticket(NextTicket, Ticket),
- NewState = State#ch{next_ticket = NextTicket + 1},
- {reply, #'access.request_ok'{ticket = NextTicket}, NewState};
- {error, not_found} ->
- rabbit_misc:protocol_error(
- invalid_path, "no ~s", [rabbit_misc:rs(RealmName)]);
- {error, bad_realm_path} ->
- %% FIXME: spec bug? access_refused is a soft error, spec requires it to be hard
- rabbit_misc:protocol_error(
- access_refused, "bad path for ~s", [rabbit_misc:rs(RealmName)]);
- {error, resource_locked} ->
- rabbit_misc:protocol_error(
- resource_locked, "~s is locked", [rabbit_misc:rs(RealmName)]);
- {error, access_refused} ->
- rabbit_misc:protocol_error(
- access_refused,
- "~w permissions denied for user '~s' attempting to access ~s",
- [rabbit_misc:permission_list(Ticket),
- Username, rabbit_misc:rs(RealmName)])
- end;
+handle_method(#'access.request'{},_, State) ->
+ {reply, #'access.request_ok'{ticket = 1}, State};
-handle_method(#'basic.publish'{ticket = TicketNumber,
- exchange = ExchangeNameBin,
+handle_method(#'basic.publish'{exchange = ExchangeNameBin,
routing_key = RoutingKey,
mandatory = Mandatory,
immediate = Immediate},
Content, State = #ch{ virtual_host = VHostPath}) ->
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
- check_ticket(TicketNumber, #ticket.write_flag, ExchangeName, State),
Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
%% We decode the content's properties here because we're almost
%% certain to want to look at delivery-mode and priority.
@@ -323,13 +275,11 @@ handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
uncommitted_ack_q = NewUAQ})
end};
-handle_method(#'basic.get'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'basic.get'{queue = QueueNameBin,
no_ack = NoAck},
_, State = #ch{ proxy_pid = ProxyPid, writer_pid = WriterPid,
next_tag = DeliveryTag }) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
- check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
case rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:basic_get(Q, ProxyPid, NoAck) end) of
@@ -352,8 +302,7 @@ handle_method(#'basic.get'{ticket = TicketNumber,
{reply, #'basic.get_empty'{cluster_id = <<>>}, State}
end;
-handle_method(#'basic.consume'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'basic.consume'{queue = QueueNameBin,
consumer_tag = ConsumerTag,
no_local = _, % FIXME: implement
no_ack = NoAck,
@@ -365,7 +314,6 @@ handle_method(#'basic.consume'{ticket = TicketNumber,
case dict:find(ConsumerTag, ConsumerMapping) of
error ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
- check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
ActualConsumerTag =
case ConsumerTag of
<<>> -> rabbit_misc:binstring_guid("amq.ctag");
@@ -391,7 +339,7 @@ handle_method(#'basic.consume'{ticket = TicketNumber,
ConsumerMapping)}};
{error, queue_owned_by_another_connection} ->
%% The spec is silent on which exception to use
- %% here. This seems reasonable?
+ %% here. This seems reasonable?
%% FIXME: check this
rabbit_misc:protocol_error(
@@ -495,8 +443,7 @@ handle_method(#'basic.recover'{}, _, _State) ->
rabbit_misc:protocol_error(
not_allowed, "attempt to recover a transactional channel",[]);
-handle_method(#'exchange.declare'{ticket = TicketNumber,
- exchange = ExchangeNameBin,
+handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
type = TypeNameBin,
passive = false,
durable = Durable,
@@ -505,17 +452,13 @@ handle_method(#'exchange.declare'{ticket = TicketNumber,
nowait = NoWait,
arguments = Args},
_, State = #ch{ virtual_host = VHostPath }) ->
- #ticket{realm_name = RealmName} =
- lookup_ticket(TicketNumber, #ticket.active_flag, State),
CheckedType = rabbit_exchange:check_type(TypeNameBin),
- %% FIXME: clarify spec as per declare wrt differing realms
- X = case rabbit_exchange:lookup(
- rabbit_misc:r(VHostPath, exchange, ExchangeNameBin)) of
+ ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
+ X = case rabbit_exchange:lookup(ExchangeName) of
{ok, FoundX} -> FoundX;
{error, not_found} ->
- ActualNameBin = check_name('exchange', ExchangeNameBin),
- rabbit_exchange:declare(RealmName,
- ActualNameBin,
+ check_name('exchange', ExchangeNameBin),
+ rabbit_exchange:declare(ExchangeName,
CheckedType,
Durable,
AutoDelete,
@@ -524,26 +467,21 @@ handle_method(#'exchange.declare'{ticket = TicketNumber,
ok = rabbit_exchange:assert_type(X, CheckedType),
return_ok(State, NoWait, #'exchange.declare_ok'{});
-handle_method(#'exchange.declare'{ticket = TicketNumber,
- exchange = ExchangeNameBin,
+handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
type = TypeNameBin,
passive = true,
nowait = NoWait},
_, State = #ch{ virtual_host = VHostPath }) ->
- %% FIXME: spec issue: permit active_flag here as well as passive_flag?
- #ticket{} = lookup_ticket(TicketNumber, #ticket.passive_flag, State),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
X = rabbit_exchange:lookup_or_die(ExchangeName),
ok = rabbit_exchange:assert_type(X, rabbit_exchange:check_type(TypeNameBin)),
return_ok(State, NoWait, #'exchange.declare_ok'{});
-handle_method(#'exchange.delete'{ticket = TicketNumber,
- exchange = ExchangeNameBin,
+handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
if_unused = IfUnused,
nowait = NoWait},
_, State = #ch { virtual_host = VHostPath }) ->
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
- check_ticket(TicketNumber, #ticket.active_flag, ExchangeName, State),
case rabbit_exchange:delete(ExchangeName, IfUnused) of
{error, not_found} ->
rabbit_misc:protocol_error(
@@ -555,8 +493,7 @@ handle_method(#'exchange.delete'{ticket = TicketNumber,
return_ok(State, NoWait, #'exchange.delete_ok'{})
end;
-handle_method(#'queue.declare'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'queue.declare'{queue = QueueNameBin,
passive = false,
durable = Durable,
exclusive = ExclusiveDeclare,
@@ -565,8 +502,6 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
arguments = Args},
_, State = #ch { virtual_host = VHostPath,
reader_pid = ReaderPid }) ->
- #ticket{realm_name = RealmName} =
- lookup_ticket(TicketNumber, #ticket.active_flag, State),
%% FIXME: atomic create&claim
Finish =
fun (Q) ->
@@ -587,7 +522,6 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
end,
Q
end,
- %% FIXME: clarify spec as per declare wrt differing realms
Q = case rabbit_amqqueue:with(
rabbit_misc:r(VHostPath, queue, QueueNameBin),
Finish) of
@@ -597,34 +531,28 @@ handle_method(#'queue.declare'{ticket = TicketNumber,
<<>> -> rabbit_misc:binstring_guid("amq.gen");
Other -> check_name('queue', Other)
end,
- Finish(rabbit_amqqueue:declare(RealmName,
- ActualNameBin,
- Durable,
- AutoDelete,
- Args));
+ QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin),
+ Finish(rabbit_amqqueue:declare(QueueName,
+ Durable, AutoDelete, Args));
Other -> Other
end,
return_queue_declare_ok(State, NoWait, Q);
-handle_method(#'queue.declare'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'queue.declare'{queue = QueueNameBin,
passive = true,
nowait = NoWait},
_, State = #ch{ virtual_host = VHostPath }) ->
- #ticket{} = lookup_ticket(TicketNumber, #ticket.passive_flag, State),
QueueName = rabbit_misc:r(VHostPath, queue, QueueNameBin),
Q = rabbit_amqqueue:with_or_die(QueueName, fun (Q) -> Q end),
return_queue_declare_ok(State, NoWait, Q);
-handle_method(#'queue.delete'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'queue.delete'{queue = QueueNameBin,
if_unused = IfUnused,
if_empty = IfEmpty,
nowait = NoWait
},
_, State) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
- check_ticket(TicketNumber, #ticket.active_flag, QueueName, State),
case rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:delete(Q, IfUnused, IfEmpty) end) of
@@ -640,8 +568,7 @@ handle_method(#'queue.delete'{ticket = TicketNumber,
message_count = PurgedMessageCount})
end;
-handle_method(#'queue.bind'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'queue.bind'{queue = QueueNameBin,
exchange = ExchangeNameBin,
routing_key = RoutingKey,
nowait = NoWait,
@@ -652,14 +579,13 @@ handle_method(#'queue.bind'{ticket = TicketNumber,
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
ActualRoutingKey = expand_routing_key_shortcut(QueueNameBin, RoutingKey,
State),
- check_ticket(TicketNumber, #ticket.active_flag, QueueName, State),
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
case rabbit_amqqueue:add_binding(QueueName, ExchangeName,
ActualRoutingKey, Arguments) of
- {error, queue_not_found} ->
+ {error, queue_not_found} ->
rabbit_misc:protocol_error(
not_found, "no ~s", [rabbit_misc:rs(QueueName)]);
- {error, exchange_not_found} ->
+ {error, exchange_not_found} ->
rabbit_misc:protocol_error(
not_found, "no ~s", [rabbit_misc:rs(ExchangeName)]);
{error, durability_settings_incompatible} ->
@@ -670,12 +596,10 @@ handle_method(#'queue.bind'{ticket = TicketNumber,
return_ok(State, NoWait, #'queue.bind_ok'{})
end;
-handle_method(#'queue.purge'{ticket = TicketNumber,
- queue = QueueNameBin,
+handle_method(#'queue.purge'{queue = QueueNameBin,
nowait = NoWait},
_, State) ->
QueueName = expand_queue_name_shortcut(QueueNameBin, State),
- check_ticket(TicketNumber, #ticket.read_flag, QueueName, State),
{ok, PurgedMessageCount} = rabbit_amqqueue:with_or_die(
QueueName,
fun (Q) -> rabbit_amqqueue:purge(Q) end),
@@ -783,21 +707,6 @@ ack(ProxyPid, TxnKey, UAQ) ->
make_tx_id() -> rabbit_misc:guid().
-safe_pmap_set_ok(F, S) ->
- case lists:filter(fun (R) -> R =/= ok end,
- rabbit_misc:upmap(
- fun (V) ->
- try F(V)
- catch Class:Reason -> {Class, Reason}
- end
- end, sets:to_list(S))) of
- [] -> ok;
- Errors -> {error, Errors}
- end.
-
-notify_participants(F, TxnKey, Participants) ->
- safe_pmap_set_ok(fun (QPid) -> F(QPid, TxnKey) end, Participants).
-
new_tx(State) ->
State#ch{transaction_id = make_tx_id(),
tx_participants = sets:new(),
@@ -805,8 +714,8 @@ new_tx(State) ->
internal_commit(State = #ch{transaction_id = TxnKey,
tx_participants = Participants}) ->
- case notify_participants(fun rabbit_amqqueue:commit/2,
- TxnKey, Participants) of
+ case rabbit_amqqueue:commit_all(sets:to_list(Participants),
+ TxnKey) of
ok -> new_tx(State);
{error, Errors} -> exit({commit_failed, Errors})
end.
@@ -819,8 +728,8 @@ internal_rollback(State = #ch{transaction_id = TxnKey,
[self(),
queue:len(UAQ),
queue:len(UAMQ)]),
- case notify_participants(fun rabbit_amqqueue:rollback/2,
- TxnKey, Participants) of
+ case rabbit_amqqueue:rollback_all(sets:to_list(Participants),
+ TxnKey) of
ok -> NewUAMQ = queue:join(UAQ, UAMQ),
new_tx(State#ch{unacked_message_q = NewUAMQ});
{error, Errors} -> exit({rollback_failed, Errors})
@@ -843,23 +752,18 @@ fold_per_queue(F, Acc0, UAQ) ->
Acc0, D).
notify_queues(#ch{proxy_pid = ProxyPid, consumer_mapping = Consumers}) ->
- safe_pmap_set_ok(
- fun (QueueName) ->
- case rabbit_amqqueue:with(
- QueueName,
- fun (Q) ->
- rabbit_amqqueue:notify_down(Q, ProxyPid)
- end) of
- ok ->
- ok;
- {error, not_found} ->
- %% queue has been deleted in the meantime
- ok
- end
- end,
- dict:fold(fun (_ConsumerTag, QueueName, S) ->
- sets:add_element(QueueName, S)
- end, sets:new(), Consumers)).
+ rabbit_amqqueue:notify_down_all(
+ [QPid || QueueName <-
+ sets:to_list(
+ dict:fold(fun (_ConsumerTag, QueueName, S) ->
+ sets:add_element(QueueName, S)
+ end, sets:new(), Consumers)),
+ case rabbit_amqqueue:lookup(QueueName) of
+ {ok, Q} -> QPid = Q#amqqueue.pid, true;
+ %% queue has been deleted in the meantime
+ {error, not_found} -> QPid = none, false
+ end],
+ ProxyPid).
is_message_persistent(#content{properties = #'P_basic'{
delivery_mode = Mode}}) ->
View Lorry Controller Status