You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2010/08/03 17:33:13 UTC
svn commit: r981933 - in /qpid/trunk/qpid/cpp/src: qpid/SessionState.cpp
qpid/SessionState.h qpid/amqp_0_10/SessionHandler.cpp
qpid/broker/SessionState.cpp qpid/broker/SessionState.h
tests/BrokerFixture.h tests/ClusterFailover.cpp tests/ClusterFixture.h
Author: aconway
Date: Tue Aug 3 15:33:12 2010
New Revision: 981933
URL: http://svn.apache.org/viewvc?rev=981933&view=rev
Log:
Disable non-0 session timeouts.
Since session resume is not fully implemented, non-0 session timeouts
are of no use. Moreover the partial implementation causes problems in
a cluster as stale sessions kept alive by a timeout can interfere with
failover and updates.
Modified:
qpid/trunk/qpid/cpp/src/qpid/SessionState.cpp
qpid/trunk/qpid/cpp/src/qpid/SessionState.h
qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
qpid/trunk/qpid/cpp/src/tests/ClusterFailover.cpp
qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h
Modified: qpid/trunk/qpid/cpp/src/qpid/SessionState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/SessionState.cpp?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/SessionState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/SessionState.cpp Tue Aug 3 15:33:12 2010
@@ -95,6 +95,9 @@ SessionState::SendState::SendState() : u
SessionState::ReceiveState::ReceiveState() : bytesSinceKnownCompleted() {}
+uint32_t SessionState::getTimeout() const { return timeout; }
+void SessionState::setTimeout(uint32_t seconds) { timeout = seconds; }
+
SessionPoint SessionState::senderGetCommandPoint() { return sender.sendPoint; }
SequenceSet SessionState::senderGetIncomplete() const { return sender.incomplete; }
SessionPoint SessionState::senderGetReplayPoint() const { return sender.replayPoint; }
@@ -240,7 +243,7 @@ SessionState::Configuration::Configurati
replayFlushLimit(flush), replayHardLimit(hard) {}
SessionState::SessionState(const SessionId& i, const Configuration& c)
- : id(i), timeout(), config(c), stateful(), receiverTrackingDisabled(false)
+ : id(i), timeout(0), config(c), stateful(false), receiverTrackingDisabled(false)
{
QPID_LOG(debug, "SessionState::SessionState " << id << ": " << this);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/SessionState.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/SessionState.h?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/SessionState.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/SessionState.h Tue Aug 3 15:33:12 2010
@@ -92,8 +92,8 @@ class SessionState {
const SessionId& getId() const { return id; }
- uint32_t getTimeout() const { return timeout; }
- void setTimeout(uint32_t seconds) { timeout = seconds; }
+ QPID_COMMON_EXTERN virtual uint32_t getTimeout() const;
+ QPID_COMMON_EXTERN virtual void setTimeout(uint32_t seconds);
bool operator==(const SessionId& other) const { return id == other; }
bool operator==(const SessionState& other) const { return id == other.id; }
Modified: qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp Tue Aug 3 15:33:12 2010
@@ -205,7 +205,7 @@ void SessionHandler::handleDetach() {
void SessionHandler::requestTimeout(uint32_t t) {
checkAttached();
getState()->setTimeout(t);
- peer.timeout(t);
+ peer.timeout(getState()->getTimeout());
}
void SessionHandler::timeout(uint32_t t) {
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp Tue Aug 3 15:33:12 2010
@@ -380,6 +380,11 @@ void SessionState::readyToSend() {
Broker& SessionState::getBroker() { return broker; }
+// Session resume is not fully implemented so it is useless to set a
+// non-0 timeout. Moreover it creates problems in a cluster because
+// dead sessions are kept and interfere with failover.
+void SessionState::setTimeout(uint32_t) { }
+
framing::AMQP_ClientProxy& SessionState::getClusterOrderProxy() {
return handler->getClusterOrderProxy();
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.h Tue Aug 3 15:33:12 2010
@@ -92,6 +92,8 @@ class SessionState : public qpid::Sessio
Broker& getBroker();
+ void setTimeout(uint32_t seconds);
+
/** OutputControl **/
void abort();
void activateOutput();
Modified: qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h (original)
+++ qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h Tue Aug 3 15:33:12 2010
@@ -122,10 +122,10 @@ struct ClientT {
qpid::client::LocalQueue lq;
std::string name;
- ClientT(uint16_t port, const std::string& name_=std::string())
- : connection(port), session(connection.newSession(name_)), subs(session), name(name_) {}
- ClientT(const qpid::client::ConnectionSettings& settings, const std::string& name_=std::string())
- : connection(settings), session(connection.newSession(name_)), subs(session), name(name_) {}
+ ClientT(uint16_t port, const std::string& name_=std::string(), int timeout=0)
+ : connection(port), session(connection.newSession(name_,timeout)), subs(session), name(name_) {}
+ ClientT(const qpid::client::ConnectionSettings& settings, const std::string& name_=std::string(), int timeout=0)
+ : connection(settings), session(connection.newSession(name_, timeout)), subs(session), name(name_) {}
~ClientT() { close(); }
void close() { session.close(); connection.close(); }
Modified: qpid/trunk/qpid/cpp/src/tests/ClusterFailover.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClusterFailover.cpp?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClusterFailover.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClusterFailover.cpp Tue Aug 3 15:33:12 2010
@@ -50,19 +50,60 @@ using boost::shared_ptr;
// Timeout for tests that wait for messages
const sys::Duration TIMEOUT=sys::TIME_SEC/4;
+ClusterFixture::Args getArgs() {
+ ClusterFixture::Args args;
+ args += "--auth", "no", "--no-module-dir", "--load-module", getLibPath("CLUSTER_LIB");
+ return args;
+}
// Test re-connecting with same session name after a failure.
QPID_AUTO_TEST_CASE(testReconnectSameSessionName) {
- ostringstream clusterLib;
- clusterLib << getLibPath("CLUSTER_LIB");
- ClusterFixture::Args args = list_of<string>("--auth")("no")("--no-module-dir")("--no-data-dir")("--load-module")(clusterLib.str());
- ClusterFixture cluster(2, args, -1);
- Client c0(cluster[0], "foo");
+ ClusterFixture cluster(2, getArgs(), -1);
+ // Specify a timeout to make sure it is ignored, session resume is
+ // not implemented so sessions belonging to dead brokers should
+ // not be kept.
+ Client c0(cluster[0], "foo", 5);
BOOST_CHECK_EQUAL(2u, knownBrokerPorts(c0.connection, 2).size()); // wait for both.
+ c0.session.queueDeclare("q");
+ c0.session.messageTransfer(arg::content=Message("sendme", "q"));
+ BOOST_CHECK_EQUAL(c0.subs.get("q").getData(), "sendme");
cluster.killWithSilencer(0, c0.connection, 9);
- Client c1(cluster[1], "foo"); // Using same name, should be cleaned up.
+ Client c1(cluster[1], "foo", 5);
+ c1.session.queueQuery(); // Try to use the session.
}
+QPID_AUTO_TEST_CASE(testReconnectExclusiveQueue) {
+ // Regresion test. Session timeouts should be ignored
+ // by the broker as session resume is not implemented.
+ ClusterFixture cluster(2, getArgs(), -1);
+ Client c0(cluster[0], "foo", 5);
+ c0.session.queueDeclare("exq", arg::exclusive=true);
+ SubscriptionSettings settings;
+ settings.exclusive = true;
+ settings.autoAck = 0;
+ Subscription s0 = c0.subs.subscribe(c0.lq, "exq", settings, "exsub");
+ c0.session.messageTransfer(arg::content=Message("sendme", "exq"));
+ BOOST_CHECK_EQUAL(c0.lq.get().getData(), "sendme");
+
+ // Regression: core dump on exit if unacked messages were left in
+ // a session with a timeout.
+ cluster.kill(0);
+
+ // Regression: session timeouts prevented re-connecting to
+ // exclusive queue.
+ Client c1(cluster[1]);
+ c1.session.queueDeclare("exq", arg::exclusive=true);
+ Subscription s1 = c1.subs.subscribe(c1.lq, "exq", settings, "exsub");
+ s1.cancel();
+
+ // Regression: session timeouts prevented new member joining
+ // cluster with exclusive queues.
+ cluster.add();
+ Client c2(cluster[2]);
+ c2.session.queueQuery();
+}
+
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
Modified: qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h?rev=981933&r1=981932&r2=981933&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClusterFixture.h Tue Aug 3 15:33:12 2010
@@ -89,7 +89,7 @@ class ClusterFixture : public vector<uin
/** Kill a forked broker with sig, or shutdown localBroker. */
void kill(size_t n, int sig=SIGINT);
- /** Kill a broker and suppressing errors from closing connection c. */
+ /** Kill a broker and suppress errors from closing connection c. */
void killWithSilencer(size_t n, client::Connection& c, int sig=SIGINT);
private:
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org