Apache Qpid - AMQP Messaging for Java JMS, C++, Python, Ruby, and .NET Apache Qpid Documentation
qpid/console/Broker.h
Go to the documentation of this file.
00001 /*
00002  *
00003  * Licensed to the Apache Software Foundation (ASF) under one
00004  * or more contributor license agreements.  See the NOTICE file
00005  * distributed with this work for additional information
00006  * regarding copyright ownership.  The ASF licenses this file
00007  * to you under the Apache License, Version 2.0 (the
00008  * "License"); you may not use this file except in compliance
00009  * with the License.  You may obtain a copy of the License at
00010  *
00011  *   http://www.apache.org/licenses/LICENSE-2.0
00012  *
00013  * Unless required by applicable law or agreed to in writing,
00014  * software distributed under the License is distributed on an
00015  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
00016  * KIND, either express or implied.  See the License for the
00017  * specific language governing permissions and limitations
00018  * under the License.
00019  *
00020  */
00021 #ifndef _QPID_CONSOLE_BROKER_H_
00022 #define _QPID_CONSOLE_BROKER_H_
00023 
00024 #include "qpid/console/ConsoleImportExport.h"
00025 #include "qpid/client/Connection.h"
00026 #include "qpid/client/ConnectionSettings.h"
00027 #include "qpid/client/SubscriptionManager.h"
00028 #include "qpid/client/Session.h"
00029 #include "qpid/client/AsyncSession.h"
00030 #include "qpid/client/Message.h"
00031 #include "qpid/client/MessageListener.h"
00032 #include "qpid/sys/Thread.h"
00033 #include "qpid/sys/Runnable.h"
00034 #include "qpid/sys/Mutex.h"
00035 #include "qpid/sys/Condition.h"
00036 #include "qpid/Url.h"
00037 #include "qpid/framing/Buffer.h"
00038 #include "qpid/framing/Uuid.h"
00039 #include <string>
00040 #include <iostream>
00041 
00042 namespace qpid {
00043 namespace console {
00044     class SessionManager;
00045     class Agent;
00046     class Object;
00047 
00052     class Broker : public client::MessageListener {
00053     public:
00054         QPID_CONSOLE_EXTERN Broker(SessionManager& sm,
00055                                    client::ConnectionSettings& settings);
00056         QPID_CONSOLE_EXTERN ~Broker();
00057 
00058         QPID_CONSOLE_INLINE_EXTERN bool isConnected() const { return connected; }
00059         QPID_CONSOLE_INLINE_EXTERN const std::string& getError() const { return error; }
00060         QPID_CONSOLE_INLINE_EXTERN const std::string& getSessionId() const { return amqpSessionId; }
00061         QPID_CONSOLE_INLINE_EXTERN const framing::Uuid& getBrokerId() const { return brokerId; }
00062         QPID_CONSOLE_INLINE_EXTERN uint32_t getBrokerBank() const { return 1; }
00063         QPID_CONSOLE_INLINE_EXTERN void addBinding(const std::string& key) {
00064             connThreadBody.bindExchange("qpid.management", key);
00065         }
00066         QPID_CONSOLE_EXTERN std::string getUrl() const;
00067         QPID_CONSOLE_EXTERN void waitForStable();
00068 
00069     private:
00070         friend class SessionManager;
00071         friend class Object;
00072         typedef std::map<uint64_t,Agent*> AgentMap;
00073         static const int SYNC_TIME = 60;
00074 
00075         SessionManager& sessionManager;
00076         AgentMap agents;
00077         bool connected;
00078         std::string error;
00079         std::string amqpSessionId;
00080         client::ConnectionSettings connectionSettings;
00081         sys::Mutex lock;
00082         sys::Condition cond;
00083         framing::Uuid brokerId;
00084         uint32_t reqsOutstanding;
00085         bool syncInFlight;
00086         bool topicBound;
00087         Object* methodObject;
00088 
00089         friend class ConnectionThread;
00090         class ConnectionThread : public sys::Runnable {
00091             bool operational;
00092             bool shuttingDown;
00093             Broker&              broker;
00094             framing::Uuid        sessionId;
00095             client::Connection   connection;
00096             client::Session      session;
00097             client::SubscriptionManager* subscriptions;
00098             std::stringstream queueName;
00099             sys::Mutex        connLock;
00100             void run();
00101         public:
00102             ConnectionThread(Broker& _broker) :
00103                 operational(false), shuttingDown(false), broker(_broker), subscriptions(0) {}
00104             ~ConnectionThread();
00105             void sendBuffer(qpid::framing::Buffer& buf,
00106                             uint32_t               length,
00107                             const std::string&     exchange = "qpid.management",
00108                             const std::string&     routingKey = "broker");
00109             void bindExchange(const std::string& exchange, const std::string& key);
00110             void shutdown();
00111         };
00112 
00113         ConnectionThread connThreadBody;
00114         sys::Thread      connThread;
00115 
00116         void encodeHeader(framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0) const;
00117         bool checkHeader(framing::Buffer& buf, uint8_t *opcode, uint32_t *seq) const;
00118         void received(client::Message& msg);
00119         void resetAgents();
00120         void updateAgent(const Object& object);
00121         void incOutstanding();
00122         void decOutstanding();
00123         void setBrokerId(const framing::Uuid& id) { brokerId = id; }
00124         void appendAgents(std::vector<Agent*>& agents) const;
00125 
00126         friend std::ostream& operator<<(std::ostream& o, const Broker& k);
00127     };
00128 
00129     std::ostream& operator<<(std::ostream& o, const Broker& k);
00130 }
00131 }
00132 
00133 #endif

Qpid C++ API Reference
Generated on Mon Aug 22 2011 for Qpid C++ Client API by doxygen 1.7.4