summaryrefslogtreecommitdiff
path: root/cpp/src/qpid/client
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid/client')
-rw-r--r--cpp/src/qpid/client/Connection.cpp17
-rw-r--r--cpp/src/qpid/client/Connection.h9
-rw-r--r--cpp/src/qpid/client/ConnectionAccess.h41
-rw-r--r--cpp/src/qpid/client/ConnectionImpl.cpp29
-rw-r--r--cpp/src/qpid/client/ConnectionImpl.h13
-rw-r--r--cpp/src/qpid/client/FailoverListener.cpp41
-rw-r--r--cpp/src/qpid/client/FailoverListener.h17
-rw-r--r--cpp/src/qpid/client/SessionImpl.cpp59
-rw-r--r--cpp/src/qpid/client/SessionImpl.h18
-rw-r--r--cpp/src/qpid/client/SubscriptionManager.h6
10 files changed, 177 insertions, 73 deletions
diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index b8b4f9ccf3..27706fab8c 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -44,7 +44,7 @@ using namespace qpid::sys;
namespace qpid {
namespace client {
-Connection::Connection() : channelIdCounter(0), version(framing::highestProtocolVersion) {}
+Connection::Connection() : version(framing::highestProtocolVersion) {}
Connection::~Connection(){ }
@@ -106,26 +106,19 @@ void Connection::open(const ConnectionSettings& settings)
impl = shared_ptr<ConnectionImpl>(new ConnectionImpl(version, settings));
impl->open();
- max_frame_size = impl->getNegotiatedSettings().maxFrameSize;
}
-Session Connection::newSession(const std::string& name) {
+Session Connection::newSession(const std::string& name, uint32_t timeout) {
if (!isOpen())
throw Exception(QPID_MSG("Connection has not yet been opened"));
- shared_ptr<SessionImpl> simpl(
- new SessionImpl(name, impl, ++channelIdCounter, max_frame_size));
- impl->addSession(simpl);
- simpl->open(0);
Session s;
- SessionBase_0_10Access(s).set(simpl);
+ SessionBase_0_10Access(s).set(impl->newSession(name, timeout));
return s;
}
void Connection::resume(Session& session) {
if (!isOpen())
throw Exception(QPID_MSG("Connection is not open."));
-
- session.impl->setChannel(++channelIdCounter);
impl->addSession(session.impl);
session.impl->resume(impl);
}
@@ -134,4 +127,8 @@ void Connection::close() {
impl->close();
}
+std::vector<Url> Connection::getKnownBrokers() {
+ return isOpen() ? impl->getKnownBrokers() : std::vector<Url>();
+}
+
}} // namespace qpid::client
diff --git a/cpp/src/qpid/client/Connection.h b/cpp/src/qpid/client/Connection.h
index a1575dd524..a5ea40ff38 100644
--- a/cpp/src/qpid/client/Connection.h
+++ b/cpp/src/qpid/client/Connection.h
@@ -32,6 +32,7 @@ class Url;
namespace client {
class ConnectionSettings;
+class ConnectionImpl;
/**
* Represents a connection to an AMQP broker. All communication is
@@ -42,9 +43,7 @@ class ConnectionSettings;
*/
class Connection
{
- framing::ChannelId channelIdCounter;
framing::ProtocolVersion version;
- uint16_t max_frame_size;
protected:
boost::shared_ptr<ConnectionImpl> impl;
@@ -55,6 +54,7 @@ class Connection
* @see open()
*/
Connection();
+
~Connection();
/**
@@ -157,7 +157,7 @@ class Connection
* If the name is empty (the default) then a unique name will be
* chosen using a Universally-unique identifier (UUID) algorithm.
*/
- Session newSession(const std::string& name=std::string());
+ Session newSession(const std::string& name=std::string(), uint32_t timeoutSeconds = 0);
/**
* Resume a suspended session. A session may be resumed
@@ -167,7 +167,8 @@ class Connection
bool isOpen() const;
-
+ std::vector<Url> getKnownBrokers();
+
friend class ConnectionAccess; ///<@internal
friend class SessionBase_0_10; ///<@internal
};
diff --git a/cpp/src/qpid/client/ConnectionAccess.h b/cpp/src/qpid/client/ConnectionAccess.h
new file mode 100644
index 0000000000..b662fd5d8b
--- /dev/null
+++ b/cpp/src/qpid/client/ConnectionAccess.h
@@ -0,0 +1,41 @@
+#ifndef QPID_CLIENT_CONNECTIONACCESS_H
+#define QPID_CLIENT_CONNECTIONACCESS_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/client/Connection.h"
+
+/**@file @internal Internal use only */
+
+namespace qpid {
+namespace client {
+
+
+
+struct ConnectionAccess {
+ static void setVersion(Connection& c, const framing::ProtocolVersion& v) { c.version = v; }
+ static boost::shared_ptr<ConnectionImpl> getImpl(Connection& c) { return c.impl; }
+};
+
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_CONNECTIONACCESS_H*/
diff --git a/cpp/src/qpid/client/ConnectionImpl.cpp b/cpp/src/qpid/client/ConnectionImpl.cpp
index d9ac65c1b3..910c908ee2 100644
--- a/cpp/src/qpid/client/ConnectionImpl.cpp
+++ b/cpp/src/qpid/client/ConnectionImpl.cpp
@@ -22,8 +22,10 @@
#include "Connector.h"
#include "ConnectionSettings.h"
#include "SessionImpl.h"
+#include "FailoverListener.h"
#include "qpid/log/Statement.h"
+#include "qpid/Url.h"
#include "qpid/framing/enum.h"
#include "qpid/framing/reply_exceptions.h"
@@ -40,7 +42,9 @@ using namespace qpid::framing::connection;//for connection error codes
ConnectionImpl::ConnectionImpl(framing::ProtocolVersion v, const ConnectionSettings& settings)
: Bounds(settings.maxFrameSize * settings.bounds),
handler(settings, v),
- version(v)
+ failover(new FailoverListener()),
+ version(v),
+ nextChannel(1)
{
QPID_LOG(debug, "ConnectionImpl created for " << version);
handler.in = boost::bind(&ConnectionImpl::incoming, this, _1);
@@ -56,12 +60,14 @@ ConnectionImpl::~ConnectionImpl() {
// Important to close the connector first, to ensure the
// connector thread does not call on us while the destructor
// is running.
- if (connector) connector->close();
+ failover.reset();
+ if (connector) connector->close();
}
-void ConnectionImpl::addSession(const boost::shared_ptr<SessionImpl>& session)
+void ConnectionImpl::addSession(const boost::shared_ptr<SessionImpl>& session, uint16_t channel)
{
Mutex::ScopedLock l(lock);
+ session->setChannel(channel ? channel : nextChannel++);
boost::weak_ptr<SessionImpl>& s = sessions[session->getChannel()];
if (s.lock()) throw SessionBusyException();
s = session;
@@ -102,7 +108,9 @@ void ConnectionImpl::open()
connector->setShutdownHandler(this);
connector->connect(host, port);
connector->init();
- handler.waitForOpen();
+ handler.waitForOpen();
+
+ if (failover.get()) failover->start(shared_from_this());
}
void ConnectionImpl::idleIn()
@@ -162,3 +170,16 @@ const ConnectionSettings& ConnectionImpl::getNegotiatedSettings()
return handler;
}
+std::vector<qpid::Url> ConnectionImpl::getKnownBrokers() {
+ // FIXME aconway 2008-10-08: initialize failover list from openOk or settings
+ return failover ? failover->getKnownBrokers() : std::vector<qpid::Url>();
+}
+
+boost::shared_ptr<SessionImpl> ConnectionImpl::newSession(const std::string& name, uint32_t timeout, uint16_t channel) {
+ boost::shared_ptr<SessionImpl> simpl(new SessionImpl(name, shared_from_this()));
+ addSession(simpl, channel);
+ simpl->open(timeout);
+ return simpl;
+}
+
+void ConnectionImpl::stopFailoverListener() { failover.reset(); }
diff --git a/cpp/src/qpid/client/ConnectionImpl.h b/cpp/src/qpid/client/ConnectionImpl.h
index aca26b963d..22450a7ddf 100644
--- a/cpp/src/qpid/client/ConnectionImpl.h
+++ b/cpp/src/qpid/client/ConnectionImpl.h
@@ -24,6 +24,7 @@
#include "Bounds.h"
#include "ConnectionHandler.h"
+
#include "qpid/framing/FrameHandler.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/ShutdownHandler.h"
@@ -41,6 +42,7 @@ namespace client {
class Connector;
class ConnectionSettings;
class SessionImpl;
+class FailoverListener;
class ConnectionImpl : public Bounds,
public framing::FrameHandler,
@@ -54,7 +56,9 @@ class ConnectionImpl : public Bounds,
SessionMap sessions;
ConnectionHandler handler;
boost::scoped_ptr<Connector> connector;
+ boost::scoped_ptr<FailoverListener> failover;
framing::ProtocolVersion version;
+ uint16_t nextChannel;
sys::Mutex lock;
template <class F> void closeInternal(const F&);
@@ -72,13 +76,18 @@ class ConnectionImpl : public Bounds,
void open();
bool isOpen() const;
- void addSession(const boost::shared_ptr<SessionImpl>&);
+ boost::shared_ptr<SessionImpl> newSession(const std::string& name, uint32_t timeout, uint16_t channel=0);
+ void addSession(const boost::shared_ptr<SessionImpl>&, uint16_t channel=0);
void close();
void handle(framing::AMQFrame& frame);
void erase(uint16_t channel);
-
+ void stopFailoverListener();
+
const ConnectionSettings& getNegotiatedSettings();
+
+ std::vector<Url> getKnownBrokers();
+
};
}}
diff --git a/cpp/src/qpid/client/FailoverListener.cpp b/cpp/src/qpid/client/FailoverListener.cpp
index 95c137d922..3254686d9c 100644
--- a/cpp/src/qpid/client/FailoverListener.cpp
+++ b/cpp/src/qpid/client/FailoverListener.cpp
@@ -19,27 +19,50 @@
*
*/
#include "FailoverListener.h"
+#include "SessionBase_0_10Access.h"
+#include "qpid/client/SubscriptionManager.h"
namespace qpid {
namespace client {
static const std::string AMQ_FAILOVER("amq.failover");
-FailoverListener::FailoverListener(Connection c)
- : connection(c), session(c.newSession()), subscriptions(session)
-{
+static Session makeSession(boost::shared_ptr<SessionImpl> si) {
+ // Hold only a weak pointer to the ConnectionImpl so a
+ // FailoverListener in a ConnectionImpl won't createa a shared_ptr
+ // cycle.
+ //
+ si->setWeakPtr(true);
+ Session s;
+ SessionBase_0_10Access(s).set(si);
+ return s;
+}
+
+FailoverListener::FailoverListener() {}
+
+void FailoverListener::start(const boost::shared_ptr<ConnectionImpl>& c) {
+ Session session = makeSession(c->newSession(std::string(), 0));
+ if (session.exchangeQuery(arg::name=AMQ_FAILOVER).getNotFound()) {
+ session.close();
+ return;
+ }
+ subscriptions.reset(new SubscriptionManager(session));
std::string qname=AMQ_FAILOVER + "." + session.getId().getName();
- if (session.exchangeQuery(arg::exchange=AMQ_FAILOVER).getType().empty())
- return; // Failover exchange not implemented.
session.queueDeclare(arg::queue=qname, arg::exclusive=true, arg::autoDelete=true);
session.exchangeBind(arg::queue=qname, arg::exchange=AMQ_FAILOVER);
- subscriptions.subscribe(*this, qname, FlowControl::unlimited());
- thread = sys::Thread(subscriptions);
+ subscriptions->subscribe(*this, qname, FlowControl::unlimited());
+ thread = sys::Thread(*subscriptions);
}
-FailoverListener::~FailoverListener() {
- subscriptions.stop();
+void FailoverListener::stop() {
+ if (subscriptions.get()) subscriptions->stop();
if (thread.id()) thread.join();
+ if (subscriptions.get()) subscriptions->getSession().close();
+ thread=sys::Thread();
+ subscriptions.reset();
+}
+FailoverListener::~FailoverListener() {
+ stop();
}
void FailoverListener::received(Message& msg) {
diff --git a/cpp/src/qpid/client/FailoverListener.h b/cpp/src/qpid/client/FailoverListener.h
index 2c06947300..39bea90bb3 100644
--- a/cpp/src/qpid/client/FailoverListener.h
+++ b/cpp/src/qpid/client/FailoverListener.h
@@ -22,10 +22,7 @@
*
*/
-#include "qpid/client/Connection.h"
-#include "qpid/client/Session.h"
#include "qpid/client/MessageListener.h"
-#include "qpid/client/SubscriptionManager.h"
#include "qpid/Url.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/Thread.h"
@@ -34,22 +31,24 @@
namespace qpid {
namespace client {
+class SubscriptionManager;
+
/**
* @internal Listen for failover updates from the amq.failover exchange.
*/
-class FailoverListener : public MessageListener
-{
+class FailoverListener : public MessageListener {
public:
- FailoverListener(Connection);
+ FailoverListener();
~FailoverListener();
+ void start(const boost::shared_ptr<ConnectionImpl>&);
+ void stop();
+
std::vector<Url> getKnownBrokers() const;
void received(Message& msg);
private:
mutable sys::Mutex lock;
- Connection connection;
- Session session;
- SubscriptionManager subscriptions;
+ std::auto_ptr<SubscriptionManager> subscriptions;
sys::Thread thread;
std::vector<Url> knowBrokers;
};
diff --git a/cpp/src/qpid/client/SessionImpl.cpp b/cpp/src/qpid/client/SessionImpl.cpp
index 2d64492bf7..49dd97e324 100644
--- a/cpp/src/qpid/client/SessionImpl.cpp
+++ b/cpp/src/qpid/client/SessionImpl.cpp
@@ -51,21 +51,19 @@ typedef sys::Monitor::ScopedUnlock UnLock;
typedef sys::ScopedLock<sys::Semaphore> Acquire;
-SessionImpl::SessionImpl(const std::string& name,
- shared_ptr<ConnectionImpl> conn,
- uint16_t ch, uint64_t _maxFrameSize)
+SessionImpl::SessionImpl(const std::string& name, shared_ptr<ConnectionImpl> conn)
: state(INACTIVE),
detachedLifetime(0),
- maxFrameSize(_maxFrameSize),
+ maxFrameSize(conn->getNegotiatedSettings().maxFrameSize),
id(conn->getNegotiatedSettings().username, name.empty() ? Uuid(true).str() : name),
- connection(conn),
- ioHandler(*this),
- channel(ch),
- proxy(ioHandler),
+ connectionShared(conn),
+ connectionWeak(conn),
+ weakPtr(false),
+ proxy(out),
nextIn(0),
nextOut(0)
{
- channel.next = connection.get();
+ channel.next = connectionShared.get();
}
SessionImpl::~SessionImpl() {
@@ -78,7 +76,8 @@ SessionImpl::~SessionImpl() {
state.waitWaiters();
}
}
- connection->erase(channel);
+ boost::shared_ptr<ConnectionImpl> c = connectionWeak.lock();
+ if (c) c->erase(channel);
}
@@ -119,6 +118,8 @@ void SessionImpl::close() //user thread
void SessionImpl::resume(shared_ptr<ConnectionImpl>) // user thread
{
+ // weakPtr sessions should not be resumed.
+ if (weakPtr) return;
throw NotImplementedException("Resume not yet implemented by client!");
}
@@ -251,7 +252,6 @@ void SessionImpl::setExceptionLH(const sys::ExceptionHolder& ex) { // Call with
*/
void SessionImpl::connectionClosed(uint16_t code, const std::string& text) {
setException(createConnectionException(code, text));
- // FIXME aconway 2008-10-07: Should closing a connection detach or close its sessions?
handleClosed();
}
@@ -259,9 +259,7 @@ void SessionImpl::connectionClosed(uint16_t code, const std::string& text) {
* Called by ConnectionImpl to notify active sessions when connection
* is disconnected
*/
-void SessionImpl::connectionBroke(uint16_t _code, const std::string& _text)
-{
- // FIXME aconway 2008-10-07: distinguish disconnect from clean close.
+void SessionImpl::connectionBroke(uint16_t _code, const std::string& _text) {
connectionClosed(_code, _text);
}
@@ -426,14 +424,11 @@ void SessionImpl::handleIn(AMQFrame& frame) // network thread
void SessionImpl::handleOut(AMQFrame& frame) // user thread
{
- connection->expand(frame.encodedSize(), true);
- channel.handle(frame);
-}
-
-void SessionImpl::proxyOut(AMQFrame& frame) // network thread
-{
- connection->expand(frame.encodedSize(), false);
- channel.handle(frame);
+ boost::shared_ptr<ConnectionImpl> c = connectionWeak.lock();
+ if (c) {
+ c->expand(frame.encodedSize(), true);
+ channel.handle(frame);
+ }
}
void SessionImpl::deliver(AMQFrame& frame) // network thread
@@ -602,11 +597,11 @@ void SessionImpl::exception(uint16_t errorCode,
const std::string& description,
const framing::FieldTable& /*errorInfo*/)
{
- QPID_LOG(warning, "Exception received from peer: " << errorCode << ":" << description
- << " [caused by " << commandId << " " << classCode << ":" << commandCode << "]");
-
Lock l(state);
setExceptionLH(createSessionException(errorCode, description));
+ QPID_LOG(warning, "Exception received from broker: " << exceptionHolder.what()
+ << " [caused by " << commandId << " " << classCode << ":" << commandCode << "]");
+
if (detachedLifetime)
setTimeout(0);
}
@@ -648,8 +643,6 @@ void SessionImpl::assertOpen() const
void SessionImpl::handleClosed()
{
- // FIXME aconway 2008-06-12: needs to be set to the correct exception type.
- //
demux.close(exceptionHolder.empty() ? new ClosedException() : exceptionHolder);
results.close();
}
@@ -662,4 +655,16 @@ uint32_t SessionImpl::setTimeout(uint32_t seconds) {
return detachedLifetime;
}
+uint32_t SessionImpl::getTimeout() const {
+ return detachedLifetime;
+}
+
+void SessionImpl::setWeakPtr(bool weak) {
+ weakPtr = weak;
+ if (weakPtr)
+ connectionShared.reset(); // Only keep weak pointer
+ else
+ connectionShared = connectionWeak.lock();
+}
+
}}
diff --git a/cpp/src/qpid/client/SessionImpl.h b/cpp/src/qpid/client/SessionImpl.h
index 49a65ed568..54ace77254 100644
--- a/cpp/src/qpid/client/SessionImpl.h
+++ b/cpp/src/qpid/client/SessionImpl.h
@@ -28,7 +28,8 @@
#include "qpid/SessionId.h"
#include "qpid/SessionState.h"
-#include "qpid/shared_ptr.h"
+#include "boost/shared_ptr.hpp"
+#include "boost/weak_ptr.hpp"
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/ChannelHandler.h"
#include "qpid/framing/SequenceNumber.h"
@@ -63,7 +64,7 @@ class SessionImpl : public framing::FrameHandler::InOutHandler,
private framing::AMQP_ClientOperations::ExecutionHandler
{
public:
- SessionImpl(const std::string& name, shared_ptr<ConnectionImpl>, uint16_t channel, uint64_t maxFrameSize);
+ SessionImpl(const std::string& name, shared_ptr<ConnectionImpl>);
~SessionImpl();
@@ -106,6 +107,11 @@ public:
/** Get timeout in seconds. */
uint32_t getTimeout() const;
+ /** Make this session use a weak_ptr to the ConnectionImpl.
+ * Used for sessions created by the ConnectionImpl itself.
+ */
+ void setWeakPtr(bool weak=true);
+
private:
enum State {
INACTIVE,
@@ -131,7 +137,6 @@ private:
void handleIn(framing::AMQFrame& frame);
void handleOut(framing::AMQFrame& frame);
- void proxyOut(framing::AMQFrame& frame);
void deliver(framing::AMQFrame& frame);
Future sendCommand(const framing::AMQBody&, const framing::MethodContent* = 0);
@@ -175,8 +180,11 @@ private:
const uint64_t maxFrameSize;
const SessionId id;
- shared_ptr<ConnectionImpl> connection;
- framing::FrameHandler::MemFunRef<SessionImpl, &SessionImpl::proxyOut> ioHandler;
+ shared_ptr<ConnectionImpl> connection();
+ shared_ptr<ConnectionImpl> connectionShared;
+ boost::weak_ptr<ConnectionImpl> connectionWeak;
+ bool weakPtr;
+
framing::ChannelHandler channel;
framing::AMQP_ServerProxy::Session proxy;
diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index 3a463d9038..c50e67effa 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -198,10 +198,10 @@ class SubscriptionManager : public sys::Runnable
* Default is to acknowledge every message automatically.
*/
void setAckPolicy(const AckPolicy& autoAck);
- /**
- *
- */
+
AckPolicy& getAckPolicy();
+
+ Session getSession() const { return session; }
};
/** AutoCancel cancels a subscription in its destructor */
View Lorry Controller Status