Some refactoring of the 0-10 codepath (being migrated to final spec) that primarily colocates the current session and execution layers to facilitate implementing the new session layer that will now encompass this behaviour.

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

1 files changed, 112 insertions, 0 deletions

diff --git a/cpp/src/qpid/broker/PreviewSessionManager.cpp b/cpp/src/qpid/broker/PreviewSessionManager.cpp
new file mode 100644
index 0000000000..ec73082817
--- /dev/null
+++ b/cpp/src/qpid/broker/PreviewSessionManager.cpp
@@ -0,0 +1,112 @@
+/*
+ *
+ * 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 "PreviewSessionManager.h"
+#include "PreviewSessionState.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/log/Statement.h"
+#include "qpid/log/Helpers.h"
+#include "qpid/memory.h"
+
+#include <boost/bind.hpp>
+#include <boost/range.hpp>
+
+#include <algorithm>
+#include <functional>
+#include <ostream>
+
+namespace qpid {
+namespace broker {
+
+using namespace sys;
+using namespace framing;
+
+PreviewSessionManager::PreviewSessionManager(uint32_t a) : ack(a) {}
+
+PreviewSessionManager::~PreviewSessionManager() {}
+
+// FIXME aconway 2008-02-01: pass handler*, allow open  unattached.
+std::auto_ptr<PreviewSessionState>  PreviewSessionManager::open(
+    PreviewSessionHandler& h, uint32_t timeout_)
+{
+    Mutex::ScopedLock l(lock);
+    std::auto_ptr<PreviewSessionState> session(
+        new PreviewSessionState(this, &h, timeout_, ack));
+    active.insert(session->getId());
+    for_each(observers.begin(), observers.end(),
+             boost::bind(&Observer::opened, _1,boost::ref(*session)));
+    return session;
+}
+
+void  PreviewSessionManager::suspend(std::auto_ptr<PreviewSessionState> session) {
+    Mutex::ScopedLock l(lock);
+    active.erase(session->getId());
+    session->suspend();
+    session->expiry = AbsTime(now(),session->getTimeout()*TIME_SEC);
+    if (session->mgmtObject.get() != 0)
+        session->mgmtObject->set_expireTime ((uint64_t) Duration (session->expiry));
+    suspended.push_back(session.release()); // In expiry order
+    eraseExpired();
+}
+
+std::auto_ptr<PreviewSessionState>  PreviewSessionManager::resume(const Uuid& id)
+{
+    Mutex::ScopedLock l(lock);
+    eraseExpired();
+    if (active.find(id) != active.end()) 
+        throw SessionBusyException(
+            QPID_MSG("Session already active: " << id));
+    Suspended::iterator i = std::find_if(
+        suspended.begin(), suspended.end(),
+        boost::bind(std::equal_to<Uuid>(), id, boost::bind(&PreviewSessionState::getId, _1))
+    );
+    if (i == suspended.end())
+        throw InvalidArgumentException(
+            QPID_MSG("No suspended session with id=" << id));
+    active.insert(id);
+    std::auto_ptr<PreviewSessionState> state(suspended.release(i).release());
+    return state;
+}
+
+void PreviewSessionManager::erase(const framing::Uuid& id)
+{
+    Mutex::ScopedLock l(lock);
+    active.erase(id);
+}
+
+void PreviewSessionManager::eraseExpired() {
+    // Called with lock held.
+    if (!suspended.empty()) {
+        Suspended::iterator keep = std::lower_bound(
+            suspended.begin(), suspended.end(), now(),
+            boost::bind(std::less<AbsTime>(), boost::bind(&PreviewSessionState::expiry, _1), _2));
+        if (suspended.begin() != keep) {
+            QPID_LOG(debug, "Expiring sessions: " << log::formatList(suspended.begin(), keep));
+            suspended.erase(suspended.begin(), keep);
+        }
+    }
+}
+
+void PreviewSessionManager::add(const intrusive_ptr<Observer>& o) {
+    observers.push_back(o);
+}
+
+}} // namespace qpid::broker