summaryrefslogtreecommitdiff
path: root/cpp/src/qpid
diff options
context:
space:
mode:
Diffstat (limited to 'cpp/src/qpid')
-rw-r--r--cpp/src/qpid/client/Channel.cpp289
-rw-r--r--cpp/src/qpid/client/Channel.h316
-rw-r--r--cpp/src/qpid/client/Connection.cpp8
-rw-r--r--cpp/src/qpid/client/Connection.h13
4 files changed, 0 insertions, 626 deletions
diff --git a/cpp/src/qpid/client/Channel.cpp b/cpp/src/qpid/client/Channel.cpp
deleted file mode 100644
index 3bcba8983c..0000000000
--- a/cpp/src/qpid/client/Channel.cpp
+++ /dev/null
@@ -1,289 +0,0 @@
-/*
- *
- * 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/log/Statement.h"
-#include <iostream>
-#include <sstream>
-#include "Channel.h"
-#include "qpid/sys/Monitor.h"
-#include "AckPolicy.h"
-#include "Message.h"
-#include "Connection.h"
-#include "Demux.h"
-#include "MessageListener.h"
-#include "MessageQueue.h"
-#include <boost/format.hpp>
-#include <boost/bind.hpp>
-#include "qpid/framing/all_method_bodies.h"
-#include "qpid/framing/reply_exceptions.h"
-
-using namespace std;
-using namespace boost;
-using namespace qpid::framing;
-using namespace qpid::sys;
-
-namespace qpid{
-namespace client{
-using namespace arg;
-
-const std::string empty;
-
-class ScopedSync
-{
- Session& session;
- const bool change;
- const bool value;
- public:
- ScopedSync(Session& s, bool desired = true) : session(s), change(s.isSynchronous() != desired), value(desired)
- {
- if (change) session.setSynchronous(value);
- }
- ~ScopedSync()
- {
- if (change) session.setSynchronous(!value);
- }
-};
-
-Channel::Channel(bool _transactional, u_int16_t _prefetch) :
- prefetch(_prefetch), transactional(_transactional), running(false),
- uniqueId(true)/*could eventually be the session id*/, nameCounter(0), active(false)
-{
-}
-
-Channel::~Channel()
-{
- join();
-}
-
-void Channel::open(const Session& s)
-{
- Mutex::ScopedLock l(stopLock);
- if (isOpen())
- throw SessionBusyException();
- active = true;
- session = s;
- if(isTransactional()) {
- session.txSelect();
- }
-}
-
-bool Channel::isOpen() const {
- Mutex::ScopedLock l(stopLock);
- return active;
-}
-
-void Channel::setPrefetch(uint32_t _prefetch){
- prefetch = _prefetch;
-}
-
-void Channel::declareExchange(Exchange& _exchange, bool synch){
- ScopedSync s(session, synch);
- session.exchangeDeclare(exchange=_exchange.getName(), type=_exchange.getType());
-}
-
-void Channel::deleteExchange(Exchange& _exchange, bool synch){
- ScopedSync s(session, synch);
- session.exchangeDelete(exchange=_exchange.getName(), ifUnused=false);
-}
-
-void Channel::declareQueue(Queue& _queue, bool synch){
- if (_queue.getName().empty()) {
- stringstream uniqueName;
- uniqueName << uniqueId << "-queue-" << ++nameCounter;
- _queue.setName(uniqueName.str());
- }
-
- ScopedSync s(session, synch);
- session.queueDeclare(queue=_queue.getName(), passive=false/*passive*/, durable=_queue.isDurable(),
- exclusive=_queue.isExclusive(), autoDelete=_queue.isAutoDelete());
-
-}
-
-void Channel::deleteQueue(Queue& _queue, bool ifunused, bool ifempty, bool synch){
- ScopedSync s(session, synch);
- session.queueDelete(queue=_queue.getName(), ifUnused=ifunused, ifEmpty=ifempty);
-}
-
-void Channel::bind(const Exchange& exchange, const Queue& queue, const std::string& key, const FieldTable& args, bool synch){
- string e = exchange.getName();
- string q = queue.getName();
- ScopedSync s(session, synch);
- session.exchangeBind(q, e, key, args);
-}
-
-void Channel::commit(){
- session.txCommit();
-}
-
-void Channel::rollback(){
- session.txRollback();
-}
-
-void Channel::consume(
- Queue& _queue, const std::string& tag, MessageListener* listener,
- AckMode ackMode, bool noLocal, bool synch, FieldTable* fields) {
-
- if (tag.empty()) {
- throw Exception("A tag must be specified for a consumer.");
- }
- {
- Mutex::ScopedLock l(lock);
- ConsumerMap::iterator i = consumers.find(tag);
- if (i != consumers.end())
- throw PreconditionFailedException(QPID_MSG("Consumer already exists with tag " << tag ));
- Consumer& c = consumers[tag];
- c.listener = listener;
- c.ackMode = ackMode;
- c.count = 0;
- }
- uint8_t confirmMode = ackMode == NO_ACK ? 1 : 0;
- ScopedSync s(session, synch);
- FieldTable ft;
- FieldTable* ftptr = fields ? fields : &ft;
- if (noLocal) {
- ftptr->setString("qpid.no-local","yes");
- }
- session.messageSubscribe(_queue.getName(), tag,
- confirmMode, 0/*pre-acquire*/,
- false, "", 0, *ftptr);
- if (!prefetch) {
- session.messageSetFlowMode(tag, 0/*credit based*/);
- }
-
- //allocate some credit:
- session.messageFlow(tag, 1/*BYTES*/, 0xFFFFFFFF);
- session.messageFlow(tag, 0/*MESSAGES*/, prefetch ? prefetch : 0xFFFFFFFF);
-}
-
-void Channel::cancel(const std::string& tag, bool synch) {
- Consumer c;
- {
- Mutex::ScopedLock l(lock);
- ConsumerMap::iterator i = consumers.find(tag);
- if (i == consumers.end())
- return;
- c = i->second;
- consumers.erase(i);
- }
- ScopedSync s(session, synch);
- session.messageCancel(tag);
-}
-
-bool Channel::get(Message& msg, const Queue& _queue, AckMode ackMode) {
- string tag = "get-handler";
- ScopedDivert handler(tag, session.getExecution().getDemux());
- Demux::QueuePtr incoming = handler.getQueue();
-
- session.messageSubscribe(destination=tag, queue=_queue.getName(), acceptMode=(ackMode == NO_ACK ? 1 : 0));
- session.messageFlow(tag, 1/*BYTES*/, 0xFFFFFFFF);
- session.messageFlow(tag, 0/*MESSAGES*/, 1);
- {
- ScopedSync s(session);
- session.messageFlush(tag);
- }
- session.messageCancel(tag);
-
- FrameSet::shared_ptr p;
- if (incoming->tryPop(p)) {
- msg.populate(*p);
- if (ackMode == AUTO_ACK) {
- AckPolicy acker;
- acker.ack(msg, session);
- } else {
- session.markCompleted(msg.getId(), false, false);
- }
- return true;
- }
- else
- return false;
-}
-
-void Channel::publish(Message& msg, const Exchange& exchange,
- const std::string& routingKey,
- bool mandatory, bool /*?TODO-restore immediate?*/) {
-
- msg.getDeliveryProperties().setRoutingKey(routingKey);
- msg.getDeliveryProperties().setDiscardUnroutable(!mandatory);
- session.messageTransfer(destination=exchange.getName(), content=msg);
-}
-
-void Channel::close()
-{
- session.close();
- {
- Mutex::ScopedLock l(stopLock);
- active = false;
- }
- stop();
-}
-
-void Channel::start(){
- running = true;
- dispatcher = Thread(*this);
-}
-
-void Channel::stop() {
- gets.close();
- join();
-}
-
-void Channel::join() {
- Mutex::ScopedLock l(stopLock);
- if(running && dispatcher.id()) {
- dispatcher.join();
- running = false;
- }
-}
-
-void Channel::dispatch(FrameSet& content, const std::string& destination)
-{
- ConsumerMap::iterator i = consumers.find(destination);
- if (i != consumers.end()) {
- Message msg;
- msg.populate(content);
- MessageListener* listener = i->second.listener;
- listener->received(msg);
- if (isOpen() && i->second.ackMode != CLIENT_ACK) {
- bool send = i->second.ackMode == AUTO_ACK
- || (prefetch && ++(i->second.count) > (prefetch / 2));
- if (send) i->second.count = 0;
- session.markCompleted(content.getId(), true, send);
- }
- } else {
- QPID_LOG(warning, "Dropping message for unrecognised consumer: " << destination);
- }
-}
-
-void Channel::run() {
- try {
- while (true) {
- FrameSet::shared_ptr content = session.get();
- //need to dispatch this to the relevant listener:
- if (content->isA<MessageTransferBody>()) {
- dispatch(*content, content->as<MessageTransferBody>()->getDestination());
- } else {
- QPID_LOG(warning, "Dropping unsupported message type: " << content->getMethod());
- }
- }
- } catch (const ClosedException&) {}
-}
-
-}}
-
diff --git a/cpp/src/qpid/client/Channel.h b/cpp/src/qpid/client/Channel.h
deleted file mode 100644
index 9a22c455c4..0000000000
--- a/cpp/src/qpid/client/Channel.h
+++ /dev/null
@@ -1,316 +0,0 @@
-#ifndef _client_Channel_h
-#define _client_Channel_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 <memory>
-#include <boost/scoped_ptr.hpp>
-#include "qpid/framing/amqp_framing.h"
-#include "qpid/framing/Uuid.h"
-#include "Exchange.h"
-#include "Message.h"
-#include "Queue.h"
-#include "ConnectionImpl.h"
-#include "qpid/client/Session.h"
-#include "qpid/Exception.h"
-#include "qpid/sys/Mutex.h"
-#include "qpid/sys/Runnable.h"
-#include "qpid/sys/Thread.h"
-#include "AckMode.h"
-
-namespace qpid {
-
-namespace framing {
-class ChannelCloseBody;
-class AMQMethodBody;
-}
-
-namespace client {
-
-class Connection;
-class MessageChannel;
-class MessageListener;
-class ReturnedMessageHandler;
-
-/**
- * THIS CLASS IS DEPRECATED AND WILL BE SHORTLY REMOVED
- *
- * Represents an AMQP channel, i.e. loosely a session of work. It
- * is through a channel that most of the AMQP 'methods' are
- * exposed.
- *
- */
-class Channel : private sys::Runnable
-{
- private:
- struct Consumer{
- MessageListener* listener;
- AckMode ackMode;
- uint32_t count;
- };
- typedef std::map<std::string, Consumer> ConsumerMap;
-
- mutable sys::Mutex lock;
- sys::Thread dispatcher;
-
- uint32_t prefetch;
- const bool transactional;
- framing::ProtocolVersion version;
-
- mutable sys::Mutex stopLock;
- bool running;
-
- ConsumerMap consumers;
- Session session;
- framing::ChannelId channelId;
- sys::BlockingQueue<framing::FrameSet::shared_ptr> gets;
- framing::Uuid uniqueId;
- uint32_t nameCounter;
- bool active;
-
- void stop();
-
- void open(const Session& session);
- void closeInternal();
- void join();
-
- void dispatch(framing::FrameSet& msg, const std::string& destination);
-
- friend class Connection;
-
- public:
- /**
- * Creates a channel object.
- *
- * @param transactional if true, the publishing and acknowledgement
- * of messages will be transactional and can be committed or
- * aborted in atomic units (@see commit(), @see rollback()).
- *
- * @param prefetch specifies the number of unacknowledged
- * messages the channel is willing to have sent to it
- * asynchronously.
- */
- Channel(bool transactional = false, u_int16_t prefetch = 0);
-
- ~Channel();
-
- /**
- * Declares an exchange.
- *
- * In AMQP, Exchanges are the destinations to which messages
- * are published. They have Queues bound to them and route
- * messages they receive to those queues. The routing rules
- * depend on the type of the exchange.
- *
- * @param exchange an Exchange object representing the
- * exchange to declare.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void declareExchange(Exchange& exchange, bool synch = true);
- /**
- * Deletes an exchange.
- *
- * @param exchange an Exchange object representing the exchange to delete.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void deleteExchange(Exchange& exchange, bool synch = true);
- /**
- * Declares a Queue.
- *
- * @param queue a Queue object representing the queue to declare.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void declareQueue(Queue& queue, bool synch = true);
- /**
- * Deletes a Queue.
- *
- * @param queue a Queue object representing the queue to delete.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void deleteQueue(Queue& queue, bool ifunused = false, bool ifempty = false, bool synch = true);
- /**
- * Binds a queue to an exchange. The exact semantics of this
- * (in particular how 'routing keys' and 'binding arguments'
- * are used) depends on the type of the exchange.
- *
- * @param exchange an Exchange object representing the
- * exchange to bind to.
- *
- * @param queue a Queue object representing the queue to be
- * bound.
- *
- * @param key the 'routing key' for the binding.
- *
- * @param args the 'binding arguments' for the binding.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void bind(const Exchange& exchange, const Queue& queue,
- const std::string& key,
- const framing::FieldTable& args=framing::FieldTable(),
- bool synch = true);
-
- /**
- * For a transactional channel this will commit all
- * publications and acknowledgements since the last commit (or
- * the channel that was opened if there has been no previous
- * commit). This will cause published messages to become
- * available to consumers and acknowledged messages to be
- * consumed and removed from the queues they were dispatched
- * from.
- *
- * A channel is specified as transactional or not when the channel
- * object is created (@see Channel()).
- */
- void commit();
-
- /**
- * For a transactional channel, this will rollback any
- * publications or acknowledgements. It will be as if the
- * published messages were never sent and the acknowledged
- * messages were never consumed.
- */
- void rollback();
-
- /**
- * Change the prefetch in use.
- */
- void setPrefetch(uint32_t prefetch);
-
- uint32_t getPrefetch() { return prefetch; }
-
- /**
- * Start message dispatching on a new thread.
- */
- void start();
-
- /**
- * Close the channel. Closing a channel that is not open has no
- * effect.
- */
- void close();
-
- /** True if the channel is transactional. */
- bool isTransactional() { return transactional; }
-
- /** True if the channel is open. */
- bool isOpen() const;
-
- /** Return the protocol version. */
- framing::ProtocolVersion getVersion() const { return version ; }
-
- /**
- * Creates a 'consumer' for a queue. Messages in (or arriving
- * at) that queue will be delivered to consumers
- * asynchronously.
- *
- * @param queue a Queue instance representing the queue to
- * consume from.
- *
- * @param tag an identifier to associate with the consumer
- * that can be used to cancel its subscription (if empty, this
- * will be assigned by the broker).
- *
- * @param listener a pointer to an instance of an
- * implementation of the MessageListener interface. Messages
- * received from this queue for this consumer will result in
- * invocation of the received() method on the listener, with
- * the message itself passed in.
- *
- * @param ackMode the mode of acknowledgement that the broker
- * should assume for this consumer. @see AckMode
- *
- * @param noLocal if true, this consumer will not be sent any
- * message published by this connection.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void consume(
- Queue& queue, const std::string& tag, MessageListener* listener,
- AckMode ackMode = NO_ACK, bool noLocal = false, bool synch = true,
- framing::FieldTable* fields = 0);
-
- /**
- * Cancels a subscription previously set up through a call to consume().
- *
- * @param tag the identifier used (or assigned) in the consume
- * request that set up the subscription to be cancelled.
- *
- * @param synch if true this call will block until a response
- * is received from the broker.
- */
- void cancel(const std::string& tag, bool synch = true);
- /**
- * Synchronous pull of a message from a queue.
- *
- * @param msg a message object that will contain the message
- * headers and content if the call completes.
- *
- * @param queue the queue to consume from.
- *
- * @param ackMode the acknowledgement mode to use (@see
- * AckMode)
- *
- * @return true if a message was succcessfully dequeued from
- * the queue, false if the queue was empty.
- */
- bool get(Message& msg, const Queue& queue, AckMode ackMode = NO_ACK);
-
- /**
- * Publishes (i.e. sends a message to the broker).
- *
- * @param msg the message to publish.
- *
- * @param exchange the exchange to publish the message to.
- *
- * @param routingKey the routing key to publish with.
- *
- * @param mandatory if true and the exchange to which this
- * publish is directed has no matching bindings, the message
- * will be returned (see setReturnedMessageHandler()).
- *
- * @param immediate if true and there is no consumer to
- * receive this message on publication, the message will be
- * returned (see setReturnedMessageHandler()).
- */
- void publish(Message& msg, const Exchange& exchange,
- const std::string& routingKey,
- bool mandatory = false, bool immediate = false);
-
- /**
- * Deliver incoming messages to the appropriate MessageListener.
- */
- void run();
-};
-
-}}
-
-#endif /*!_client_Channel_h*/
diff --git a/cpp/src/qpid/client/Connection.cpp b/cpp/src/qpid/client/Connection.cpp
index 4089ad79ce..82d1eac8b4 100644
--- a/cpp/src/qpid/client/Connection.cpp
+++ b/cpp/src/qpid/client/Connection.cpp
@@ -20,7 +20,6 @@
*/
#include "Connection.h"
#include "ConnectionSettings.h"
-#include "Channel.h"
#include "Message.h"
#include "SessionImpl.h"
#include "qpid/log/Logger.h"
@@ -73,13 +72,6 @@ void Connection::open(const ConnectionSettings& settings)
max_frame_size = impl->getNegotiatedSettings().maxFrameSize;
}
-void Connection::openChannel(Channel& channel)
-{
- if (!impl)
- throw Exception(QPID_MSG("Connection has not yet been opened"));
- channel.open(newSession(ASYNC));
-}
-
Session Connection::newSession(SynchronousMode sync,
uint32_t detachedLifetime)
{
diff --git a/cpp/src/qpid/client/Connection.h b/cpp/src/qpid/client/Connection.h
index 417739fd1d..0c01c77509 100644
--- a/cpp/src/qpid/client/Connection.h
+++ b/cpp/src/qpid/client/Connection.h
@@ -23,7 +23,6 @@
*/
#include <map>
#include <string>
-#include "Channel.h"
#include "ConnectionImpl.h"
#include "qpid/client/Session.h"
#include "qpid/framing/AMQP_HighestVersion.h"
@@ -108,18 +107,6 @@ class Connection
void close();
/**
- * Associate a Channel with this connection and open it for use.
- *
- * In AMQP, channels are like multiplexed 'sessions' of work over
- * a connection. Almost all the interaction with AMQP is done over
- * a channel.
- *
- * @param connection the connection object to be associated with
- * the channel. Call Channel::close() to close the channel.
- */
- void openChannel(Channel&);
-
- /**
* Create a new session on this connection. Sessions allow
* multiple streams of work to be multiplexed over the same
* connection.
View Lorry Controller Status