Author: gsim
Date: Mon Apr 21 13:34:49 2008
New Revision: 650250
URL: http://svn.apache.org/viewvc?rev=650250&view=rev
Log:
* raise error when controls other than attached are received on unattached
channel
* corrected exception handling in client and on broker (broker to issue detach)
Modified:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.cpp?rev=650250&r1=650249&r2=650250&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.cpp Mon Apr 21
13:34:49 2008
@@ -26,6 +26,7 @@
#include "qpid/framing/constants.h"
#include "qpid/framing/ClientInvoker.h"
#include "qpid/framing/ServerInvoker.h"
+#include "qpid/framing/all_method_bodies.h"
#include "qpid/log/Statement.h"
#include <boost/bind.hpp>
@@ -59,24 +60,32 @@
//
AMQMethodBody* m = f.getBody()->getMethod();
try {
- if (!ignoring) {
- if (m &&
invoke(static_cast<AMQP_ServerOperations::Session010Handler&>(*this), *m)) {
- return;
- } else if (session.get()) {
- session->handle(f);
- } else {
- throw ChannelErrorException(
- QPID_MSG("Channel " << channel.get() << " is not open"));
- }
+ if (m && isValid(m) &&
invoke(static_cast<AMQP_ServerOperations::Session010Handler&>(*this), *m)) {
+ return;
+ } else if (session.get()) {
+ session->handle(f);
+ } else if (!ignoring) {
+ throw ConnectionException(501, QPID_MSG("Channel " <<
channel.get() << " is not attached"));
}
}catch(const ConnectionException& e){
connection.close(e.code, e.what(), classId(m), methodId(m));
+ }catch(const SessionException& e){
+ //execution.exception will have been sent already
+ ignoring = true;
+ //peerSession.requestTimeout(0);
+ session->setTimeout(0);
+ peerSession.detach(name);
+ localSuspend();
}catch(const std::exception& e){
connection.close(
framing::INTERNAL_ERROR, e.what(), classId(m), methodId(m));
}
}
+bool SessionHandler::isValid(AMQMethodBody* m) {
+ return session.get() || m->isA<SessionAttachBody>();
+}
+
void SessionHandler::handleOut(AMQFrame& f) {
channel.handle(f); // Send it.
if (session->sent(f))
@@ -112,12 +121,14 @@
const ConnectionState& SessionHandler::getConnection() const { return
connection; }
//new methods:
-void SessionHandler::attach(const std::string& name, bool /*force*/)
+void SessionHandler::attach(const std::string& _name, bool /*force*/)
{
//TODO: need to revise session manager to support resume as well
assertClosed("attach");
std::auto_ptr<SessionState> state(
connection.broker.getSessionManager().open(*this, 0));
+ name = _name;//TODO: this should be used in conjunction with
+ //userid for connection as sessions identity
session.reset(state.release());
peerSession.attached(name);
peerSession.commandPoint(session->nextOut, 0);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.h?rev=650250&r1=650249&r2=650250&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionHandler.h Mon Apr 21
13:34:49 2008
@@ -100,12 +100,15 @@
void assertActive(const char* method) const;
void assertClosed(const char* method) const;
+ bool isValid(framing::AMQMethodBody*);
+
Connection& connection;
framing::ChannelHandler channel;
framing::AMQP_ClientProxy proxy;
framing::AMQP_ClientProxy::Session010 peerSession;
bool ignoring;
std::auto_ptr<SessionState> session;
+ std::string name;//TODO: this should be part of the session state and
replace the id
};
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp?rev=650250&r1=650249&r2=650250&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionState.cpp Mon Apr 21
13:34:49 2008
@@ -275,11 +275,7 @@
} else {
getProxy().getExecution010().exception(e.code, commandId, 0, 0, 0,
e.what(), FieldTable());
}
- timeout = 0;
- //The python client doesn't currently detach on receiving an exception
- //so the session state isn't destroyed. This is a temporary workaround
- //until that is addressed
- adapter.destroyExclusiveQueues();
+ throw e;
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=650250&r1=650249&r2=650250&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Mon Apr 21
13:34:49 2008
@@ -426,9 +426,12 @@
setState(ATTACHED);
}
-void SessionImpl::detach(const std::string& /*name*/)
+void SessionImpl::detach(const std::string& _name)
{
- throw NotImplementedException("Client does not support detach");
+ Lock l(state);
+ if (name != _name) throw InternalErrorException("Incorrect session name");
+ setState(DETACHED);
+ QPID_LOG(info, "Session detached by peer: " << name);
}
void SessionImpl::detached(const std::string& _name, uint8_t _code)
@@ -561,7 +564,6 @@
//should we wait for the timeout response?
detachedLifetime = 0;
}
- detach();
}
