You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2009/11/23 13:50:58 UTC
svn commit: r883325 - in /qpid/trunk/qpid/cpp: include/qpid/messaging/
src/qpid/client/amqp0_10/ src/qpid/messaging/ src/tests/
Author: gsim
Date: Mon Nov 23 12:50:58 2009
New Revision: 883325
URL: http://svn.apache.org/viewvc?rev=883325&view=rev
Log:
QPID-664: Allow session to be specified as transactional; add basic test of transaction implementation.
Modified:
qpid/trunk/qpid/cpp/include/qpid/messaging/Connection.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
qpid/trunk/qpid/cpp/src/qpid/messaging/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionImpl.h
qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
Modified: qpid/trunk/qpid/cpp/include/qpid/messaging/Connection.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/include/qpid/messaging/Connection.h?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/include/qpid/messaging/Connection.h (original)
+++ qpid/trunk/qpid/cpp/include/qpid/messaging/Connection.h Mon Nov 23 12:50:58 2009
@@ -48,7 +48,10 @@
QPID_CLIENT_EXTERN ~Connection();
QPID_CLIENT_EXTERN Connection& operator=(const Connection&);
QPID_CLIENT_EXTERN void close();
+ QPID_CLIENT_EXTERN Session newSession(bool transactional, const std::string& name = std::string());
QPID_CLIENT_EXTERN Session newSession(const std::string& name = std::string());
+ QPID_CLIENT_EXTERN Session newSession(const char* name);
+
QPID_CLIENT_EXTERN Session getSession(const std::string& name) const;
private:
friend class qpid::client::PrivateImplRef<Connection>;
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp Mon Nov 23 12:50:58 2009
@@ -118,10 +118,10 @@
}
}
-qpid::messaging::Session ConnectionImpl::newSession(const std::string& n)
+qpid::messaging::Session ConnectionImpl::newSession(bool transactional, const std::string& n)
{
std::string name = n.empty() ? Uuid(true).str() : n;
- qpid::messaging::Session impl(new SessionImpl(*this));
+ qpid::messaging::Session impl(new SessionImpl(*this, transactional));
{
qpid::sys::Mutex::ScopedLock l(lock);
sessions[name] = impl;
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h Mon Nov 23 12:50:58 2009
@@ -42,7 +42,7 @@
public:
ConnectionImpl(const std::string& url, const qpid::messaging::Variant::Map& options);
void close();
- qpid::messaging::Session newSession(const std::string& name);
+ qpid::messaging::Session newSession(bool transactional, const std::string& name);
qpid::messaging::Session getSession(const std::string& name) const;
void closed(SessionImpl&);
void reconnect();
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp Mon Nov 23 12:50:58 2009
@@ -50,7 +50,7 @@
namespace client {
namespace amqp0_10 {
-SessionImpl::SessionImpl(ConnectionImpl& c) : connection(c) {}
+SessionImpl::SessionImpl(ConnectionImpl& c, bool t) : connection(c), transactional(t) {}
void SessionImpl::sync()
@@ -134,6 +134,7 @@
qpid::sys::Mutex::ScopedLock l(lock);
session = s;
incoming.setSession(session);
+ if (transactional) session.txSelect();
for (Receivers::iterator i = receivers.begin(); i != receivers.end(); ++i) {
getImplPtr<Receiver, ReceiverImpl>(i->second)->init(session, resolver);
}
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.h Mon Nov 23 12:50:58 2009
@@ -54,7 +54,7 @@
class SessionImpl : public qpid::messaging::SessionImpl
{
public:
- SessionImpl(ConnectionImpl&);
+ SessionImpl(ConnectionImpl&, bool transactional);
void commit();
void rollback();
void acknowledge();
@@ -111,6 +111,7 @@
IncomingMessages incoming;
Receivers receivers;
Senders senders;
+ const bool transactional;
bool accept(ReceiverImpl*, qpid::messaging::Message*, IncomingMessages::MessageTransfer&);
bool getIncoming(IncomingMessages::Handler& handler, qpid::sys::Duration timeout);
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/Connection.cpp?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/Connection.cpp Mon Nov 23 12:50:58 2009
@@ -50,7 +50,12 @@
Connection::~Connection() { PI::dtor(*this); }
void Connection::close() { impl->close(); }
-Session Connection::newSession(const std::string& name) { return impl->newSession(name); }
+Session Connection::newSession(const char* name) { return impl->newSession(false, name); }
+Session Connection::newSession(const std::string& name) { return impl->newSession(false, name); }
+Session Connection::newSession(bool transactional, const std::string& name)
+{
+ return impl->newSession(transactional, name);
+}
Session Connection::getSession(const std::string& name) const { return impl->getSession(name); }
InvalidOptionString::InvalidOptionString(const std::string& msg) : Exception(msg) {}
Modified: qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionImpl.h?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/ConnectionImpl.h Mon Nov 23 12:50:58 2009
@@ -37,7 +37,7 @@
public:
virtual ~ConnectionImpl() {}
virtual void close() = 0;
- virtual Session newSession(const std::string& name) = 0;
+ virtual Session newSession(bool transactional, const std::string& name) = 0;
virtual Session getSession(const std::string& name) const = 0;
private:
};
Modified: qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp?rev=883325&r1=883324&r2=883325&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp Mon Nov 23 12:50:58 2009
@@ -745,6 +745,33 @@
BOOST_CHECK_EQUAL(out.getContent(), in.getContent());
}
+QPID_AUTO_TEST_CASE(testTx)
+{
+ QueueFixture fix;
+ Session ssn1 = fix.connection.newSession(true);
+ Session ssn2 = fix.connection.newSession(true);
+ Sender sender1 = ssn1.createSender(fix.queue);
+ Sender sender2 = ssn2.createSender(fix.queue);
+ Receiver receiver1 = ssn1.createReceiver(fix.queue);
+ Receiver receiver2 = ssn2.createReceiver(fix.queue);
+ Message in;
+
+ send(sender1, 5, 1, "A");
+ send(sender2, 5, 1, "B");
+ ssn2.commit();
+ receive(receiver1, 5, 1, "B");//(only those from sender2 should be received)
+ BOOST_CHECK(!receiver1.fetch(in, 0));//check there are no more messages
+ ssn1.rollback();
+ receive(receiver2, 5, 1, "B");
+ BOOST_CHECK(!receiver2.fetch(in, 0));//check there are no more messages
+ ssn2.rollback();
+ receive(receiver1, 5, 1, "B");
+ BOOST_CHECK(!receiver1.fetch(in, 0));//check there are no more messages
+ ssn1.commit();
+ //check neither receiver gets any more messages:
+ BOOST_CHECK(!receiver1.fetch(in, 0));
+ BOOST_CHECK(!receiver2.fetch(in, 0));
+}
QPID_AUTO_TEST_SUITE_END()
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org