summaryrefslogtreecommitdiff
path: root/qpid/cpp/src
diff options
context:
space:
mode:
Diffstat (limited to 'qpid/cpp/src')
-rw-r--r--qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp21
-rw-r--r--qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h7
-rw-r--r--qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp2
-rw-r--r--qpid/cpp/src/qpid/messaging/Connection.cpp12
-rw-r--r--qpid/cpp/src/qpid/messaging/ConnectionImpl.h4
-rw-r--r--qpid/cpp/src/tests/MessagingSessionTests.cpp4
-rw-r--r--qpid/cpp/src/tests/qpid_recv.cpp4
-rw-r--r--qpid/cpp/src/tests/qpid_send.cpp4
-rw-r--r--qpid/cpp/src/tests/qpid_stream.cpp4
9 files changed, 36 insertions, 26 deletions
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
index 5828449d5e..30b75ff4ff 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.cpp
@@ -91,16 +91,17 @@ void convert(const Variant::Map& from, ConnectionSettings& to)
setIfFound(from, "max-frame-size", to.maxFrameSize);
setIfFound(from, "bounds", to.bounds);
- setIfFound(from, "protocol", to.protocol);
+ setIfFound(from, "transport", to.protocol);
}
-ConnectionImpl::ConnectionImpl(const Variant::Map& options) :
+ConnectionImpl::ConnectionImpl(const std::string& url, const Variant::Map& options) :
reconnect(true), timeout(-1), limit(-1),
minReconnectInterval(3), maxReconnectInterval(60),
retries(0)
{
QPID_LOG(debug, "Created connection with " << options);
setOptions(options);
+ urls.push_back(url);
}
void ConnectionImpl::setOptions(const Variant::Map& options)
@@ -127,12 +128,6 @@ void ConnectionImpl::setOption(const std::string& name, const Variant& value)
QPID_LOG(debug, "Set " << name << " to " << value);
}
-void ConnectionImpl::open(const std::string& u)
-{
- urls.push_back(u);
- connect();
-}
-
void ConnectionImpl::close()
{
std::vector<std::string> names;
@@ -143,11 +138,21 @@ void ConnectionImpl::close()
for (std::vector<std::string>::const_iterator i = names.begin(); i != names.end(); ++i) {
getSession(*i).close();
}
+ detach();
+}
+void ConnectionImpl::detach()
+{
qpid::sys::Mutex::ScopedLock l(lock);
connection.close();
}
+bool ConnectionImpl::isConnected()
+{
+ qpid::sys::Mutex::ScopedLock l(lock);
+ return connection.isOpen();
+}
+
boost::intrusive_ptr<SessionImpl> getImplPtr(qpid::messaging::Session& session)
{
return boost::dynamic_pointer_cast<SessionImpl>(
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h b/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
index 2c59d8add8..9d992c1375 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
+++ b/qpid/cpp/src/qpid/client/amqp0_10/ConnectionImpl.h
@@ -39,13 +39,14 @@ class SessionImpl;
class ConnectionImpl : public qpid::messaging::ConnectionImpl
{
public:
- ConnectionImpl(const qpid::types::Variant::Map& options);
- void open(const std::string& url);
+ ConnectionImpl(const std::string& url, const qpid::types::Variant::Map& options);
+ void connect();
+ bool isConnected();
void close();
qpid::messaging::Session newSession(bool transactional, const std::string& name);
qpid::messaging::Session getSession(const std::string& name) const;
void closed(SessionImpl&);
- void connect();
+ void detach();
void setOption(const std::string& name, const qpid::types::Variant& value);
private:
typedef std::map<std::string, qpid::messaging::Session> Sessions;
diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
index 8f9751a967..969ad93da9 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SessionImpl.cpp
@@ -108,7 +108,7 @@ void SessionImpl::close()
for (std::vector<std::string>::const_iterator i = s.begin(); i != s.end(); ++i) getSender(*i).close();
for (std::vector<std::string>::const_iterator i = r.begin(); i != r.end(); ++i) getReceiver(*i).close();
-
+
connection->closed(*this);
session.close();
}
diff --git a/qpid/cpp/src/qpid/messaging/Connection.cpp b/qpid/cpp/src/qpid/messaging/Connection.cpp
index b9437c7931..81a72bb876 100644
--- a/qpid/cpp/src/qpid/messaging/Connection.cpp
+++ b/qpid/cpp/src/qpid/messaging/Connection.cpp
@@ -39,22 +39,24 @@ Connection::Connection(const Connection& c) : Handle<ConnectionImpl>() { PI::cop
Connection& Connection::operator=(const Connection& c) { return PI::assign(*this, c); }
Connection::~Connection() { PI::dtor(*this); }
-Connection::Connection(const std::string& o)
+Connection::Connection(const std::string& url, const std::string& o)
{
Variant::Map options;
AddressParser parser(o);
if (o.empty() || parser.parseMap(options)) {
- PI::ctor(*this, new qpid::client::amqp0_10::ConnectionImpl(options));
+ PI::ctor(*this, new qpid::client::amqp0_10::ConnectionImpl(url, options));
} else {
throw InvalidOptionString(o);
}
}
-Connection::Connection(const Variant::Map& options)
+Connection::Connection(const std::string& url, const Variant::Map& options)
{
- PI::ctor(*this, new qpid::client::amqp0_10::ConnectionImpl(options));
+ PI::ctor(*this, new qpid::client::amqp0_10::ConnectionImpl(url, options));
}
-void Connection::open(const std::string& url) { impl->open(url); }
+void Connection::connect() { impl->connect(); }
+bool Connection::isConnected() { return impl->isConnected(); }
+void Connection::detach() { impl->detach(); }
void Connection::close() { impl->close(); }
Session Connection::createSession(const std::string& name) { return impl->newSession(false, name); }
Session Connection::createTransactionalSession(const std::string& name)
diff --git a/qpid/cpp/src/qpid/messaging/ConnectionImpl.h b/qpid/cpp/src/qpid/messaging/ConnectionImpl.h
index 474f486736..148105476d 100644
--- a/qpid/cpp/src/qpid/messaging/ConnectionImpl.h
+++ b/qpid/cpp/src/qpid/messaging/ConnectionImpl.h
@@ -38,7 +38,9 @@ class ConnectionImpl : public virtual qpid::RefCounted
{
public:
virtual ~ConnectionImpl() {}
- virtual void open(const std::string& url) = 0;
+ virtual void connect() = 0;
+ virtual bool isConnected() = 0;
+ virtual void detach() = 0;
virtual void close() = 0;
virtual Session newSession(bool transactional, const std::string& name) = 0;
virtual Session getSession(const std::string& name) const = 0;
diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
index 6341ef3f28..bdd5422690 100644
--- a/qpid/cpp/src/tests/MessagingSessionTests.cpp
+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
@@ -115,8 +115,8 @@ struct MessagingFixture : public BrokerFixture
static Connection open(uint16_t port)
{
- Connection connection;
- connection.open((boost::format("amqp:tcp:localhost:%1%") % (port)).str());
+ Connection connection((boost::format("amqp:tcp:localhost:%1%") % (port)).str());
+ connection.connect();
return connection;
}
diff --git a/qpid/cpp/src/tests/qpid_recv.cpp b/qpid/cpp/src/tests/qpid_recv.cpp
index 602f91d74e..77e9cd180a 100644
--- a/qpid/cpp/src/tests/qpid_recv.cpp
+++ b/qpid/cpp/src/tests/qpid_recv.cpp
@@ -159,9 +159,9 @@ int main(int argc, char ** argv)
{
Options opts;
if (opts.parse(argc, argv)) {
- Connection connection(opts.connectionOptions);
+ Connection connection(opts.url, opts.connectionOptions);
try {
- connection.open(opts.url);
+ connection.connect();
std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
Session session = opts.tx ? connection.createTransactionalSession() : connection.createSession();
Receiver receiver = session.createReceiver(opts.address);
diff --git a/qpid/cpp/src/tests/qpid_send.cpp b/qpid/cpp/src/tests/qpid_send.cpp
index feeabefbdd..e51c5a93d2 100644
--- a/qpid/cpp/src/tests/qpid_send.cpp
+++ b/qpid/cpp/src/tests/qpid_send.cpp
@@ -216,9 +216,9 @@ int main(int argc, char ** argv)
{
Options opts;
if (opts.parse(argc, argv)) {
- Connection connection(opts.connectionOptions);
+ Connection connection(opts.url, opts.connectionOptions);
try {
- connection.open(opts.url);
+ connection.connect();
std::auto_ptr<FailoverUpdates> updates(opts.failoverUpdates ? new FailoverUpdates(connection) : 0);
Session session = opts.tx ? connection.createTransactionalSession() : connection.createSession();
Sender sender = session.createSender(opts.address);
diff --git a/qpid/cpp/src/tests/qpid_stream.cpp b/qpid/cpp/src/tests/qpid_stream.cpp
index eafa6e96fa..b3fe493922 100644
--- a/qpid/cpp/src/tests/qpid_stream.cpp
+++ b/qpid/cpp/src/tests/qpid_stream.cpp
@@ -88,9 +88,9 @@ struct Client : qpid::sys::Runnable
void run()
{
- Connection connection;
+ Connection connection(opts.url);
try {
- connection.open(opts.url);
+ connection.connect();
Session session = connection.createSession();
doWork(session);
session.close();
View Lorry Controller Status