From fb9ad93a3d422c1e83c998f44c4782f7bf1d1a66 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Fri, 1 Dec 2006 05:11:45 +0000 Subject: 2006-12-01 Jim Meyering This delta imposes two major changes on the C++ hierarchy: - adds autoconf, automake, libtool support - makes the hierarchy flatter and renames a few files (e.g., Queue.h, Queue.cpp) that appeared twice, once under client/ and again under broker/. In the process, I've changed many #include directives, mostly to remove a qpid/ or qpid/framing/ prefix from the file name argument. Although most changes were to .cpp and .h files under qpid/cpp/, there were also several to template files under qpid/gentools, and even one to CppGenerator.java. Nearly all files are moved to a new position in the hierarchy. The new hierarchy looks like this: src # this is the new home of qpidd.cpp tests # all tests are here. See Makefile.am. gen # As before, all generated files go here. lib # This is just a container for the 3 lib dirs: lib/client lib/broker lib/common lib/common/framing lib/common/sys lib/common/sys/posix lib/common/sys/apr build-aux m4 git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@481159 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/lib/broker/BrokerChannel.h | 125 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 cpp/lib/broker/BrokerChannel.h (limited to 'cpp/lib/broker/BrokerChannel.h') diff --git a/cpp/lib/broker/BrokerChannel.h b/cpp/lib/broker/BrokerChannel.h new file mode 100644 index 0000000000..804d6866b1 --- /dev/null +++ b/cpp/lib/broker/BrokerChannel.h @@ -0,0 +1,125 @@ +/* + * + * 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. + * + */ +#ifndef _Channel_ +#define _Channel_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace qpid { + namespace broker { + using qpid::framing::string; + + /** + * Maintains state for an AMQP channel. Handles incoming and + * outgoing messages for that channel. + */ + class Channel : private MessageBuilder::CompletionHandler{ + class ConsumerImpl : public virtual Consumer{ + Channel* parent; + const string tag; + Queue::shared_ptr queue; + ConnectionToken* const connection; + const bool ackExpected; + bool blocked; + public: + ConsumerImpl(Channel* parent, const string& tag, Queue::shared_ptr queue, ConnectionToken* const connection, bool ack); + virtual bool deliver(Message::shared_ptr& msg); + void cancel(); + void requestDispatch(); + }; + + typedef std::map::iterator consumer_iterator; + const int id; + qpid::framing::OutputHandler* out; + u_int64_t currentDeliveryTag; + Queue::shared_ptr defaultQueue; + bool transactional; + std::map consumers; + u_int32_t prefetchSize; + u_int16_t prefetchCount; + Prefetch outstanding; + u_int32_t framesize; + NameGenerator tagGenerator; + std::list unacked; + qpid::sys::Mutex deliveryLock; + TxBuffer txBuffer; + AccumulatedAck accumulatedAck; + TransactionalStore* store; + MessageBuilder messageBuilder;//builder for in-progress message + Exchange::shared_ptr exchange;//exchange to which any in-progress message was published to + + virtual void complete(Message::shared_ptr& msg); + void deliver(Message::shared_ptr& msg, const string& tag, Queue::shared_ptr& queue, bool ackExpected); + void cancel(consumer_iterator consumer); + bool checkPrefetch(Message::shared_ptr& msg); + + public: + Channel(qpid::framing::OutputHandler* out, int id, u_int32_t framesize); + ~Channel(); + inline void setDefaultQueue(Queue::shared_ptr queue){ defaultQueue = queue; } + inline Queue::shared_ptr getDefaultQueue(){ return defaultQueue; } + inline u_int32_t setPrefetchSize(u_int32_t size){ return prefetchSize = size; } + inline u_int16_t setPrefetchCount(u_int16_t count){ return prefetchCount = count; } + bool exists(const string& consumerTag); + void consume(string& tag, Queue::shared_ptr queue, bool acks, bool exclusive, ConnectionToken* const connection = 0); + void cancel(const string& tag); + bool get(Queue::shared_ptr queue, bool ackExpected); + void begin(); + void close(); + void commit(); + void rollback(); + void ack(u_int64_t deliveryTag, bool multiple); + void recover(bool requeue); + void deliver(Message::shared_ptr& msg, const string& consumerTag, u_int64_t deliveryTag); + void handlePublish(Message* msg, Exchange::shared_ptr exchange); + void handleHeader(qpid::framing::AMQHeaderBody::shared_ptr header); + void handleContent(qpid::framing::AMQContentBody::shared_ptr content); + }; + + struct InvalidAckException{}; + } +} + + +#endif -- cgit v1.2.1