summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMatthias Radestock <matthias@rabbitmq.com>2011-06-24 22:36:22 +0100
committerMatthias Radestock <matthias@rabbitmq.com>2011-06-24 22:36:22 +0100
commitbee6e92332ca8f3de1f7b043e04dd2fa4eea5eab (patch)
tree404949c84beb96f54b47021213de699e22310e1b
parent3c63aba85361beae3f5733574fd470222e4f3a8c (diff)
downloadrabbitmq-server-bee6e92332ca8f3de1f7b043e04dd2fa4eea5eab.tar.gz
tx gone from everywhere
-rw-r--r--docs/rabbitmqctl.1.xml12
-rw-r--r--include/rabbit.hrl3
-rw-r--r--src/rabbit_amqqueue.erl23
-rw-r--r--src/rabbit_amqqueue_process.erl7
-rw-r--r--src/rabbit_basic.erl32
-rw-r--r--src/rabbit_channel.erl141
-rw-r--r--src/rabbit_control.erl2
-rw-r--r--src/rabbit_error_logger.erl2
-rw-r--r--src/rabbit_mirror_queue_coordinator.erl39
-rw-r--r--src/rabbit_mirror_queue_slave.erl10
-rw-r--r--src/rabbit_queue_index.erl18
-rw-r--r--src/rabbit_tests.erl15
-rw-r--r--src/rabbit_types.erl6
13 files changed, 67 insertions, 243 deletions
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index a0f03192..fdb49912 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -1196,10 +1196,6 @@
<listitem><para>Virtual host in which the channel operates.</para></listitem>
</varlistentry>
<varlistentry>
- <term>transactional</term>
- <listitem><para>True if the channel is in transactional mode, false otherwise.</para></listitem>
- </varlistentry>
- <varlistentry>
<term>consumer_count</term>
<listitem><para>Number of logical AMQP consumers retrieving messages via
the channel.</para></listitem>
@@ -1210,11 +1206,6 @@
yet acknowledged.</para></listitem>
</varlistentry>
<varlistentry>
- <term>acks_uncommitted</term>
- <listitem><para>Number of acknowledgements received in an as yet
- uncommitted transaction.</para></listitem>
- </varlistentry>
- <varlistentry>
<term>prefetch_count</term>
<listitem><para>QoS prefetch count limit in force, 0 if unlimited.</para></listitem>
</varlistentry>
@@ -1239,8 +1230,7 @@
</variablelist>
<para>
If no <command>channelinfoitem</command>s are specified then pid,
- user, transactional, consumer_count, and
- messages_unacknowledged are assumed.
+ user, consumer_count, and messages_unacknowledged are assumed.
</para>
<para role="example-prefix">
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 00b7e6e9..3861df2a 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -67,8 +67,7 @@
is_persistent}).
-record(ssl_socket, {tcp, ssl}).
--record(delivery, {mandatory, immediate, txn, sender, message,
- msg_seq_no}).
+-record(delivery, {mandatory, immediate, sender, message, msg_seq_no}).
-record(amqp_error, {name, explanation = "", method = none}).
-record(event, {type, props, timestamp}).
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index bacb1d21..4d6aaa18 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -20,12 +20,12 @@
-export([pseudo_queue/2]).
-export([lookup/1, with/2, with_or_die/2, assert_equivalence/5,
check_exclusive_access/2, with_exclusive_access_or_die/3,
- stat/1, deliver/2, requeue/3, ack/4, reject/4]).
+ stat/1, deliver/2, requeue/3, ack/3, reject/4]).
-export([list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2]).
-export([consumers/1, consumers_all/1, consumer_info_keys/0]).
-export([basic_get/3, basic_consume/7, basic_cancel/4]).
-export([notify_sent/2, unblock/2, flush_all/2]).
--export([commit_all/3, rollback_all/3, notify_down_all/2, limit_all/3]).
+-export([notify_down_all/2, limit_all/3]).
-export([on_node_down/1]).
-export([store_queue/1]).
@@ -117,12 +117,8 @@
-spec(purge/1 :: (rabbit_types:amqqueue()) -> qlen()).
-spec(deliver/2 :: (pid(), rabbit_types:delivery()) -> boolean()).
-spec(requeue/3 :: (pid(), [msg_id()], pid()) -> 'ok').
--spec(ack/4 ::
- (pid(), rabbit_types:maybe(rabbit_types:txn()), [msg_id()], pid())
- -> 'ok').
+-spec(ack/3 :: (pid(), [msg_id()], pid()) -> 'ok').
-spec(reject/4 :: (pid(), [msg_id()], boolean(), pid()) -> 'ok').
--spec(commit_all/3 :: ([pid()], rabbit_types:txn(), pid()) -> ok_or_errors()).
--spec(rollback_all/3 :: ([pid()], rabbit_types:txn(), pid()) -> 'ok').
-spec(notify_down_all/2 :: ([pid()], pid()) -> ok_or_errors()).
-spec(limit_all/3 :: ([pid()], pid(), pid() | 'undefined') -> ok_or_errors()).
-spec(basic_get/3 :: (rabbit_types:amqqueue(), pid(), boolean()) ->
@@ -436,21 +432,12 @@ deliver(QPid, Delivery) ->
requeue(QPid, MsgIds, ChPid) ->
delegate_call(QPid, {requeue, MsgIds, ChPid}).
-ack(QPid, Txn, MsgIds, ChPid) ->
- delegate_cast(QPid, {ack, Txn, MsgIds, ChPid}).
+ack(QPid, MsgIds, ChPid) ->
+ delegate_cast(QPid, {ack, MsgIds, ChPid}).
reject(QPid, MsgIds, Requeue, ChPid) ->
delegate_cast(QPid, {reject, MsgIds, Requeue, ChPid}).
-commit_all(QPids, Txn, ChPid) ->
- safe_delegate_call_ok(
- fun (QPid) -> gen_server2:call(QPid, {commit, Txn, ChPid}, infinity) end,
- QPids).
-
-rollback_all(QPids, Txn, ChPid) ->
- delegate:invoke_no_result(
- QPids, fun (QPid) -> gen_server2:cast(QPid, {rollback, Txn, ChPid}) end).
-
notify_down_all(QPids, ChPid) ->
safe_delegate_call_ok(
fun (QPid) -> gen_server2:call(QPid, {notify_down, ChPid}, infinity) end,
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 28fced98..87cdf925 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -495,8 +495,7 @@ run_message_queue(State) ->
{_IsEmpty1, State2} = deliver_msgs_to_consumers(Funs, IsEmpty, State1),
State2.
-attempt_delivery(Delivery = #delivery{txn = none,
- sender = ChPid,
+attempt_delivery(Delivery = #delivery{sender = ChPid,
message = Message,
msg_seq_no = MsgSeqNo},
State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
@@ -801,7 +800,7 @@ prioritise_cast(Msg, _State) ->
maybe_expire -> 8;
drop_expired -> 8;
emit_stats -> 7;
- {ack, _Txn, _AckTags, _ChPid} -> 7;
+ {ack, _AckTags, _ChPid} -> 7;
{reject, _AckTags, _Requeue, _ChPid} -> 7;
{notify_sent, _ChPid} -> 7;
{unblock, _ChPid} -> 7;
@@ -1028,7 +1027,7 @@ handle_cast({deliver, Delivery}, State) ->
%% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
noreply(deliver_or_enqueue(Delivery, State));
-handle_cast({ack, none, AckTags, ChPid},
+handle_cast({ack, AckTags, ChPid},
State = #q{backing_queue = BQ, backing_queue_state = BQS}) ->
case lookup_ch(ChPid) of
not_found ->
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index fa7e3a5a..ec8ed351 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -18,8 +18,8 @@
-include("rabbit.hrl").
-include("rabbit_framing.hrl").
--export([publish/1, message/3, message/4, properties/1, delivery/5]).
--export([publish/4, publish/7]).
+-export([publish/1, message/3, message/4, properties/1, delivery/4]).
+-export([publish/4, publish/6]).
-export([build_content/2, from_content/1]).
%%----------------------------------------------------------------------------
@@ -37,9 +37,8 @@
-spec(publish/1 ::
(rabbit_types:delivery()) -> publish_result()).
--spec(delivery/5 ::
- (boolean(), boolean(), rabbit_types:maybe(rabbit_types:txn()),
- rabbit_types:message(), undefined | integer()) ->
+-spec(delivery/4 ::
+ (boolean(), boolean(), rabbit_types:message(), undefined | integer()) ->
rabbit_types:delivery()).
-spec(message/4 ::
(rabbit_exchange:name(), rabbit_router:routing_key(),
@@ -53,10 +52,9 @@
-spec(publish/4 ::
(exchange_input(), rabbit_router:routing_key(), properties_input(),
body_input()) -> publish_result()).
--spec(publish/7 ::
+-spec(publish/6 ::
(exchange_input(), rabbit_router:routing_key(), boolean(), boolean(),
- rabbit_types:maybe(rabbit_types:txn()), properties_input(),
- body_input()) -> publish_result()).
+ properties_input(), body_input()) -> publish_result()).
-spec(build_content/2 :: (rabbit_framing:amqp_property_record(),
binary() | [binary()]) -> rabbit_types:content()).
-spec(from_content/1 :: (rabbit_types:content()) ->
@@ -73,9 +71,9 @@ publish(Delivery = #delivery{
Other -> Other
end.
-delivery(Mandatory, Immediate, Txn, Message, MsgSeqNo) ->
- #delivery{mandatory = Mandatory, immediate = Immediate, txn = Txn,
- sender = self(), message = Message, msg_seq_no = MsgSeqNo}.
+delivery(Mandatory, Immediate, Message, MsgSeqNo) ->
+ #delivery{mandatory = Mandatory, immediate = Immediate, sender = self(),
+ message = Message, msg_seq_no = MsgSeqNo}.
build_content(Properties, BodyBin) when is_binary(BodyBin) ->
build_content(Properties, [BodyBin]);
@@ -157,19 +155,17 @@ indexof([_ | Rest], Element, N) -> indexof(Rest, Element, N + 1).
%% Convenience function, for avoiding round-trips in calls across the
%% erlang distributed network.
publish(Exchange, RoutingKeyBin, Properties, Body) ->
- publish(Exchange, RoutingKeyBin, false, false, none, Properties,
- Body).
+ publish(Exchange, RoutingKeyBin, false, false, Properties, Body).
%% Convenience function, for avoiding round-trips in calls across the
%% erlang distributed network.
-publish(X = #exchange{name = XName}, RKey, Mandatory, Immediate, Txn,
- Props, Body) ->
- publish(X, delivery(Mandatory, Immediate, Txn,
+publish(X = #exchange{name = XName}, RKey, Mandatory, Immediate, Props, Body) ->
+ publish(X, delivery(Mandatory, Immediate,
message(XName, RKey, properties(Props), Body),
undefined));
-publish(XName, RKey, Mandatory, Immediate, Txn, Props, Body) ->
+publish(XName, RKey, Mandatory, Immediate, Props, Body) ->
case rabbit_exchange:lookup(XName) of
- {ok, X} -> publish(X, RKey, Mandatory, Immediate, Txn, Props, Body);
+ {ok, X} -> publish(X, RKey, Mandatory, Immediate, Props, Body);
Err -> Err
end.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 991b0b06..36471bf5 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -30,8 +30,7 @@
prioritise_cast/2]).
-record(ch, {state, protocol, channel, reader_pid, writer_pid, conn_pid,
- limiter_pid, start_limiter_fun, transaction_id, tx_participants,
- next_tag, uncommitted_ack_q, unacked_message_q,
+ limiter_pid, start_limiter_fun, next_tag, unacked_message_q,
user, virtual_host, most_recently_declared_queue,
consumer_mapping, blocking, consumer_monitors, queue_collector_pid,
stats_timer, confirm_enabled, publish_seqno, unconfirmed_mq,
@@ -41,12 +40,10 @@
-define(STATISTICS_KEYS,
[pid,
- transactional,
confirm,
consumer_count,
messages_unacknowledged,
messages_unconfirmed,
- acks_uncommitted,
prefetch_count,
client_flow_blocked]).
@@ -173,10 +170,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, Protocol, User, VHost,
conn_pid = ConnPid,
limiter_pid = undefined,
start_limiter_fun = StartLimiterFun,
- transaction_id = none,
- tx_participants = sets:new(),
next_tag = 1,
- uncommitted_ack_q = queue:new(),
unacked_message_q = queue:new(),
user = User,
virtual_host = VHost,
@@ -331,7 +325,7 @@ handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
{hibernate, State#ch{stats_timer = StatsTimer1}}.
terminate(Reason, State) ->
- {Res, _State1} = rollback_and_notify(State),
+ {Res, _State1} = notify_queues(State),
case Reason of
normal -> ok = Res;
shutdown -> ok = Res;
@@ -386,8 +380,8 @@ send_exception(Reason, State = #ch{protocol = Protocol,
rabbit_binary_generator:map_exception(Channel, Reason, Protocol),
rabbit_log:error("connection ~p, channel ~p - error:~n~p~n",
[ConnPid, Channel, Reason]),
- %% something bad's happened: rollback_and_notify may not be 'ok'
- {_Result, State1} = rollback_and_notify(State),
+ %% something bad's happened: notify_queues may not be 'ok'
+ {_Result, State1} = notify_queues(State),
case CloseChannel of
Channel -> ok = rabbit_writer:send_command(WriterPid, CloseMethod),
{noreply, State1};
@@ -589,7 +583,7 @@ handle_method(_Method, _, State = #ch{state = closing}) ->
{noreply, State};
handle_method(#'channel.close'{}, _, State = #ch{reader_pid = ReaderPid}) ->
- {ok, State1} = rollback_and_notify(State),
+ {ok, State1} = notify_queues(State),
ReaderPid ! {channel_closing, self()},
{noreply, State1};
@@ -601,7 +595,6 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin,
mandatory = Mandatory,
immediate = Immediate},
Content, State = #ch{virtual_host = VHostPath,
- transaction_id = TxnKey,
confirm_enabled = ConfirmEnabled,
trace_state = TraceState}) ->
ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
@@ -623,19 +616,15 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin,
rabbit_trace:tap_trace_in(Message, TraceState),
{RoutingRes, DeliveredQPids} =
rabbit_exchange:publish(
- Exchange,
- rabbit_basic:delivery(Mandatory, Immediate, TxnKey, Message,
- MsgSeqNo)),
+ Exchange, rabbit_basic:delivery(Mandatory, Immediate, Message,
+ MsgSeqNo)),
State2 = process_routing_result(RoutingRes, DeliveredQPids,
ExchangeName, MsgSeqNo, Message,
State1),
maybe_incr_stats([{ExchangeName, 1} |
[{{QPid, ExchangeName}, 1} ||
QPid <- DeliveredQPids]], publish, State2),
- {noreply, case TxnKey of
- none -> State2;
- _ -> add_tx_participants(DeliveredQPids, State2)
- end};
+ {noreply, State2};
{error, Reason} ->
rabbit_misc:protocol_error(precondition_failed,
"invalid message: ~p", [Reason])
@@ -649,22 +638,12 @@ handle_method(#'basic.nack'{delivery_tag = DeliveryTag,
handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
multiple = Multiple},
- _, State = #ch{transaction_id = TxnKey,
- unacked_message_q = UAMQ}) ->
+ _, State = #ch{unacked_message_q = UAMQ}) ->
{Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, Multiple),
- QIncs = ack(TxnKey, Acked),
- Participants = [QPid || {QPid, _} <- QIncs],
+ QIncs = ack(Acked),
maybe_incr_stats(QIncs, ack, State),
- {noreply, case TxnKey of
- none -> ok = notify_limiter(State#ch.limiter_pid, Acked),
- State#ch{unacked_message_q = Remaining};
- _ -> NewUAQ = queue:join(State#ch.uncommitted_ack_q,
- Acked),
- add_tx_participants(
- Participants,
- State#ch{unacked_message_q = Remaining,
- uncommitted_ack_q = NewUAQ})
- end};
+ ok = notify_limiter(State#ch.limiter_pid, Acked),
+ {noreply, State#ch{unacked_message_q = Remaining}};
handle_method(#'basic.get'{queue = QueueNameBin,
no_ack = NoAck},
@@ -1048,35 +1027,6 @@ handle_method(#'queue.purge'{queue = QueueNameBin,
#'queue.purge_ok'{message_count = PurgedMessageCount});
-handle_method(#'tx.select'{}, _, #ch{confirm_enabled = true}) ->
- rabbit_misc:protocol_error(
- precondition_failed, "cannot switch from confirm to tx mode", []);
-
-handle_method(#'tx.select'{}, _, State = #ch{transaction_id = none}) ->
- {reply, #'tx.select_ok'{}, new_tx(State)};
-
-handle_method(#'tx.select'{}, _, State) ->
- {reply, #'tx.select_ok'{}, State};
-
-handle_method(#'tx.commit'{}, _, #ch{transaction_id = none}) ->
- rabbit_misc:protocol_error(
- precondition_failed, "channel is not transactional", []);
-
-handle_method(#'tx.commit'{}, _, State) ->
- {reply, #'tx.commit_ok'{}, internal_commit(State)};
-
-handle_method(#'tx.rollback'{}, _, #ch{transaction_id = none}) ->
- rabbit_misc:protocol_error(
- precondition_failed, "channel is not transactional", []);
-
-handle_method(#'tx.rollback'{}, _, State) ->
- {reply, #'tx.rollback_ok'{}, internal_rollback(State)};
-
-handle_method(#'confirm.select'{}, _, #ch{transaction_id = TxId})
- when TxId =/= none ->
- rabbit_misc:protocol_error(
- precondition_failed, "cannot switch from tx to confirm mode", []);
-
handle_method(#'confirm.select'{nowait = NoWait}, _, State) ->
return_ok(State#ch{confirm_enabled = true},
NoWait, #'confirm.select_ok'{});
@@ -1252,55 +1202,17 @@ collect_acks(ToAcc, PrefixAcc, Q, DeliveryTag, Multiple) ->
precondition_failed, "unknown delivery tag ~w", [DeliveryTag])
end.
-add_tx_participants(MoreP, State = #ch{tx_participants = Participants}) ->
- State#ch{tx_participants = sets:union(Participants,
- sets:from_list(MoreP))}.
-
-ack(TxnKey, UAQ) ->
- fold_per_queue(
- fun (QPid, MsgIds, L) ->
- ok = rabbit_amqqueue:ack(QPid, TxnKey, MsgIds, self()),
- [{QPid, length(MsgIds)} | L]
- end, [], UAQ).
-
-make_tx_id() -> rabbit_guid:guid().
-
-new_tx(State) ->
- State#ch{transaction_id = make_tx_id(),
- tx_participants = sets:new(),
- uncommitted_ack_q = queue:new()}.
-
-internal_commit(State = #ch{transaction_id = TxnKey,
- tx_participants = Participants}) ->
- case rabbit_amqqueue:commit_all(sets:to_list(Participants),
- TxnKey, self()) of
- ok -> ok = notify_limiter(State#ch.limiter_pid,
- State#ch.uncommitted_ack_q),
- new_tx(State);
- {error, Errors} -> rabbit_misc:protocol_error(
- internal_error, "commit failed: ~w", [Errors])
- end.
+ack(UAQ) ->
+ fold_per_queue(fun (QPid, MsgIds, L) ->
+ ok = rabbit_amqqueue:ack(QPid, MsgIds, self()),
+ [{QPid, length(MsgIds)} | L]
+ end, [], UAQ).
-internal_rollback(State = #ch{transaction_id = TxnKey,
- tx_participants = Participants,
- uncommitted_ack_q = UAQ,
- unacked_message_q = UAMQ}) ->
- ?LOGDEBUG("rollback ~p~n - ~p acks uncommitted, ~p messages unacked~n",
- [self(),
- queue:len(UAQ),
- queue:len(UAMQ)]),
- ok = rabbit_amqqueue:rollback_all(sets:to_list(Participants),
- TxnKey, self()),
- NewUAMQ = queue:join(UAQ, UAMQ),
- new_tx(State#ch{unacked_message_q = NewUAMQ}).
-
-rollback_and_notify(State = #ch{state = closing}) ->
+notify_queues(State = #ch{state = closing}) ->
{ok, State};
-rollback_and_notify(State = #ch{transaction_id = none}) ->
- {notify_queues(State), State#ch{state = closing}};
-rollback_and_notify(State) ->
- State1 = internal_rollback(State),
- {notify_queues(State1), State1#ch{state = closing}}.
+notify_queues(State = #ch{consumer_mapping = Consumers}) ->
+ {rabbit_amqqueue:notify_down_all(consumer_queues(Consumers), self()),
+ State#ch{state = closing}}.
fold_per_queue(F, Acc0, UAQ) ->
D = rabbit_misc:queue_fold(
@@ -1319,9 +1231,6 @@ start_limiter(State = #ch{unacked_message_q = UAMQ, start_limiter_fun = SLF}) ->
ok = limit_queues(LPid, State),
LPid.
-notify_queues(#ch{consumer_mapping = Consumers}) ->
- rabbit_amqqueue:notify_down_all(consumer_queues(Consumers), self()).
-
unlimit_queues(State) ->
ok = limit_queues(undefined, State),
undefined.
@@ -1436,17 +1345,13 @@ i(connection, #ch{conn_pid = ConnPid}) -> ConnPid;
i(number, #ch{channel = Channel}) -> Channel;
i(user, #ch{user = User}) -> User#user.username;
i(vhost, #ch{virtual_host = VHost}) -> VHost;
-i(transactional, #ch{transaction_id = TxnKey}) -> TxnKey =/= none;
i(confirm, #ch{confirm_enabled = CE}) -> CE;
i(consumer_count, #ch{consumer_mapping = ConsumerMapping}) ->
dict:size(ConsumerMapping);
i(messages_unconfirmed, #ch{unconfirmed_mq = UMQ}) ->
gb_trees:size(UMQ);
-i(messages_unacknowledged, #ch{unacked_message_q = UAMQ,
- uncommitted_ack_q = UAQ}) ->
- queue:len(UAMQ) + queue:len(UAQ);
-i(acks_uncommitted, #ch{uncommitted_ack_q = UAQ}) ->
- queue:len(UAQ);
+i(messages_unacknowledged, #ch{unacked_message_q = UAMQ}) ->
+ queue:len(UAMQ);
i(prefetch_count, #ch{limiter_pid = LimiterPid}) ->
rabbit_limiter:get_limit(LimiterPid);
i(client_flow_blocked, #ch{limiter_pid = LimiterPid}) ->
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 9eef384a..6eb1aaba 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -301,7 +301,7 @@ action(list_connections, Node, Args, _Opts, Inform) ->
action(list_channels, Node, Args, _Opts, Inform) ->
Inform("Listing channels", []),
- ArgAtoms = default_if_empty(Args, [pid, user, transactional, consumer_count,
+ ArgAtoms = default_if_empty(Args, [pid, user, consumer_count,
messages_unacknowledged]),
display_info_list(rpc_call(Node, rabbit_channel, info_all, [ArgAtoms]),
ArgAtoms);
diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl
index 3fb0817a..93aad9e3 100644
--- a/src/rabbit_error_logger.erl
+++ b/src/rabbit_error_logger.erl
@@ -71,7 +71,7 @@ publish1(RoutingKey, Format, Data, LogExch) ->
%% second resolution, not millisecond.
Timestamp = rabbit_misc:now_ms() div 1000,
{ok, _RoutingRes, _DeliveredQPids} =
- rabbit_basic:publish(LogExch, RoutingKey, false, false, none,
+ rabbit_basic:publish(LogExch, RoutingKey, false, false,
#'P_basic'{content_type = <<"text/plain">>,
timestamp = Timestamp},
list_to_binary(io_lib:format(Format, Data))),
diff --git a/src/rabbit_mirror_queue_coordinator.erl b/src/rabbit_mirror_queue_coordinator.erl
index 2727c1d0..d1c3a2e5 100644
--- a/src/rabbit_mirror_queue_coordinator.erl
+++ b/src/rabbit_mirror_queue_coordinator.erl
@@ -254,45 +254,6 @@
%% sender_death message. The slave will then be able to tidy up its
%% state as normal.
%%
-%% We don't support transactions on mirror queues. To do so is
-%% challenging. The underlying bq is free to add the contents of the
-%% txn to the queue proper at any point after the tx.commit comes in
-%% but before the tx.commit-ok goes out. This means that it is not
-%% safe for all mirrors to simply issue the bq:tx_commit at the same
-%% time, as the addition of the txn's contents to the queue may
-%% subsequently be inconsistently interwoven with other actions on the
-%% bq. The solution to this is, in the master, wrap the PostCommitFun
-%% and do the gm:broadcast in there: at that point, you're in the bq
-%% (well, there's actually nothing to stop that function being invoked
-%% by some other process, but let's pretend for now: you could always
-%% use run_backing_queue to ensure you really are in the queue process
-%% (the _async variant would be unsafe from an ordering pov)), the
-%% gm:broadcast is safe because you don't have to worry about races
-%% with other gm:broadcast calls (same process). Thus this signal
-%% would indicate sufficiently to all the slaves that they must insert
-%% the complete contents of the txn at precisely this point in the
-%% stream of events.
-%%
-%% However, it's quite difficult for the slaves to make that happen:
-%% they would be forced to issue the bq:tx_commit at that point, but
-%% then stall processing any further instructions from gm until they
-%% receive the notification from their bq that the tx_commit has fully
-%% completed (i.e. they need to treat what is an async system as being
-%% fully synchronous). This is not too bad (apart from the
-%% vomit-inducing notion of it all): just need a queue of instructions
-%% from the GM; but then it gets rather worse when you consider what
-%% needs to happen if the master dies at this point and the slave in
-%% the middle of this tx_commit needs to be promoted.
-%%
-%% Finally, we can't possibly hope to make transactions atomic across
-%% mirror queues, and it's not even clear that that's desirable: if a
-%% slave fails whilst there's an open transaction in progress then
-%% when the channel comes to commit the txn, it will detect the
-%% failure and destroy the channel. However, the txn will have
-%% actually committed successfully in all the other mirrors (including
-%% master). To do this bit properly would require 2PC and all the
-%% baggage that goes with that.
-%%
%% Recovery of mirrored queues is straightforward: as nodes die, the
%% remaining nodes record this, and eventually a situation is reached
%% in which only one node is alive, which is the master. This is the
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 55d61d41..66ff575f 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -488,7 +488,7 @@ promote_me(From, #state { q = Q,
%%
%% Everything that's in MA gets requeued. Consequently the new
%% master should start with a fresh AM as there are no messages
- %% pending acks (txns will have been rolled back).
+ %% pending acks.
MSList = dict:to_list(MS),
SS = dict:from_list(
@@ -612,8 +612,7 @@ confirm_sender_death(Pid) ->
maybe_enqueue_message(
Delivery = #delivery { message = #basic_message { id = MsgId },
msg_seq_no = MsgSeqNo,
- sender = ChPid,
- txn = none },
+ sender = ChPid },
EnqueueOnPromotion,
State = #state { sender_queues = SQ, msg_id_status = MS }) ->
State1 = ensure_monitoring(ChPid, State),
@@ -655,10 +654,7 @@ maybe_enqueue_message(
SQ1 = remove_from_pending_ch(MsgId, ChPid, SQ),
State1 #state { msg_id_status = dict:erase(MsgId, MS),
sender_queues = SQ1 }
- end;
-maybe_enqueue_message(_Delivery, _EnqueueOnPromotion, State) ->
- %% We don't support txns in mirror queues.
- State.
+ end.
get_sender_queue(ChPid, SQ) ->
case dict:find(ChPid, SQ) of
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index aaf3df78..bf89cdb2 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -76,11 +76,10 @@
%% the segment file combined with the journal, no writing needs to be
%% done to the segment file either (in fact it is deleted if it exists
%% at all). This is safe given that the set of acks is a subset of the
-%% set of publishes. When it's necessary to sync messages because of
-%% transactions, it's only necessary to fsync on the journal: when
-%% entries are distributed from the journal to segment files, those
-%% segments appended to are fsync'd prior to the journal being
-%% truncated.
+%% set of publishes. When it is necessary to sync messages, it is
+%% sufficient to fsync on the journal: when entries are distributed
+%% from the journal to segment files, those segments appended to are
+%% fsync'd prior to the journal being truncated.
%%
%% This module is also responsible for scanning the queue index files
%% and seeding the message store on start up.
@@ -289,14 +288,13 @@ sync(State = #qistate { unsynced_msg_ids = MsgIds }) ->
sync_if([] =/= MsgIds, State).
sync(SeqIds, State) ->
- %% The SeqIds here contains the SeqId of every publish and ack in
- %% the transaction. Ideally we should go through these seqids and
- %% only sync the journal if the pubs or acks appear in the
+ %% The SeqIds here contains the SeqId of every publish and ack to
+ %% be sync'ed. Ideally we should go through these seqids and only
+ %% sync the journal if the pubs or acks appear in the
%% journal. However, this would be complex to do, and given that
%% the variable queue publishes and acks to the qi, and then
%% syncs, all in one operation, there is no possibility of the
- %% seqids not being in the journal, provided the transaction isn't
- %% emptied (handled by sync_if anyway).
+ %% seqids not being in the journal.
sync_if([] =/= SeqIds, State).
flush(State = #qistate { dirty_count = 0 }) -> State;
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 6e44c7a0..bc1b00b2 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -705,7 +705,6 @@ test_topic_expect_match(X, List) ->
Res = rabbit_exchange_type_topic:route(
X, #delivery{mandatory = false,
immediate = false,
- txn = none,
sender = self(),
message = Message}),
ExpectedRes = lists:map(
@@ -1669,8 +1668,8 @@ test_backing_queue() ->
passed = test_queue_index(),
passed = test_queue_index_props(),
passed = test_variable_queue(),
- %% FIXME: replace the use of tx in these with confirms
- %% passed = test_variable_queue_delete_msg_store_files_callback(),
+ passed = test_variable_queue_delete_msg_store_files_callback(),
+ %% FIXME: re-enable once fixed
%% passed = test_queue_recover(),
application:set_env(rabbit, queue_index_max_journal_entries,
MaxJournal, infinity),
@@ -2319,17 +2318,16 @@ test_variable_queue_all_the_bits_not_covered_elsewhere2(VQ0) ->
test_queue_recover() ->
Count = 2 * rabbit_queue_index:next_segment_boundary(0),
- TxID = rabbit_guid:guid(),
{new, #amqqueue { pid = QPid, name = QName } = Q} =
rabbit_amqqueue:declare(test_queue(), true, false, [], none),
[begin
Msg = rabbit_basic:message(rabbit_misc:r(<<>>, exchange, <<>>),
<<>>, #'P_basic'{delivery_mode = 2}, <<>>),
- Delivery = #delivery{mandatory = false, immediate = false, txn = TxID,
+ Delivery = #delivery{mandatory = false, immediate = false,
sender = self(), message = Msg},
true = rabbit_amqqueue:deliver(QPid, Delivery)
end || _ <- lists:seq(1, Count)],
- rabbit_amqqueue:commit_all([QPid], TxID, self()),
+ %% FIXME: wait for confirms of all publishes
exit(QPid, kill),
MRef = erlang:monitor(process, QPid),
receive {'DOWN', MRef, process, QPid, _Info} -> ok
@@ -2356,18 +2354,17 @@ test_variable_queue_delete_msg_store_files_callback() ->
ok = restart_msg_store_empty(),
{new, #amqqueue { pid = QPid, name = QName } = Q} =
rabbit_amqqueue:declare(test_queue(), true, false, [], none),
- TxID = rabbit_guid:guid(),
Payload = <<0:8388608>>, %% 1MB
Count = 30,
[begin
Msg = rabbit_basic:message(
rabbit_misc:r(<<>>, exchange, <<>>),
<<>>, #'P_basic'{delivery_mode = 2}, Payload),
- Delivery = #delivery{mandatory = false, immediate = false, txn = TxID,
+ Delivery = #delivery{mandatory = false, immediate = false,
sender = self(), message = Msg},
true = rabbit_amqqueue:deliver(QPid, Delivery)
end || _ <- lists:seq(1, Count)],
- rabbit_amqqueue:commit_all([QPid], TxID, self()),
+ %% FIXME: wait for confirms of all publishes
rabbit_amqqueue:set_ram_duration_target(QPid, 0),
CountMinusOne = Count - 1,
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index 03b2c9e8..2db960ac 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -20,7 +20,7 @@
-ifdef(use_specs).
--export_type([txn/0, maybe/1, info/0, infos/0, info_key/0, info_keys/0,
+-export_type([maybe/1, info/0, infos/0, info_key/0, info_keys/0,
message/0, msg_id/0, basic_message/0,
delivery/0, content/0, decoded_content/0, undecoded_content/0,
unencoded_content/0, encoded_content/0, message_properties/0,
@@ -73,16 +73,12 @@
-type(delivery() ::
#delivery{mandatory :: boolean(),
immediate :: boolean(),
- txn :: maybe(txn()),
sender :: pid(),
message :: message()}).
-type(message_properties() ::
#message_properties{expiry :: pos_integer() | 'undefined',
needs_confirming :: boolean()}).
-%% this is really an abstract type, but dialyzer does not support them
--type(txn() :: rabbit_guid:guid()).
-
-type(info_key() :: atom()).
-type(info_keys() :: [info_key()]).