00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022 #ifndef _ConnectionImpl_
00023 #define _ConnectionImpl_
00024
00025 #include "Bounds.h"
00026 #include "ConnectionHandler.h"
00027
00028 #include "qpid/framing/FrameHandler.h"
00029 #include "qpid/sys/Mutex.h"
00030 #include "qpid/sys/ShutdownHandler.h"
00031 #include "qpid/sys/TimeoutHandler.h"
00032
00033 #include <map>
00034 #include <boost/shared_ptr.hpp>
00035 #include <boost/weak_ptr.hpp>
00036 #include <boost/scoped_ptr.hpp>
00037 #include <boost/enable_shared_from_this.hpp>
00038
00039 namespace qpid {
00040 namespace client {
00041
00042 class Connector;
00043 struct ConnectionSettings;
00044 class SessionImpl;
00045 class FailoverListener;
00046
00047 class ConnectionImpl : public Bounds,
00048 public framing::FrameHandler,
00049 public sys::TimeoutHandler,
00050 public sys::ShutdownHandler,
00051 public boost::enable_shared_from_this<ConnectionImpl>
00052
00053 {
00054 typedef std::map<uint16_t, boost::weak_ptr<SessionImpl> > SessionMap;
00055
00056 static const uint16_t NEXT_CHANNEL;
00057
00058 SessionMap sessions;
00059 ConnectionHandler handler;
00060 boost::scoped_ptr<Connector> connector;
00061 boost::scoped_ptr<FailoverListener> failover;
00062 framing::ProtocolVersion version;
00063 uint16_t nextChannel;
00064 sys::Mutex lock;
00065
00066 boost::intrusive_ptr<qpid::sys::TimerTask> heartbeatTask;
00067
00068 template <class F> void closeInternal(const F&);
00069
00070 void incoming(framing::AMQFrame& frame);
00071 void closed(uint16_t, const std::string&);
00072 void idleOut();
00073 void idleIn();
00074 void shutdown();
00075
00076 boost::function<void ()> failureCallback;
00077
00078 public:
00079 ConnectionImpl(framing::ProtocolVersion version, const ConnectionSettings& settings);
00080 ~ConnectionImpl();
00081
00082 void open();
00083 bool isOpen() const;
00084
00085 boost::shared_ptr<SessionImpl> newSession(const std::string& name, uint32_t timeout, uint16_t channel=NEXT_CHANNEL);
00086 void addSession(const boost::shared_ptr<SessionImpl>&, uint16_t channel=NEXT_CHANNEL);
00087
00088 void close();
00089 void handle(framing::AMQFrame& frame);
00090 void erase(uint16_t channel);
00091 const ConnectionSettings& getNegotiatedSettings();
00092
00093 std::vector<Url> getKnownBrokers();
00094 void registerFailureCallback ( boost::function<void ()> fn ) { failureCallback = fn; }
00095 void stopFailoverListener();
00096
00097 framing::ProtocolVersion getVersion() { return version; }
00098 };
00099
00100 }}
00101
00102
00103 #endif