Cluster prototype: handles client-initiated commands (not dequeues)

Details
 - Cluster.cpp: serializes all frames thru cluster (see below)
 - broker/ConnectionManager: Added handler chain in front of Connection::received.
 - sys::Fork and ForkWithMessage - abstractions for forking with posix impl.
 - tests/ForkedBroker.h: test utility to fork a broker process. 
 - broker/SignalHandler: Encapsulated signal handling from qpidd.cpp
 - Various minor logging & error message improvements to aid debugging.

NB: current impl will not scale. It is functional working starting point so we
can start testing & profiling to find the right optimizations.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@674107 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 227 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index 531817db83..00fb4b9995 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -23,6 +23,7 @@
  */
 
 #include "ConnectionFactory.h"
+#include "ConnectionManager.h"
 #include "ConnectionToken.h"
 #include "DirectExchange.h"
 #include "DtxManager.h"
@@ -120,6 +121,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
     Options& getOptions() { return config; }
 
     SessionManager& getSessionManager() { return sessionManager; }
+    ConnectionManager& getConnectionManager() { return connectionManager; }
 
     management::ManagementObject::shared_ptr GetManagementObject (void) const;
     management::Manageable*                  GetVhostObject      (void) const;
@@ -158,6 +160,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
     ConnectionFactory factory;
     DtxManager dtxManager;
     SessionManager sessionManager;
+    ConnectionManager connectionManager;
     management::ManagementAgent::shared_ptr managementAgent;
     management::Broker::shared_ptr mgmtObject;
     Vhost::shared_ptr              vhostObject;
diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index 9e763f6775..f6d35ff6ca 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -53,7 +53,9 @@ Connection::Connection(ConnectionOutputHandler* out_, Broker& broker_, const std
     isLink(isLink_),
     mgmtClosing(false),
     mgmtId(mgmtId_),
-    links(broker_.getLinks())
+    links(broker_.getLinks()),
+    lastInHandler(*this),
+    inChain(lastInHandler)
 {
     Manageable* parent = broker.GetVhostObject();
 
@@ -86,7 +88,9 @@ Connection::~Connection()
         links.notifyClosed(mgmtId);
 }
 
-void Connection::received(framing::AMQFrame& frame){
+void Connection::received(framing::AMQFrame& frame){ inChain(frame); }
+    
+void Connection::receivedLast(framing::AMQFrame& frame){
     if (frame.getChannel() == 0 && frame.getMethod()) {
         adapter.handle(frame);
     } else {
diff --git a/cpp/src/qpid/broker/Connection.h b/cpp/src/qpid/broker/Connection.h
index 9e713140dd..6b3530366d 100644
--- a/cpp/src/qpid/broker/Connection.h
+++ b/cpp/src/qpid/broker/Connection.h
@@ -56,6 +56,7 @@ class Connection : public sys::ConnectionInputHandler,
 {
   public:
     typedef boost::shared_ptr<Connection> shared_ptr;
+
     Connection(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isLink = false);
     ~Connection ();
 
@@ -90,10 +91,15 @@ class Connection : public sys::ConnectionInputHandler,
     void notifyConnectionForced(const std::string& text);
     void setUserId(const string& uid);
 
+    framing::FrameHandler::Chain& getInChain() { return inChain; } 
+
   private:
     typedef boost::ptr_map<framing::ChannelId, SessionHandler> ChannelMap;
     typedef std::vector<Queue::shared_ptr>::iterator queue_iterator;
 
+    // End of the received handler chain.
+    void receivedLast(framing::AMQFrame& frame);
+
     ChannelMap channels;
     framing::AMQP_ClientProxy::Connection* client;
     ConnectionHandler adapter;
@@ -103,6 +109,9 @@ class Connection : public sys::ConnectionInputHandler,
     boost::function0<void> ioCallback;
     management::Connection::shared_ptr mgmtObject;
     LinkRegistry& links;
+    framing::FrameHandler::MemFunRef<Connection, &Connection::receivedLast> lastInHandler;
+    framing::FrameHandler::Chain inChain;
+
 };
 
 }}
diff --git a/cpp/src/qpid/broker/ConnectionManager.cpp b/cpp/src/qpid/broker/ConnectionManager.cpp
new file mode 100644
index 0000000000..165de5220e
--- /dev/null
+++ b/cpp/src/qpid/broker/ConnectionManager.cpp
@@ -0,0 +1,41 @@
+/*
+ *
+ * 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 "ConnectionManager.h"
+#include "Connection.h"
+
+namespace qpid {
+namespace broker {
+
+std::auto_ptr<Connection>
+ConnectionManager::create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient) {
+    std::auto_ptr<Connection> c(new Connection(out, broker, mgmtId, isClient));
+    sys::Mutex::ScopedLock l(lock);
+    std::for_each(observers.begin(), observers.end(),
+                  boost::bind(&Observer::created, _1, boost::ref(*c)));
+    return c;
+}
+
+void ConnectionManager::add(const boost::intrusive_ptr<Observer>& observer) {
+    sys::Mutex::ScopedLock l(lock);
+    observers.push_back(observer);
+}
+
+}} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/ConnectionManager.h b/cpp/src/qpid/broker/ConnectionManager.h
new file mode 100644
index 0000000000..a999523d0d
--- /dev/null
+++ b/cpp/src/qpid/broker/ConnectionManager.h
@@ -0,0 +1,70 @@
+#ifndef QPID_BROKER_CONNECTIONMANAGER_H
+#define QPID_BROKER_CONNECTIONMANAGER_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 "qpid/RefCounted.h"
+#include "qpid/sys/Mutex.h"
+#include <boost/intrusive_ptr.hpp>
+#include <vector>
+#include <memory>
+
+namespace qpid {
+
+namespace sys {
+class ConnectionOutputHandler;
+}
+
+namespace broker {
+
+class Broker;
+class Connection;
+
+/**
+ * Manages connections and observers.
+ */
+class ConnectionManager {
+  public:
+
+    /**
+     * Observer notified of ConnectionManager events.
+     */
+    struct Observer : public RefCounted {
+        /** Called when a connection is attached. */
+        virtual void created(Connection&) {}
+    };
+
+    /** Called to create a new Connection, applies observers. */
+    std::auto_ptr<Connection> create(sys::ConnectionOutputHandler* out, Broker& broker, const std::string& mgmtId, bool isClient = false);
+
+    /** Add an observer */
+    void add(const boost::intrusive_ptr<Observer>&);
+
+  private:
+    typedef std::vector<boost::intrusive_ptr<Observer> > Observers;
+
+    sys::Mutex lock;
+    Observers observers;
+};
+}} // namespace qpid::broker
+
+#endif  /*!QPID_BROKER_CONNECTIONMANAGER_H*/
diff --git a/cpp/src/qpid/broker/SignalHandler.cpp b/cpp/src/qpid/broker/SignalHandler.cpp
new file mode 100644
index 0000000000..c6d7b10218
--- /dev/null
+++ b/cpp/src/qpid/broker/SignalHandler.cpp
@@ -0,0 +1,51 @@
+/*
+ *
+ * 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 "SignalHandler.h"
+#include "Broker.h"
+#include <signal.h>
+
+namespace qpid {
+namespace broker {
+
+boost::shared_ptr<Broker> SignalHandler::broker;
+
+void SignalHandler::setBroker(const boost::shared_ptr<Broker>& b) {
+    broker = b;
+
+    signal(SIGINT,shutdownHandler); 
+    signal(SIGTERM, shutdownHandler);
+
+    signal(SIGHUP,SIG_IGN); // TODO aconway 2007-07-18: reload config.
+
+    signal(SIGCHLD,SIG_IGN); 
+    signal(SIGTSTP,SIG_IGN); 
+    signal(SIGTTOU,SIG_IGN);
+    signal(SIGTTIN,SIG_IGN);
+}
+
+void SignalHandler::shutdownHandler(int) {
+    if (broker.get()) {
+        broker->shutdown();
+        broker.reset();
+    }
+}
+
+}} // namespace qpid::broker
diff --git a/cpp/src/qpid/broker/SignalHandler.h b/cpp/src/qpid/broker/SignalHandler.h
new file mode 100644
index 0000000000..3da7d09756
--- /dev/null
+++ b/cpp/src/qpid/broker/SignalHandler.h
@@ -0,0 +1,47 @@
+#ifndef QPID_BROKER_SIGNALHANDLER_H
+#define QPID_BROKER_SIGNALHANDLER_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 <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace broker {
+
+class Broker;
+
+/**
+ * Handle signals e.g. to shut-down a broker.
+ */
+class SignalHandler
+{
+  public:
+    /** Set the broker to be shutdown on signals */
+    static void setBroker(const boost::shared_ptr<Broker>& broker);
+
+  private:
+    static void shutdownHandler(int);
+    static boost::shared_ptr<Broker> broker;
+};
+}} // namespace qpid::broker
+
+#endif  /*!QPID_BROKER_SIGNALHANDLER_H*/