From f61e1ef7589da893b9b54448224dc0961515eb40 Mon Sep 17 00:00:00 2001
From: Alan Conway <aconway@apache.org>
Date: Fri, 26 Oct 2007 19:48:31 +0000
Subject: Session resume support in client & broker: Client can resume a
 session after voluntary suspend() or network failure. Frames lost in network
 failure are automatically re-transmitted for transparent re-connection.

client::Session improvements:
 - Locking to avoid races between network & user threads.
 - Replaced client::StateManager with sys::StateMonitor - avoid heap allocation.

qpid::Exception clean up:
 - use QPID_MSG consistently to format exception messages.
 - throw typed exceptions (in reply_exceptions.h) for AMQP exceptions.
 - re-throw correct typed exception on client for exceptions from broker.
 - Removed QpidError.h

rubygen/templates/constants.rb:
 - constants.h: Added FOO_CLASS_ID and FOO_BAR_METHOD_ID constants.
 - reply_constants.h: Added throwReplyException(code, text)

log::Logger:
 - Fixed shutdown race in Statement::~Initializer()


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@588761 13f79535-47bb-0310-9956-ffa450edef68
---
 cpp/src/qpid/framing/SessionState.cpp | 120 ++++++++++++++++++++++++++++++++++
 1 file changed, 120 insertions(+)
 create mode 100644 cpp/src/qpid/framing/SessionState.cpp

(limited to 'cpp/src/qpid/framing/SessionState.cpp')

diff --git a/cpp/src/qpid/framing/SessionState.cpp b/cpp/src/qpid/framing/SessionState.cpp
new file mode 100644
index 0000000000..045a0ae115
--- /dev/null
+++ b/cpp/src/qpid/framing/SessionState.cpp
@@ -0,0 +1,120 @@
+/*
+ *
+ * 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 WARRANTIE4bS OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "SessionState.h"
+#include "qpid/framing/reply_exceptions.h"
+#include "qpid/framing/constants.h"
+#include "qpid/framing/AMQMethodBody.h"
+#include "qpid/log/Statement.h"
+
+#include <algorithm>
+
+#include <boost/bind.hpp>
+#include <boost/none.hpp>
+
+namespace qpid {
+namespace framing {
+
+SessionState::SessionState(uint32_t ack,  const Uuid& uuid) :
+    state(ATTACHED),
+    id(uuid),
+    lastReceived(-1),
+    lastSent(-1),
+    ackInterval(ack),
+    sendAckAt(lastReceived+ackInterval),
+    solicitAckAt(lastSent+ackInterval),
+    ackSolicited(false)
+{
+    assert(ackInterval > 0);
+}
+
+namespace {
+bool isSessionCommand(const AMQFrame& f) {
+    return f.getMethod() && f.getMethod()->amqpClassId() == SESSION_CLASS_ID;
+}
+}
+
+boost::optional<SequenceNumber> SessionState::received(const AMQFrame& f) {
+    if (isSessionCommand(f))
+        return boost::none;
+    if (state==RESUMING)
+        throw CommandInvalidException(
+            QPID_MSG("Invalid frame: Resuming session, expected session-ack"));
+    assert(state = ATTACHED);
+    assert(lastReceived<sendAckAt);
+    ++lastReceived;
+    QPID_LOG(trace, "Recv # "<< lastReceived << " " << id);
+    if (lastReceived == sendAckAt)
+        return sendingAck();
+    else
+        return boost::none;
+}
+
+bool SessionState::sent(const AMQFrame& f) {
+    if (isSessionCommand(f))
+        return false;
+    unackedOut.push_back(f);
+    ++lastSent;
+    QPID_LOG(trace, "Sent # "<< lastSent << " " << id);
+    return (state!=RESUMING) &&
+        (lastSent == solicitAckAt) &&
+        sendingSolicit();
+}
+
+SessionState::Replay SessionState::replay() {
+    Replay r(unackedOut.size());
+    std::copy(unackedOut.begin(), unackedOut.end(), r.begin());
+    return r;
+}
+
+void SessionState::receivedAck(SequenceNumber acked) {
+    if (state==RESUMING) state=ATTACHED;
+    assert(state==ATTACHED);
+     if (lastSent < acked)
+        throw InvalidArgumentException("Invalid sequence number in ack");
+    size_t keep = lastSent - acked;
+    if (keep < unackedOut.size()) 
+        unackedOut.erase(unackedOut.begin(), unackedOut.end()-keep);
+    solicitAckAt = std::max(solicitAckAt, SequenceNumber(acked+ackInterval));
+}
+
+SequenceNumber SessionState::sendingAck() {
+    sendAckAt = lastReceived+ackInterval;
+    return lastReceived;
+}
+
+bool SessionState::sendingSolicit() {
+    assert(state == ATTACHED);
+    if (ackSolicited)
+        return false;
+    solicitAckAt = lastSent + ackInterval;
+    return true;
+}
+
+SequenceNumber SessionState::resuming() {
+    state = RESUMING;
+    return sendingAck();
+}
+
+void SessionState::suspend() {
+    state = SUSPENDED;
+}
+
+}} // namespace qpid::framing
-- 
cgit v1.2.1