1 files changed, 275 insertions, 0 deletions
diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
new file mode 100644
index 0000000000..3fb2579e8c
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
@@ -0,0 +1,275 @@
+/*
+ * 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 "SessionHandler.h"
+#include "qpid/SessionState.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/AllInvoker.h"
+#include "qpid/log/Statement.h"
+
+
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace amqp_0_10 {
+using namespace framing;
+using namespace std;
+
+SessionHandler::SessionHandler() : peer(channel), ignoring(), sendReady(), receiveReady() {}
+
+SessionHandler::SessionHandler(FrameHandler& out, ChannelId ch)
+    : channel(ch, &out), peer(channel), ignoring(false) {}
+
+SessionHandler::~SessionHandler() {}
+
+namespace {
+bool isSessionControl(AMQMethodBody* m) {
+    return m &&
+        m->amqpClassId() == SESSION_CLASS_ID;
+}
+bool isSessionDetachedControl(AMQMethodBody* m) {
+    return isSessionControl(m) &&
+        m->amqpMethodId() == SESSION_DETACHED_METHOD_ID;
+}
+} // namespace
+
+void SessionHandler::checkAttached() {
+    if (!getState())
+        throw NotAttachedException(
+            QPID_MSG("Channel " << channel.get() << " is not attached"));
+    assert(getInHandler());
+    assert(channel.next);
+}
+
+void SessionHandler::invoke(const AMQMethodBody& m) {
+    framing::invoke(*this, m);    
+}
+
+void SessionHandler::handleIn(AMQFrame& f) {
+    // Note on channel states: a channel is attached if session != 0
+    AMQMethodBody* m = f.getBody()->getMethod();
+    try {
+        if (ignoring && !isSessionDetachedControl(m))
+            return;
+        else if (isSessionControl(m))
+            invoke(*m);
+        else {
+            checkAttached();
+            if (!receiveReady)
+                throw IllegalStateException(QPID_MSG(getState()->getId() << ": Not ready to receive data"));
+            if (!getState()->receiver.record(f))
+                return; // Ignore duplicates.
+            getInHandler()->handle(f);
+        }
+    }
+    catch(const ChannelException& e){
+        QPID_LOG(error, "Channel exception: " << e.what());
+        if (getState())
+            peer.detached(getState()->getId().getName(), e.code);
+        channelException(e.code, e.getMessage());
+    }
+    catch(const ConnectionException& e) {
+        QPID_LOG(error, "Connection exception: " << e.what());
+        connectionException(e.code, e.getMessage());
+    }
+    catch(const std::exception& e) {
+        QPID_LOG(error, "Unexpected exception: " << e.what());
+        connectionException(connection::FRAMING_ERROR, e.what());
+    }
+}
+
+void SessionHandler::handleOut(AMQFrame& f) {
+    checkAttached();
+    if (!sendReady)
+        throw IllegalStateException(QPID_MSG(getState()->getId() << ": Not ready to send data"));
+    getState()->sender.record(f); 
+    if (getState()->sender.needFlush()) {
+        peer.flush(false, true, true); 
+        getState()->sender.recordFlush();
+    }
+    channel.handle(f);
+}
+
+void SessionHandler::checkName(const std::string& name) {
+    checkAttached();
+    if (name != getState()->getId().getName())
+        throw InvalidArgumentException(
+            QPID_MSG("Incorrect session name: " << name
+                     << ", expecting: " << getState()->getId().getName()));
+}
+
+void SessionHandler::attach(const std::string& name, bool force) {
+    if (getState() && name == getState()->getId().getName())
+        return;                 // Idempotent
+    if (getState())
+        throw SessionBusyException(
+            QPID_MSG("Channel " << channel.get() << " already attached to " << getState()->getId()));
+    setState(name, force);
+    QPID_LOG(debug, "Attached channel " << channel.get() << " to " << getState()->getId());
+    peer.attached(name);
+    if (getState()->hasState())
+        peer.flush(true, true, true);
+    else
+        sendCommandPoint();
+}
+
+void SessionHandler::attached(const std::string& name) {
+    checkName(name);
+}
+
+void SessionHandler::detach(const std::string& name) {
+    checkName(name);
+    peer.detached(name, session::NORMAL);
+    handleDetach();
+}
+
+void SessionHandler::detached(const std::string& name, uint8_t code) {
+    checkName(name);
+    ignoring = false;
+    if (code != session::NORMAL)
+        channelException(code, "session.detached from peer.");
+    else {
+        handleDetach();
+    }
+}
+
+void SessionHandler::handleDetach() {
+    sendReady = receiveReady = false;
+}
+
+void SessionHandler::requestTimeout(uint32_t t) {
+    checkAttached();
+    getState()->setTimeout(t);
+    peer.timeout(t);
+}
+
+void SessionHandler::timeout(uint32_t t) {
+    checkAttached();
+    getState()->setTimeout(t);
+}
+
+void SessionHandler::commandPoint(const SequenceNumber& id, uint64_t offset) {
+    checkAttached();
+    getState()->receiver.setCommandPoint(SessionPoint(id, offset));
+    if (!receiveReady) {
+        receiveReady = true;
+        readyToReceive();
+    }
+}
+
+void SessionHandler::expected(const SequenceSet& commands, const Array& /*fragments*/) {
+    checkAttached();
+    if (commands.empty() && getState()->hasState())
+        throw IllegalStateException(
+            QPID_MSG(getState()->getId() << ": has state but client is attaching as new session."));        
+    getState()->sender.expected(commands.empty() ? SequenceNumber(0) : commands.front());
+    if (!sendReady)  // send command point if not already sent
+        sendCommandPoint();
+}
+
+void SessionHandler::confirmed(const SequenceSet& commands, const Array& /*fragments*/) {
+    checkAttached();
+    // Ignore non-contiguous confirmations.
+    if (!commands.empty() && commands.front() >= getState()->sender.getReplayPoint()) {
+        getState()->sender.confirmed(commands.rangesBegin()->last());
+        }
+}
+
+void SessionHandler::completed(const SequenceSet& commands, bool /*timelyReply*/) {
+    checkAttached();
+    getState()->sender.completed(commands);
+    if (!commands.empty())
+        peer.knownCompleted(commands);    // Always send a timely reply
+}
+
+void SessionHandler::knownCompleted(const SequenceSet& commands) {
+    checkAttached();
+    getState()->receiver.knownCompleted(commands);
+}
+
+void SessionHandler::flush(bool expected, bool confirmed, bool completed) {
+    checkAttached();
+    if (expected)  {
+        SequenceSet expectSet;
+        if (getState()->hasState())
+            expectSet.add(getState()->receiver.getExpected().command);
+        peer.expected(expectSet, Array());
+    }
+    if (confirmed) {
+        SequenceSet confirmSet;
+        if (!getState()->receiver.getUnknownComplete().empty()) 
+            confirmSet.add(getState()->receiver.getUnknownComplete().front(),
+                           getState()->receiver.getReceived().command);
+        peer.confirmed(confirmSet, Array());
+    }
+    if (completed)
+        peer.completed(getState()->receiver.getUnknownComplete(), true);
+}
+
+void SessionHandler::gap(const SequenceSet& /*commands*/) {
+    throw NotImplementedException("session.gap not supported");
+}
+
+void SessionHandler::sendDetach()
+{
+    checkAttached();
+    ignoring = true;
+    peer.detach(getState()->getId().getName());
+}
+
+void SessionHandler::sendCompletion() {
+    checkAttached();
+    peer.completed(getState()->receiver.getUnknownComplete(), true);
+}
+
+void SessionHandler::sendAttach(bool force) {
+    checkAttached();
+    peer.attach(getState()->getId().getName(), force);
+    if (getState()->hasState())
+        peer.flush(true, true, true);
+    else
+        sendCommandPoint();
+}
+
+void SessionHandler::sendCommandPoint()  {
+    SessionPoint point(getState()->sender.getCommandPoint());
+    peer.commandPoint(point.command, point.offset);
+    if (!sendReady) {
+        sendReady = true;
+        readyToSend();
+    }
+}
+
+void SessionHandler::sendTimeout(uint32_t t) {
+    checkAttached();
+    peer.requestTimeout(t);
+}
+
+void SessionHandler::sendFlush() {
+    peer.flush(false, true, true);
+}
+
+bool SessionHandler::ready() const {
+    return  sendReady && receiveReady;
+}
+
+
+}} // namespace qpid::broker