Author: aconway
Date: Fri May 1 17:05:00 2009
New Revision: 770756
URL: http://svn.apache.org/viewvc?rev=770756&view=rev
Log:
Apply PIMPL pattern to qpid::client::LocalQueue
Added:
qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp (with props)
qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h (with props)
Modified:
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/client/Completion.h
qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.cpp
qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.h
qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=770756&r1=770755&r2=770756&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Fri May 1 17:05:00 2009
@@ -451,6 +451,8 @@
qpid/client/FutureResult.cpp \
qpid/client/LoadPlugins.cpp \
qpid/client/LocalQueue.cpp \
+ qpid/client/LocalQueueImpl.cpp \
+ qpid/client/LocalQueueImpl.h \
qpid/client/Message.cpp \
qpid/client/MessageImpl.cpp \
qpid/client/MessageImpl.h \
Modified: qpid/trunk/qpid/cpp/src/qpid/client/Completion.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/Completion.h?rev=770756&r1=770755&r2=770756&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/Completion.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/Completion.h Fri May 1 17:05:00 2009
@@ -62,7 +62,6 @@
private:
typedef CompletionImpl Impl;
- Impl* impl;
friend class PrivateImplRef<Completion>;
};
Modified: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.cpp?rev=770756&r1=770755&r2=770756&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.cpp Fri May 1 17:05:00 2009
@@ -19,6 +19,7 @@
*
*/
#include "LocalQueue.h"
+#include "LocalQueueImpl.h"
#include "MessageImpl.h"
#include "qpid/Exception.h"
#include "qpid/framing/FrameSet.h"
@@ -26,56 +27,26 @@
#include "qpid/framing/reply_exceptions.h"
#include "PrivateImplRef.h"
#include "SubscriptionImpl.h"
-#include "CompletionImpl.h"
namespace qpid {
namespace client {
using namespace framing;
-LocalQueue::LocalQueue() {}
-LocalQueue::~LocalQueue() {}
+typedef PrivateImplRef<LocalQueue> PI;
-Message LocalQueue::pop(sys::Duration timeout) { return get(timeout); }
+LocalQueue::LocalQueue() { PI::ctor(*this, new LocalQueueImpl()); }
+LocalQueue::LocalQueue(const LocalQueue& x) : Handle<LocalQueueImpl>() {
PI::copy(*this, x); }
+LocalQueue::~LocalQueue() { PI::dtor(*this); }
+LocalQueue& LocalQueue::operator=(const LocalQueue& x) { return
PI::assign(*this, x); }
-Message LocalQueue::get(sys::Duration timeout) {
- Message result;
- bool ok = get(result, timeout);
- if (!ok) throw Exception("Timed out waiting for a message");
- return result;
-}
-
-bool LocalQueue::get(Message& result, sys::Duration timeout) {
- if (!queue)
- throw ClosedException();
- FrameSet::shared_ptr content;
- bool ok = queue->pop(content, timeout);
- if (!ok) return false;
- if (content->isA<MessageTransferBody>()) {
-
- *MessageImpl::get(result) = MessageImpl(*content);
- boost::intrusive_ptr<SubscriptionImpl> si =
PrivateImplRef<Subscription>::get(subscription);
- assert(si);
- if (si) si->received(result);
- return true;
- }
- else
- throw CommandInvalidException(
- QPID_MSG("Unexpected method: " << content->getMethod()));
-}
-
-bool LocalQueue::empty() const
-{
- if (!queue)
- throw ClosedException();
- return queue->empty();
-}
-
-size_t LocalQueue::size() const
-{
- if (!queue)
- throw ClosedException();
- return queue->size();
-}
+Message LocalQueue::pop(sys::Duration timeout) { return impl->pop(timeout); }
+
+Message LocalQueue::get(sys::Duration timeout) { return impl->get(timeout); }
+
+bool LocalQueue::get(Message& result, sys::Duration timeout) { return
impl->get(result, timeout); }
+
+bool LocalQueue::empty() const { return impl->empty(); }
+size_t LocalQueue::size() const { return impl->size(); }
}} // namespace qpid::client
Modified: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.h?rev=770756&r1=770755&r2=770756&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/LocalQueue.h Fri May 1 17:05:00 2009
@@ -23,14 +23,16 @@
*/
#include "ClientImportExport.h"
-#include "qpid/client/Message.h"
-#include "qpid/client/Subscription.h"
-#include "qpid/client/Demux.h"
+#include "Handle.h"
+#include "Message.h"
#include "qpid/sys/Time.h"
namespace qpid {
namespace client {
+class LocalQueueImpl;
+template <class T> class PrivateImplRef;
+
/**
* A local queue to collect messages retrieved from a remote broker
* queue. Create a queue and subscribe it using the SubscriptionManager.
@@ -69,7 +71,7 @@
* </ul>
*/
-class LocalQueue {
+class LocalQueue : public Handle<LocalQueueImpl> {
public:
/** Create a local queue. Subscribe the local queue to a remote broker
* queue with a SubscriptionManager.
@@ -77,8 +79,9 @@
* LocalQueue is an alternative to implementing a MessageListener.
*/
QPID_CLIENT_EXTERN LocalQueue();
-
+ QPID_CLIENT_EXTERN LocalQueue(const LocalQueue&);
QPID_CLIENT_EXTERN ~LocalQueue();
+ QPID_CLIENT_EXTERN LocalQueue& operator=(const LocalQueue&);
/** Wait up to timeout for the next message from the local queue.
*...@param result Set to the message from the queue.
@@ -104,11 +107,12 @@
/** Number of messages on the local queue */
QPID_CLIENT_EXTERN size_t size() const;
- private:
- Demux::QueuePtr queue;
- Subscription subscription;
+ QPID_CLIENT_EXTERN LocalQueue(LocalQueueImpl*); ///<@internal
- friend class SubscriptionManager;
+
+ private:
+ typedef LocalQueueImpl Impl;
+ friend class PrivateImplRef<LocalQueue>;
};
}} // namespace qpid::client
Added: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp?rev=770756&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp Fri May 1 17:05:00
2009
@@ -0,0 +1,78 @@
+/*
+ *
+ * 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 "LocalQueueImpl.h"
+#include "MessageImpl.h"
+#include "qpid/Exception.h"
+#include "qpid/framing/FrameSet.h"
+#include "qpid/framing/MessageTransferBody.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "PrivateImplRef.h"
+#include "SubscriptionImpl.h"
+#include "CompletionImpl.h"
+
+namespace qpid {
+namespace client {
+
+using namespace framing;
+
+Message LocalQueueImpl::pop(sys::Duration timeout) { return get(timeout); }
+
+Message LocalQueueImpl::get(sys::Duration timeout) {
+ Message result;
+ bool ok = get(result, timeout);
+ if (!ok) throw Exception("Timed out waiting for a message");
+ return result;
+}
+
+bool LocalQueueImpl::get(Message& result, sys::Duration timeout) {
+ if (!queue)
+ throw ClosedException();
+ FrameSet::shared_ptr content;
+ bool ok = queue->pop(content, timeout);
+ if (!ok) return false;
+ if (content->isA<MessageTransferBody>()) {
+
+ *MessageImpl::get(result) = MessageImpl(*content);
+ boost::intrusive_ptr<SubscriptionImpl> si =
PrivateImplRef<Subscription>::get(subscription);
+ assert(si);
+ if (si) si->received(result);
+ return true;
+ }
+ else
+ throw CommandInvalidException(
+ QPID_MSG("Unexpected method: " << content->getMethod()));
+}
+
+bool LocalQueueImpl::empty() const
+{
+ if (!queue)
+ throw ClosedException();
+ return queue->empty();
+}
+
+size_t LocalQueueImpl::size() const
+{
+ if (!queue)
+ throw ClosedException();
+ return queue->size();
+}
+
+}} // namespace qpid::client
Propchange: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h?rev=770756&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h Fri May 1 17:05:00
2009
@@ -0,0 +1,108 @@
+#ifndef QPID_CLIENT_LOCALQUEUEIMPL_H
+#define QPID_CLIENT_LOCALQUEUEIMPL_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 "ClientImportExport.h"
+#include "Handle.h"
+#include "qpid/client/Message.h"
+#include "qpid/client/Subscription.h"
+#include "qpid/client/Demux.h"
+#include "qpid/sys/Time.h"
+#include "qpid/RefCounted.h"
+
+namespace qpid {
+namespace client {
+
+/**
+ * A local queue to collect messages retrieved from a remote broker
+ * queue. Create a queue and subscribe it using the SubscriptionManager.
+ * Messages from the remote queue on the broker will be stored in the
+ * local queue until you retrieve them.
+ *
+ * \ingroup clientapi
+ *
+ * \details Using a Local Queue
+ *
+ * <pre>
+ * LocalQueue local_queue;
+ * subscriptions.subscribe(local_queue, string("message_queue"));
+ * for (int i=0; i<10; i++) {
+ * Message message = local_queue.get();
+ * std::cout << message.getData() << std::endl;
+ * }
+ * </pre>
+ *
+ * <h2>Getting Messages</h2>
+ *
+ * <ul><li>
+ * <p>get()</p>
+ * <pre>Message message = local_queue.get();</pre>
+ * <pre>// Specifying timeouts (TIME_SEC, TIME_MSEC, TIME_USEC, TIME_NSEC)
+ *#include <qpid/sys/Time.h>
+ *Message message;
+ *local_queue.get(message, 5*sys::TIME_SEC);</pre></li></ul>
+ *
+ * <h2>Checking size</h2>
+ * <ul><li>
+ * <p>empty()</p>
+ * <pre>if (local_queue.empty()) { ... }</pre></li>
+ * <li><p>size()</p>
+ * <pre>std::cout << local_queue.size();</pre></li>
+ * </ul>
+ */
+
+class LocalQueueImpl : public RefCounted {
+ public:
+ /** Wait up to timeout for the next message from the local queue.
+ *...@param result Set to the message from the queue.
+ *...@param timeout wait up this timeout for a message to appear.
+ *...@return true if result was set, false if queue was empty after
timeout.
+ */
+ bool get(Message& result, sys::Duration timeout=0);
+
+ /** Get the next message off the local queue, or wait up to the timeout
+ * for message from the broker queue.
+ *...@param timeout wait up this timeout for a message to appear.
+ *...@return message from the queue.
+ *...@throw ClosedException if subscription is closed or timeout exceeded.
+ */
+ Message get(sys::Duration timeout=sys::TIME_INFINITE);
+
+ /** Synonym for get() */
+ Message pop(sys::Duration timeout=sys::TIME_INFINITE);
+
+ /** Return true if local queue is empty. */
+ bool empty() const;
+
+ /** Number of messages on the local queue */
+ size_t size() const;
+
+ private:
+ Demux::QueuePtr queue;
+ Subscription subscription;
+ friend class SubscriptionManager;
+};
+
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_LOCALQUEUEIMPL_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/client/LocalQueueImpl.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp?rev=770756&r1=770755&r2=770756&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp Fri May 1
17:05:00 2009
@@ -23,6 +23,8 @@
#include "SubscriptionManager.h"
#include "SubscriptionImpl.h"
+#include "LocalQueueImpl.h"
+#include "PrivateImplRef.h"
#include <qpid/client/Dispatcher.h>
#include <qpid/client/Session.h>
#include <qpid/client/MessageListener.h>
@@ -56,10 +58,11 @@
sys::Mutex::ScopedLock l(lock);
std::string name=n.empty() ? q:n;
boost::intrusive_ptr<SubscriptionImpl> si = new SubscriptionImpl(*this, q,
ss, name, 0);
- lq.queue=si->divert();
+ boost::intrusive_ptr<LocalQueueImpl> lqi =
PrivateImplRef<LocalQueue>::get(lq);
+ lqi->queue=si->divert();
si->subscribe();
- lq.subscription = Subscription(si.get());
- return subscriptions[name] = lq.subscription;
+ lqi->subscription = Subscription(si.get());
+ return subscriptions[name] = lqi->subscription;
}
Subscription SubscriptionManager::subscribe(
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
