You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ai...@apache.org on 2008/04/22 14:38:37 UTC
svn commit: r650478 [2/3] - in /incubator/qpid/branches/thegreatmerge: ./
qpid/cpp/examples/examples/direct/ qpid/cpp/examples/examples/fanout/
qpid/cpp/examples/examples/pub-sub/
qpid/cpp/examples/examples/request-response/ qpid/cpp/rubygen/ qpid/cpp/...
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionContext.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionContext.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionContext.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionContext.h Tue Apr 22 05:37:29 2008
@@ -27,6 +27,7 @@
#include "qpid/framing/amqp_types.h"
#include "qpid/sys/OutputControl.h"
#include "ConnectionState.h"
+#include "OwnershipToken.h"
#include <boost/noncopyable.hpp>
@@ -34,7 +35,7 @@
namespace qpid {
namespace broker {
-class SessionContext : public sys::OutputControl
+class SessionContext : public OwnershipToken, public sys::OutputControl
{
public:
virtual ~SessionContext(){}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.cpp Tue Apr 22 05:37:29 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,28 +60,30 @@
//
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));
}
}
-void SessionHandler::destroy() {
- ignoring=true; // Ignore trailing frames sent by client.
- session->detach();
- session.reset();
+bool SessionHandler::isValid(AMQMethodBody* m) {
+ return session.get() || m->isA<SessionAttachBody>();
}
void SessionHandler::handleOut(AMQFrame& f) {
@@ -118,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);
@@ -160,12 +165,12 @@
void SessionHandler::requestTimeout(uint32_t t)
{
session->setTimeout(t);
- //proxy.timeout(t);
+ peerSession.timeout(t);
}
-void SessionHandler::timeout(uint32_t)
+void SessionHandler::timeout(uint32_t t)
{
- //not sure what we need to do on the server for this...
+ session->setTimeout(t);
}
void SessionHandler::commandPoint(const framing::SequenceNumber& id, uint64_t offset)
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionHandler.h Tue Apr 22 05:37:29 2008
@@ -70,7 +70,6 @@
void localSuspend();
void detach() { localSuspend(); }
void sendCompletion();
- void destroy();
protected:
void handleIn(framing::AMQFrame&);
@@ -101,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/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionState.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionState.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionState.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/broker/SessionState.cpp Tue Apr 22 05:37:29 2008
@@ -25,6 +25,9 @@
#include "SemanticHandler.h"
#include "SessionManager.h"
#include "SessionHandler.h"
+#include "qpid/framing/AMQContentBody.h"
+#include "qpid/framing/AMQHeaderBody.h"
+#include "qpid/framing/AMQMethodBody.h"
#include "qpid/framing/reply_exceptions.h"
#include "qpid/framing/ServerInvoker.h"
@@ -182,7 +185,7 @@
completed.add(id);
if (!invocation.wasHandled()) {
- throw NotImplementedException("Not implemented");
+ throw NotImplementedException(QPID_MSG("Not implemented: " << *method));
} else if (invocation.hasResult()) {
nextOut++;//execution result is now a command, so the counter must be incremented
getProxy().getExecution010().result(id, invocation.getResult());
@@ -206,6 +209,14 @@
}
msgBuilder.handle(frame);
if (frame.getEof() && frame.getEos()) {//end of frameset
+ if (frame.getBof()) {
+ //i.e this is a just a command frame, add a dummy header
+ AMQFrame header;
+ header.setBody(AMQHeaderBody());
+ header.setBof(false);
+ header.setEof(false);
+ msg->getFrames().append(header);
+ }
msg->setPublisher(&getConnection());
semanticState.handle(msg);
msgBuilder.end();
@@ -242,13 +253,14 @@
SequenceNumber commandId;
try {
//TODO: make command handling more uniform, regardless of whether
- //commands carry content. (For now, assume all single frame
- //assemblies are non-content bearing and all content-bearing
- //assemblies will have more than one frame):
- if (frame.getBof() && frame.getEof()) {
- handleCommand(frame.getMethod(), commandId);
- } else {
+ //commands carry content.
+ AMQMethodBody* m = frame.getMethod();
+ if (m == 0 || m->isContentBearing()) {
handleContent(frame, commandId);
+ } else if (frame.getBof() && frame.getEof()) {
+ handleCommand(frame.getMethod(), commandId);
+ } else {
+ throw InternalErrorException("Cannot handle multi-frame command segments yet");
}
} catch(const SessionException& e) {
//TODO: better implementation of new exception handling mechanism
@@ -263,7 +275,7 @@
} else {
getProxy().getExecution010().exception(e.code, commandId, 0, 0, 0, e.what(), FieldTable());
}
- handler->destroy();
+ throw e;
}
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/AckPolicy.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/AckPolicy.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/AckPolicy.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/AckPolicy.h Tue Apr 22 05:37:29 2008
@@ -21,6 +21,8 @@
*
*/
+#include "qpid/framing/SequenceSet.h"
+
namespace qpid {
namespace client {
@@ -31,21 +33,36 @@
*/
class AckPolicy
{
+ framing::SequenceSet accepted;
size_t interval;
size_t count;
public:
/**
*@param n: acknowledge every n messages.
- *n==0 means no automatick acknowledgement.
+ *n==0 means no automatic acknowledgement.
*/
AckPolicy(size_t n=1) : interval(n), count(n) {}
- void ack(const Message& msg) {
+ void ack(const Message& msg, Session& session) {
+ accepted.add(msg.getId());
if (!interval) return;
- bool send=(--count==0);
- msg.acknowledge(true, send);
- if (send) count = interval;
+ if (--count==0) {
+ session.markCompleted(msg.getId(), false, true);
+ session.messageAccept(accepted);
+ accepted.clear();
+ count = interval;
+ } else {
+ session.markCompleted(msg.getId(), false, false);
+ }
+ }
+
+ void ackOutstanding(Session& session) {
+ if (!accepted.empty()) {
+ session.messageAccept(accepted);
+ accepted.clear();
+ session.sendCompletion();
+ }
}
};
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.cpp Tue Apr 22 05:37:29 2008
@@ -23,10 +23,10 @@
#include <sstream>
#include "Channel.h"
#include "qpid/sys/Monitor.h"
+#include "AckPolicy.h"
#include "Message.h"
#include "Connection.h"
#include "Demux.h"
-#include "FutureResponse.h"
#include "MessageListener.h"
#include "MessageQueue.h"
#include <boost/format.hpp>
@@ -47,9 +47,17 @@
class ScopedSync
{
Session& session;
+ const bool change;
+ const bool value;
public:
- ScopedSync(Session& s, bool enabled = true) : session(s) { session.setSynchronous(enabled); }
- ~ScopedSync() { session.setSynchronous(false); }
+ ScopedSync(Session& s, bool desired = true) : session(s), change(s.isSynchronous() != desired), value(desired)
+ {
+ if (change) session.setSynchronous(value);
+ }
+ ~ScopedSync()
+ {
+ if (change) session.setSynchronous(!value);
+ }
};
Channel::Channel(bool _transactional, u_int16_t _prefetch) :
@@ -116,7 +124,7 @@
string e = exchange.getName();
string q = queue.getName();
ScopedSync s(session, synch);
- session.queueBind(0, q, e, key, args);
+ session.exchangeBind(q, e, key, args);
}
void Channel::commit(){
@@ -129,7 +137,7 @@
void Channel::consume(
Queue& _queue, const std::string& tag, MessageListener* listener,
- AckMode ackMode, bool noLocal, bool synch, const FieldTable* fields) {
+ AckMode ackMode, bool noLocal, bool synch, FieldTable* fields) {
if (tag.empty()) {
throw Exception("A tag must be specified for a consumer.");
@@ -144,13 +152,18 @@
c.ackMode = ackMode;
c.count = 0;
}
- uint8_t confirmMode = ackMode == NO_ACK ? 0 : 1;
+ uint8_t confirmMode = ackMode == NO_ACK ? 1 : 0;
ScopedSync s(session, synch);
- session.messageSubscribe(0, _queue.getName(), tag, noLocal,
+ FieldTable ft;
+ FieldTable* ftptr = fields ? fields : &ft;
+ if (noLocal) {
+ ftptr->setString("qpid.no-local","yes");
+ }
+ session.messageSubscribe(_queue.getName(), tag,
confirmMode, 0/*pre-acquire*/,
- false, fields ? *fields : FieldTable());
+ false, "", 0, *ftptr);
if (!prefetch) {
- session.messageFlowMode(tag, 0/*credit based*/);
+ session.messageSetFlowMode(tag, 0/*credit based*/);
}
//allocate some credit:
@@ -177,17 +190,24 @@
ScopedDivert handler(tag, session.getExecution().getDemux());
Demux::QueuePtr incoming = handler.getQueue();
- session.messageSubscribe(destination=tag, queue=_queue.getName(), confirmMode=(ackMode == NO_ACK ? 0 : 1));
+ session.messageSubscribe(destination=tag, queue=_queue.getName(), acceptMode=(ackMode == NO_ACK ? 1 : 0));
session.messageFlow(tag, 1/*BYTES*/, 0xFFFFFFFF);
session.messageFlow(tag, 0/*MESSAGES*/, 1);
- Completion status = session.messageFlush(tag);
- status.sync();
+ {
+ ScopedSync s(session);
+ session.messageFlush(tag);
+ }
session.messageCancel(tag);
FrameSet::shared_ptr p;
if (incoming->tryPop(p)) {
msg.populate(*p);
- if (ackMode == AUTO_ACK) msg.acknowledge(session, false, true);
+ if (ackMode == AUTO_ACK) {
+ AckPolicy acker;
+ acker.ack(msg, session);
+ } else {
+ session.markCompleted(msg.getId(), false, false);
+ }
return true;
}
else
@@ -243,7 +263,7 @@
bool send = i->second.ackMode == AUTO_ACK
|| (prefetch && ++(i->second.count) > (prefetch / 2));
if (send) i->second.count = 0;
- session.getExecution().completed(content.getId(), true, send);
+ session.markCompleted(content.getId(), true, send);
}
} else {
QPID_LOG(warning, "Dropping message for unrecognised consumer: " << destination);
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Channel.h Tue Apr 22 05:37:29 2008
@@ -256,7 +256,7 @@
void consume(
Queue& queue, const std::string& tag, MessageListener* listener,
AckMode ackMode = NO_ACK, bool noLocal = false, bool synch = true,
- const framing::FieldTable* fields = 0);
+ framing::FieldTable* fields = 0);
/**
* Cancels a subscription previously set up through a call to consume().
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Completion.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Completion.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Completion.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Completion.h Tue Apr 22 05:37:29 2008
@@ -24,7 +24,7 @@
#include <boost/shared_ptr.hpp>
#include "Future.h"
-#include "SessionCore.h"
+#include "SessionImpl.h"
namespace qpid {
namespace client {
@@ -33,17 +33,12 @@
{
protected:
Future future;
- shared_ptr<SessionCore> session;
+ shared_ptr<SessionImpl> session;
public:
Completion() {}
- Completion(Future f, shared_ptr<SessionCore> s) : future(f), session(s) {}
-
- void sync()
- {
- future.sync(*session);
- }
+ Completion(Future f, shared_ptr<SessionImpl> s) : future(f), session(s) {}
void wait()
{
@@ -52,10 +47,6 @@
bool isComplete() {
return future.isComplete(*session);
- }
-
- bool isCompleteUpTo() {
- return future.isCompleteUpTo(*session);
}
};
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.cpp Tue Apr 22 05:37:29 2008
@@ -25,7 +25,7 @@
#include "Connection.h"
#include "Channel.h"
#include "Message.h"
-#include "SessionCore.h"
+#include "SessionImpl.h"
#include "qpid/log/Logger.h"
#include "qpid/log/Options.h"
#include "qpid/log/Statement.h"
@@ -76,8 +76,8 @@
Session Connection::newSession(SynchronousMode sync,
uint32_t detachedLifetime)
{
- shared_ptr<SessionCore> core(
- new SessionCore(impl, ++channelIdCounter, max_frame_size));
+ shared_ptr<SessionImpl> core(
+ new SessionImpl(impl, ++channelIdCounter, max_frame_size));
core->setSync(sync);
impl->addSession(core);
core->open(detachedLifetime);
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connection.h Tue Apr 22 05:37:29 2008
@@ -137,7 +137,7 @@
Session newSession(SynchronousMode sync, uint32_t detachedLifetime=0);
/**
- * Resume a suspendded session. A session may be resumed
+ * Resume a suspended session. A session may be resumed
* on a different connection to the one that created it.
*/
void resume(Session& session);
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.cpp Tue Apr 22 05:37:29 2008
@@ -24,6 +24,7 @@
#include "qpid/framing/amqp_framing.h"
#include "qpid/framing/AMQP_HighestVersion.h"
#include "qpid/framing/all_method_bodies.h"
+#include "qpid/framing/ClientInvoker.h"
using namespace qpid::client;
using namespace qpid::framing;
@@ -31,14 +32,21 @@
namespace {
const std::string OK("OK");
+const std::string PLAIN("PLAIN");
+const std::string en_US("en_US");
+
+const std::string INVALID_STATE_START("start received in invalid state");
+const std::string INVALID_STATE_TUNE("tune received in invalid state");
+const std::string INVALID_STATE_OPEN_OK("open-ok received in invalid state");
+const std::string INVALID_STATE_CLOSE_OK("close-ok received in invalid state");
}
ConnectionHandler::ConnectionHandler()
- : StateManager(NOT_STARTED)
+ : StateManager(NOT_STARTED), outHandler(*this), proxy(outHandler), errorCode(200)
{
- mechanism = "PLAIN";
- locale = "en_US";
+ mechanism = PLAIN;
+ locale = en_US;
heartbeat = 0;
maxChannels = 32767;
maxFrameSize = 65535;
@@ -46,40 +54,36 @@
version = framing::highestProtocolVersion;
ESTABLISHED.insert(FAILED);
+ ESTABLISHED.insert(CLOSED);
ESTABLISHED.insert(OPEN);
}
void ConnectionHandler::incoming(AMQFrame& frame)
{
if (getState() == CLOSED) {
- throw Exception("Connection is closed.");
+ throw Exception("Received frame on closed connection");
}
+
AMQBody* body = frame.getBody();
- if (frame.getChannel() == 0) {
- if (body->getMethod()) {
- handle(body->getMethod());
- } else {
- error(503, "Cannot send content on channel zero.");
- }
- } else {
- switch(getState()) {
- case OPEN:
- try {
+ try {
+ if (frame.getChannel() != 0 || !invoke(static_cast<ConnectionOperations&>(*this), *body)) {
+ switch(getState()) {
+ case OPEN:
in(frame);
- }catch(ConnectionException& e){
- error(e.code, e.what(), body);
- }catch(std::exception& e){
- error(541/*internal error*/, e.what(), body);
+ break;
+ case CLOSING:
+ QPID_LOG(warning, "Ignoring frame while closing connection: " << frame);
+ break;
+ default:
+ throw Exception("Cannot receive frames on non-zero channel until connection is established.");
}
- break;
- case CLOSING:
- QPID_LOG(warning, "Received frame on non-zero channel while closing connection; frame ignored.");
- break;
- default:
- //must be in connection initialisation:
- fail("Cannot receive frames on non-zero channel until connection is established.");
}
+ }catch(std::exception& e){
+ QPID_LOG(warning, "Closing connection due to " << e.what());
+ setState(CLOSING);
+ proxy.close(501, e.what());
+ if (onError) onError(501, e.what());
}
}
@@ -95,8 +99,8 @@
void ConnectionHandler::waitForOpen()
{
waitFor(ESTABLISHED);
- if (getState() == FAILED) {
- throw Exception("Failed to establish connection.");
+ if (getState() == FAILED || getState() == CLOSED) {
+ throw ConnectionException(errorCode, errorText);
}
}
@@ -105,105 +109,85 @@
switch (getState()) {
case NEGOTIATING:
case OPENING:
- setState(FAILED);
+ fail("Connection closed before it was established");
break;
case OPEN:
setState(CLOSING);
- send(ConnectionCloseBody(version, 200, OK, 0, 0));
+ proxy.close(200, OK);
waitFor(CLOSED);
break;
// Nothing to do for CLOSING, CLOSED, FAILED or NOT_STARTED
}
}
-void ConnectionHandler::send(const framing::AMQBody& body)
+void ConnectionHandler::checkState(STATES s, const std::string& msg)
{
- AMQFrame f(body);
- out(f);
+ if (getState() != s) {
+ throw CommandInvalidException(msg);
+ }
}
-void ConnectionHandler::error(uint16_t code, const std::string& message, uint16_t classId, uint16_t methodId)
+void ConnectionHandler::fail(const std::string& message)
{
- setState(CLOSING);
- send(ConnectionCloseBody(version, code, message, classId, methodId));
+ errorCode = 502;
+ errorText = message;
+ QPID_LOG(warning, message);
+ setState(FAILED);
}
-void ConnectionHandler::error(uint16_t code, const std::string& message, AMQBody* body)
+void ConnectionHandler::start(const FieldTable& /*serverProps*/, const Array& /*mechanisms*/, const Array& /*locales*/)
{
- if (onError)
- onError(code, message);
- AMQMethodBody* method = body->getMethod();
- if (method)
- error(code, message, method->amqpClassId(), method->amqpMethodId());
- else
- error(code, message);
+ checkState(NOT_STARTED, INVALID_STATE_START);
+ setState(NEGOTIATING);
+ //TODO: verify that desired mechanism and locale are supported
+ string response = ((char)0) + uid + ((char)0) + pwd;
+ proxy.startOk(properties, mechanism, response, locale);
}
+void ConnectionHandler::secure(const std::string& /*challenge*/)
+{
+ throw NotImplementedException("Challenge-response cycle not yet implemented in client");
+}
-void ConnectionHandler::fail(const std::string& message)
+void ConnectionHandler::tune(uint16_t channelMax, uint16_t /*frameMax*/, uint16_t /*heartbeatMin*/, uint16_t /*heartbeatMax*/)
{
- QPID_LOG(warning, message);
- setState(FAILED);
+ checkState(NEGOTIATING, INVALID_STATE_TUNE);
+ //TODO: verify that desired heartbeat and max frame size are valid
+ maxChannels = channelMax;
+ proxy.tuneOk(maxChannels, maxFrameSize, heartbeat);
+ setState(OPENING);
+ proxy.open(vhost, capabilities, insist);
}
-void ConnectionHandler::handle(AMQMethodBody* method)
+void ConnectionHandler::openOk(const framing::Array& /*knownHosts*/)
{
- switch (getState()) {
- case NOT_STARTED:
- if (method->isA<ConnectionStartBody>()) {
- setState(NEGOTIATING);
- string response = ((char)0) + uid + ((char)0) + pwd;
- send(ConnectionStartOkBody(version, properties, mechanism, response, locale));
- } else {
- fail("Bad method sequence, expected connection-start.");
- }
- break;
- case NEGOTIATING:
- if (method->isA<ConnectionTuneBody>()) {
- ConnectionTuneBody* proposal=polymorphic_downcast<ConnectionTuneBody*>(method);
- heartbeat = proposal->getHeartbeat();
- maxChannels = proposal->getChannelMax();
- send(ConnectionTuneOkBody(version, maxChannels, maxFrameSize, heartbeat));
- setState(OPENING);
- send(ConnectionOpenBody(version, vhost, capabilities, insist));
- //TODO: support for further security challenges
- //} else if (method->isA<ConnectionSecureBody>()) {
- } else {
- fail("Unexpected method sequence, expected connection-tune.");
- }
- break;
- case OPENING:
- if (method->isA<ConnectionOpenOkBody>()) {
- setState(OPEN);
- //TODO: support for redirection
- //} else if (method->isA<ConnectionRedirectBody>()) {
- } else {
- fail("Unexpected method sequence, expected connection-open-ok.");
- }
- break;
- case OPEN:
- if (method->isA<ConnectionCloseBody>()) {
- send(ConnectionCloseOkBody(version));
- setState(CLOSED);
- ConnectionCloseBody* c=polymorphic_downcast<ConnectionCloseBody*>(method);
- QPID_LOG(warning, "Broker closed connection: " << c->getReplyCode()
- << ", " << c->getReplyText());
- if (onError) {
- onError(c->getReplyCode(), c->getReplyText());
- }
- } else {
- error(503, "Unexpected method on channel zero.", method->amqpClassId(), method->amqpMethodId());
- }
- break;
- case CLOSING:
- if (method->isA<ConnectionCloseOkBody>()) {
- if (onClose) {
- onClose();
- }
- setState(CLOSED);
- } else {
- QPID_LOG(warning, "Received frame on channel zero while closing connection; frame ignored.");
- }
- break;
+ checkState(OPENING, INVALID_STATE_OPEN_OK);
+ //TODO: store knownHosts for reconnection etc
+ setState(OPEN);
+}
+
+void ConnectionHandler::redirect(const std::string& /*host*/, const Array& /*knownHosts*/)
+{
+ throw NotImplementedException("Redirection received from broker; not yet implemented in client");
+}
+
+void ConnectionHandler::close(uint16_t replyCode, const std::string& replyText)
+{
+ proxy.closeOk();
+ setState(CLOSED);
+ errorCode = replyCode;
+ errorText = replyText;
+ QPID_LOG(warning, "Broker closed connection: " << replyCode << ", " << replyText);
+ if (onError) {
+ onError(replyCode, replyText);
+ }
+}
+
+void ConnectionHandler::closeOk()
+{
+ checkState(CLOSING, INVALID_STATE_CLOSE_OK);
+ if (onClose) {
+ onClose();
}
+ setState(CLOSED);
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionHandler.h Tue Apr 22 05:37:29 2008
@@ -21,12 +21,16 @@
#ifndef _ConnectionHandler_
#define _ConnectionHandler_
+#include "ChainableFrameHandler.h"
#include "Connector.h"
#include "StateManager.h"
-#include "ChainableFrameHandler.h"
-#include "qpid/framing/InputHandler.h"
-#include "qpid/framing/FieldTable.h"
#include "qpid/framing/AMQMethodBody.h"
+#include "qpid/framing/AMQP_ClientOperations.h"
+#include "qpid/framing/AMQP_ServerProxy.h"
+#include "qpid/framing/Array.h"
+#include "qpid/framing/FieldTable.h"
+#include "qpid/framing/FrameHandler.h"
+#include "qpid/framing/InputHandler.h"
namespace qpid {
namespace client {
@@ -39,7 +43,7 @@
framing::FieldTable properties;
std::string mechanism;
std::string locale;
- std::string capabilities;
+ framing::Array capabilities;
uint16_t heartbeat;
uint16_t maxChannels;
uint64_t maxFrameSize;
@@ -48,17 +52,44 @@
};
class ConnectionHandler : private StateManager,
- public ConnectionProperties,
- public ChainableFrameHandler,
- public framing::InputHandler
+ public ConnectionProperties,
+ public ChainableFrameHandler,
+ public framing::InputHandler,
+ private framing::AMQP_ClientOperations::Connection010Handler
{
+ typedef framing::AMQP_ClientOperations::Connection010Handler ConnectionOperations;
enum STATES {NOT_STARTED, NEGOTIATING, OPENING, OPEN, CLOSING, CLOSED, FAILED};
std::set<int> ESTABLISHED;
- void handle(framing::AMQMethodBody* method);
- void send(const framing::AMQBody& body);
- void error(uint16_t code, const std::string& message, uint16_t classId = 0, uint16_t methodId = 0);
- void error(uint16_t code, const std::string& message, framing::AMQBody* body);
+ class Adapter : public framing::FrameHandler
+ {
+ ConnectionHandler& handler;
+ public:
+ Adapter(ConnectionHandler& h) : handler(h) {}
+ void handle(framing::AMQFrame& f) { handler.out(f); }
+ };
+
+ Adapter outHandler;
+ framing::AMQP_ServerProxy::Connection010 proxy;
+ uint16_t errorCode;
+ std::string errorText;
+
+ void checkState(STATES s, const std::string& msg);
+
+ //methods corresponding to connection controls:
+ void start(const framing::FieldTable& serverProperties,
+ const framing::Array& mechanisms,
+ const framing::Array& locales);
+ void secure(const std::string& challenge);
+ void tune(uint16_t channelMax,
+ uint16_t frameMax,
+ uint16_t heartbeatMin,
+ uint16_t heartbeatMax);
+ void openOk(const framing::Array& knownHosts);
+ void redirect(const std::string& host,
+ const framing::Array& knownHosts);
+ void close(uint16_t replyCode, const std::string& replyText);
+ void closeOk();
public:
using InputHandler::handle;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.cpp Tue Apr 22 05:37:29 2008
@@ -23,7 +23,7 @@
#include "qpid/framing/reply_exceptions.h"
#include "ConnectionImpl.h"
-#include "SessionCore.h"
+#include "SessionImpl.h"
#include <boost/bind.hpp>
#include <boost/format.hpp>
@@ -32,6 +32,7 @@
using namespace qpid::framing;
using namespace qpid::sys;
+
ConnectionImpl::ConnectionImpl(boost::shared_ptr<Connector> c)
: connector(c), isClosed(false), isClosing(false)
{
@@ -52,10 +53,10 @@
connector->close();
}
-void ConnectionImpl::addSession(const boost::shared_ptr<SessionCore>& session)
+void ConnectionImpl::addSession(const boost::shared_ptr<SessionImpl>& session)
{
Mutex::ScopedLock l(lock);
- boost::weak_ptr<SessionCore>& s = sessions[session->getChannel()];
+ boost::weak_ptr<SessionImpl>& s = sessions[session->getChannel()];
if (s.lock()) throw ChannelBusyException();
s = session;
}
@@ -67,7 +68,7 @@
void ConnectionImpl::incoming(framing::AMQFrame& frame)
{
- boost::shared_ptr<SessionCore> s;
+ boost::shared_ptr<SessionImpl> s;
{
Mutex::ScopedLock l(lock);
s = sessions[frame.getChannel()].lock();
@@ -122,7 +123,7 @@
connector->close();
SessionVector save;
for (SessionMap::iterator i = sessions.begin(); i != sessions.end(); ++i) {
- boost::shared_ptr<SessionCore> s = i->second.lock();
+ boost::shared_ptr<SessionImpl> s = i->second.lock();
if (s) save.push_back(s);
}
sessions.clear();
@@ -135,7 +136,7 @@
if (isClosed) return;
SessionVector save(closeInternal(l));
Mutex::ScopedUnlock u(lock);
- std::for_each(save.begin(), save.end(), boost::bind(&SessionCore::connectionClosed, _1, code, text));
+ std::for_each(save.begin(), save.end(), boost::bind(&SessionImpl::connectionClosed, _1, code, text));
}
static const std::string CONN_CLOSED("Connection closed by broker");
@@ -148,7 +149,7 @@
handler.fail(CONN_CLOSED);
Mutex::ScopedUnlock u(lock);
std::for_each(save.begin(), save.end(),
- boost::bind(&SessionCore::connectionBroke, _1, INTERNAL_ERROR, CONN_CLOSED));
+ boost::bind(&SessionImpl::connectionBroke, _1, INTERNAL_ERROR, CONN_CLOSED));
}
void ConnectionImpl::erase(uint16_t ch) {
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/ConnectionImpl.h Tue Apr 22 05:37:29 2008
@@ -35,15 +35,15 @@
namespace qpid {
namespace client {
-class SessionCore;
+class SessionImpl;
class ConnectionImpl : public framing::FrameHandler,
public sys::TimeoutHandler,
public sys::ShutdownHandler
{
- typedef std::map<uint16_t, boost::weak_ptr<SessionCore> > SessionMap;
- typedef std::vector<boost::shared_ptr<SessionCore> > SessionVector;
+ typedef std::map<uint16_t, boost::weak_ptr<SessionImpl> > SessionMap;
+ typedef std::vector<boost::shared_ptr<SessionImpl> > SessionVector;
SessionMap sessions;
ConnectionHandler handler;
@@ -69,7 +69,7 @@
ConnectionImpl(boost::shared_ptr<Connector> c);
~ConnectionImpl();
- void addSession(const boost::shared_ptr<SessionCore>&);
+ void addSession(const boost::shared_ptr<SessionImpl>&);
void open(const std::string& host, int port = 5672,
const std::string& uid = "guest",
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.cpp Tue Apr 22 05:37:29 2008
@@ -46,6 +46,7 @@
receive_buffer_size(buffer_size),
send_buffer_size(buffer_size),
version(ver),
+ initiated(false),
closed(true),
joined(true),
timeout(0),
@@ -240,6 +241,14 @@
void Connector::readbuff(AsynchIO& aio, AsynchIO::BufferBase* buff) {
framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);
+ if (!initiated) {
+ framing::ProtocolInitiation protocolInit;
+ if (protocolInit.decode(in)) {
+ //TODO: check the version is correct
+ QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")");
+ }
+ initiated = true;
+ }
AMQFrame frame;
while(frame.decode(in)){
QPID_LOG(trace, "RECV " << identifier << ": " << frame);
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Connector.h Tue Apr 22 05:37:29 2008
@@ -77,8 +77,9 @@
const int receive_buffer_size;
const int send_buffer_size;
framing::ProtocolVersion version;
+ bool initiated;
- sys::Mutex closedLock;
+ sys::Mutex closedLock;
bool closed;
bool joined;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Dispatcher.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Dispatcher.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Dispatcher.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Dispatcher.cpp Tue Apr 22 05:37:29 2008
@@ -43,7 +43,7 @@
{
if (listener) {
listener->received(msg);
- autoAck.ack(msg);
+ autoAck.ack(msg, session);
}
}
@@ -72,7 +72,7 @@
Mutex::ScopedUnlock u(lock);
FrameSet::shared_ptr content = queue->pop();
if (content->isA<MessageTransferBody>()) {
- Message msg(*content, session);
+ Message msg(*content);
Subscriber::shared_ptr listener = find(msg.getDestination());
if (!listener) {
QPID_LOG(error, "No listener found for destination " << msg.getDestination());
@@ -84,7 +84,7 @@
if (handler.get()) {
handler->handle(*content);
} else {
- QPID_LOG(error, "No handler found for " << *(content->getMethod()));
+ QPID_LOG(warning, "No handler found for " << *(content->getMethod()));
}
}
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Execution.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Execution.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Execution.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Execution.h Tue Apr 22 05:37:29 2008
@@ -28,21 +28,23 @@
namespace client {
/**
- * Provides more detailed access to the amqp 'execution layer'.
+ * Provides access to more detailed aspects of the session
+ * implementation.
*/
class Execution
{
public:
virtual ~Execution() {}
- virtual void sendSyncRequest() = 0;
- virtual void sendFlushRequest() = 0;
- virtual void completed(const framing::SequenceNumber& id, bool cumulative, bool send) = 0;
+ /**
+ * Provides access to the demultiplexing function within the
+ * session implementation
+ */
virtual Demux& getDemux() = 0;
- virtual bool isComplete(const framing::SequenceNumber& id) = 0;
- virtual bool isCompleteUpTo(const framing::SequenceNumber& id) = 0;
- virtual void setCompletionListener(boost::function<void()>) = 0;
- virtual void syncWait(const framing::SequenceNumber& id) = 0;
- virtual framing::SequenceNumber lastSent() const = 0;
+ /**
+ * Wait until notification has been received of completion of the
+ * outgoing command with the specified id.
+ */
+ void waitForCompletion(const framing::SequenceNumber& id);
};
}}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Future.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Future.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Future.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Future.h Tue Apr 22 05:37:29 2008
@@ -28,9 +28,8 @@
#include "qpid/framing/SequenceNumber.h"
#include "qpid/framing/StructHelper.h"
#include "FutureCompletion.h"
-#include "FutureResponse.h"
#include "FutureResult.h"
-#include "SessionCore.h"
+#include "SessionImpl.h"
namespace qpid {
namespace client {
@@ -38,7 +37,6 @@
class Future : private framing::StructHelper
{
framing::SequenceNumber command;
- boost::shared_ptr<FutureResponse> response;
boost::shared_ptr<FutureResult> result;
bool complete;
@@ -46,42 +44,7 @@
Future() : complete(false) {}
Future(const framing::SequenceNumber& id) : command(id), complete(false) {}
- void sync(SessionCore& session)
- {
- if (!isComplete(session)) {
- session.getExecution().syncTo(command);
- wait(session);
- }
- }
-
- void wait(SessionCore& session)
- {
- if (!isComplete(session)) {
- FutureCompletion callback;
- session.getExecution().getCompletionTracker().listenForCompletion(
- command,
- boost::bind(&FutureCompletion::completed, &callback)
- );
- callback.waitForCompletion();
- session.assertOpen();
- complete = true;
- }
- }
-
- framing::AMQMethodBody* getResponse(SessionCore& session)
- {
- if (response) {
- session.getExecution().getCompletionTracker().listenForCompletion(
- command,
- boost::bind(&FutureResponse::completed, response)
- );
- return response->getResponse(session);
- } else {
- throw Exception("Response not expected");
- }
- }
-
- template <class T> void decodeResult(T& value, SessionCore& session)
+ template <class T> void decodeResult(T& value, SessionImpl& session)
{
if (result) {
decode(value, result->getResult(session));
@@ -90,17 +53,9 @@
}
}
- bool isComplete(SessionCore& session) {
- return complete || session.getExecution().isComplete(command);
- }
-
- bool isCompleteUpTo(SessionCore& session) {
- return complete || session.getExecution().isCompleteUpTo(command);
- }
-
- void setCommandId(const framing::SequenceNumber& id) { command = id; }
- void setFutureResponse(boost::shared_ptr<FutureResponse> r) { response = r; }
- void setFutureResult(boost::shared_ptr<FutureResult> r) { result = r; }
+ void wait(SessionImpl& session);
+ bool isComplete(SessionImpl& session);
+ void setFutureResult(boost::shared_ptr<FutureResult> r);
};
}}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.cpp Tue Apr 22 05:37:29 2008
@@ -21,13 +21,13 @@
#include "FutureResult.h"
-#include "SessionCore.h"
+#include "SessionImpl.h"
using namespace qpid::client;
using namespace qpid::framing;
using namespace qpid::sys;
-const std::string& FutureResult::getResult(SessionCore& session) const
+const std::string& FutureResult::getResult(SessionImpl& session) const
{
waitForCompletion();
session.assertOpen();
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/FutureResult.h Tue Apr 22 05:37:29 2008
@@ -29,13 +29,13 @@
namespace qpid {
namespace client {
-class SessionCore;
+class SessionImpl;
class FutureResult : public FutureCompletion
{
std::string result;
public:
- const std::string& getResult(SessionCore& session) const;
+ const std::string& getResult(SessionImpl& session) const;
void received(const std::string& result);
};
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.cpp Tue Apr 22 05:37:29 2008
@@ -36,8 +36,8 @@
throw ClosedException();
FrameSet::shared_ptr content = queue->pop();
if (content->isA<MessageTransferBody>()) {
- Message m(*content, session);
- autoAck.ack(m);
+ Message m(*content);
+ autoAck.ack(m, session);
return m;
}
else
@@ -46,6 +46,7 @@
}
void LocalQueue::setAckPolicy(AckPolicy a) { autoAck=a; }
+AckPolicy& LocalQueue::getAckPolicy() { return autoAck; }
bool LocalQueue::empty() const
{
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/LocalQueue.h Tue Apr 22 05:37:29 2008
@@ -47,9 +47,10 @@
bool empty() const;
size_t size() const;
void setAckPolicy(AckPolicy);
+ AckPolicy& getAckPolicy();
private:
- friend class SubscriptionManager;
+ friend class SubscriptionManager;
Session session;
Demux::QueuePtr queue;
AckPolicy autoAck;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Message.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Message.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Message.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/Message.h Tue Apr 22 05:37:29 2008
@@ -30,7 +30,7 @@
namespace client {
/**
- * A representation of messages for sent or recived through the
+ * A representation of messages for sent or received through the
* client api.
*
* \ingroup clientapi
@@ -38,65 +38,22 @@
class Message : public framing::TransferContent
{
public:
- Message(const std::string& data_=std::string(),
+ Message(const std::string& data=std::string(),
const std::string& routingKey=std::string(),
- const std::string& exchange=std::string()
- ) : TransferContent(data_, routingKey, exchange) {}
-
- std::string getDestination() const
- {
- return method.getDestination();
- }
-
- bool isRedelivered() const
- {
- return hasDeliveryProperties() && getDeliveryProperties().getRedelivered();
- }
-
- void setRedelivered(bool redelivered)
- {
- getDeliveryProperties().setRedelivered(redelivered);
- }
-
- framing::FieldTable& getHeaders()
- {
- return getMessageProperties().getApplicationHeaders();
- }
-
- void acknowledge(Session& session, bool cumulative = true, bool send = true) const
- {
- session.getExecution().completed(id, cumulative, send);
- }
-
- void acknowledge(bool cumulative = true, bool send = true) const
- {
- const_cast<Session&>(session).getExecution().completed(id, cumulative, send);
- }
+ const std::string& exchange=std::string());
+ std::string getDestination() const;
+ bool isRedelivered() const;
+ void setRedelivered(bool redelivered);
+ framing::FieldTable& getHeaders();
+ const framing::MessageTransferBody& getMethod() const;
+ const framing::SequenceNumber& getId() const;
/**@internal for incoming messages */
- Message(const framing::FrameSet& frameset, Session s) :
- method(*frameset.as<framing::MessageTransferBody>()), id(frameset.getId()), session(s)
- {
- populate(frameset);
- }
-
- const framing::MessageTransferBody& getMethod() const
- {
- return method;
- }
-
- const framing::SequenceNumber& getId() const
- {
- return id;
- }
-
- /**@internal use for incoming messages. */
- void setSession(Session s) { session=s; }
+ Message(const framing::FrameSet& frameset);
private:
//method and id are only set for received messages:
framing::MessageTransferBody method;
framing::SequenceNumber id;
- Session session;
};
}}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.cpp Tue Apr 22 05:37:29 2008
@@ -19,6 +19,7 @@
*
*/
#include "SessionBase.h"
+#include "qpid/framing/all_method_bodies.h"
namespace qpid {
namespace client {
@@ -26,7 +27,7 @@
SessionBase::SessionBase() {}
SessionBase::~SessionBase() {}
-SessionBase::SessionBase(shared_ptr<SessionCore> core) : impl(core) {}
+SessionBase::SessionBase(shared_ptr<SessionImpl> core) : impl(core) {}
void SessionBase::suspend() { impl->suspend(); }
void SessionBase::close() { impl->close(); }
@@ -37,14 +38,40 @@
return SynchronousMode(impl->isSync());
}
-Execution& SessionBase::getExecution() { return impl->getExecution(); }
+Execution& SessionBase::getExecution()
+{
+ return *impl;
+}
+
+void SessionBase::sync()
+{
+ ExecutionSyncBody b;
+ b.setSync(true);
+ impl->send(b).wait(*impl);
+}
+
+void SessionBase::markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer)
+{
+ impl->markCompleted(id, cumulative, notifyPeer);
+}
+
+void SessionBase::sendCompletion()
+{
+ impl->sendCompletion();
+}
+
Uuid SessionBase::getId() const { return impl->getId(); }
framing::FrameSet::shared_ptr SessionBase::get() { return impl->get(); }
-void SessionBase::sync() {
- Execution& ex = getExecution();
- ex.syncWait(ex.lastSent());
- impl->assertOpen();
+SessionBase::ScopedSync::ScopedSync(SessionBase& s) : session(s), change(!s.isSynchronous())
+{
+ if (change) session.setSynchronous(true);
}
+
+SessionBase::ScopedSync::~ScopedSync()
+{
+ if (change) session.setSynchronous(false);
+}
+
}} // namespace qpid::client
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SessionBase.h Tue Apr 22 05:37:29 2008
@@ -29,8 +29,8 @@
#include "qpid/framing/TransferContent.h"
#include "qpid/client/Completion.h"
#include "qpid/client/ConnectionImpl.h"
-#include "qpid/client/Response.h"
-#include "qpid/client/SessionCore.h"
+#include "qpid/client/Execution.h"
+#include "qpid/client/SessionImpl.h"
#include "qpid/client/TypedResult.h"
#include "qpid/shared_ptr.h"
#include <string>
@@ -61,8 +61,11 @@
* <em>later</em> function call.
*
* If you need to notify some extenal agent that some actions have
- * been taken (e.g. binding queues to exchanages), you must call
- * Session::sync() first, to ensure that all the commands are complete.
+ * been taken (e.g. binding queues to exchanges), you must ensure that
+ * the broker has completed the command. In synchronous mode this is
+ * when the session method for the command returns. In asynchronous
+ * mode you can call Session::sync(), to ensure that all the commands
+ * are complete.
*
* You can freely switch between modes by calling Session::setSynchronous()
*
@@ -77,6 +80,21 @@
class SessionBase
{
public:
+ /**
+ * Instances of this class turn synchronous mode on for the
+ * duration of their scope (and revert back to async if required
+ * afterwards).
+ */
+ class ScopedSync
+ {
+ SessionBase& session;
+ const bool change;
+ public:
+ ScopedSync(SessionBase& s);
+ ~ScopedSync();
+ };
+
+
SessionBase();
~SessionBase();
@@ -110,23 +128,19 @@
/** Close the session */
void close();
-
- /**
- * Synchronize with the broker. Wait for all commands issued so far in
- * the session to complete.
- * @see SynchronousMode
- */
- void sync();
Execution& getExecution();
+ void sync();
+ void markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer);
+ void sendCompletion();
typedef framing::TransferContent DefaultContent;
protected:
- shared_ptr<SessionCore> impl;
+ shared_ptr<SessionImpl> impl;
framing::ProtocolVersion version;
friend class Connection;
- SessionBase(shared_ptr<SessionCore>);
+ SessionBase(shared_ptr<SessionImpl>);
};
}} // namespace qpid::client
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.cpp Tue Apr 22 05:37:29 2008
@@ -35,7 +35,7 @@
SubscriptionManager::SubscriptionManager(Session& s)
: dispatcher(s), session(s),
messages(UNLIMITED), bytes(UNLIMITED), window(true),
- confirmMode(true), acquireMode(false),
+ acceptMode(0), acquireMode(0),
autoStop(true)
{}
@@ -43,7 +43,7 @@
const std::string& q, const std::string& dest)
{
Completion c = session.messageSubscribe(arg::queue=q, arg::destination=dest,
- arg::confirmMode=confirmMode, arg::acquireMode=acquireMode);
+ arg::acceptMode=acceptMode, arg::acquireMode=acquireMode);
setFlowControl(dest, messages, bytes, window);
return c;
}
@@ -68,7 +68,7 @@
void SubscriptionManager::setFlowControl(
const std::string& dest, uint32_t messages, uint32_t bytes, bool window)
{
- session.messageFlowMode(dest, window);
+ session.messageSetFlowMode(dest, window);
session.messageFlow(dest, 0, messages);
session.messageFlow(dest, 1, bytes);
}
@@ -81,11 +81,13 @@
window=window_;
}
-void SubscriptionManager::setConfirmMode(bool c) { confirmMode=c; }
+void SubscriptionManager::setAcceptMode(bool c) { acceptMode=c; }
void SubscriptionManager::setAcquireMode(bool a) { acquireMode=a; }
void SubscriptionManager::setAckPolicy(const AckPolicy& a) { autoAck=a; }
+
+AckPolicy& SubscriptionManager::getAckPolicy() { return autoAck; }
void SubscriptionManager::cancel(const std::string dest)
{
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/SubscriptionManager.h Tue Apr 22 05:37:29 2008
@@ -53,7 +53,7 @@
uint32_t bytes;
bool window;
AckPolicy autoAck;
- bool confirmMode;
+ bool acceptMode;
bool acquireMode;
bool autoStop;
@@ -116,16 +116,16 @@
*/
void setFlowControl(uint32_t messages, uint32_t bytes, bool window=true);
- /** Set the confirm-mode for new subscriptions. Defaults to true.
- *@param confirm: if true messages must be confirmed by calling
+ /** Set the accept-mode for new subscriptions. Defaults to true.
+ *@param required: if true messages must be confirmed by calling
*Message::acknowledge() or automatically, see setAckPolicy()
*/
- void setConfirmMode(bool confirm);
+ void setAcceptMode(bool required);
/** Set the acquire-mode for new subscriptions. Defaults to false.
*@param acquire: if false messages pre-acquired, if true
* messages are dequed on acknowledgement or on transfer
- * depending on confirmMode.
+ * depending on acceptMode.
*/
void setAcquireMode(bool acquire);
@@ -133,6 +133,10 @@
* Default is to acknowledge every message automatically.
*/
void setAckPolicy(const AckPolicy& autoAck);
+ /**
+ *
+ */
+ AckPolicy& getAckPolicy();
};
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/TypedResult.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/TypedResult.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/TypedResult.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/client/TypedResult.h Tue Apr 22 05:37:29 2008
@@ -33,7 +33,7 @@
bool decoded;
public:
- TypedResult(Future f, shared_ptr<SessionCore> s) : Completion(f, s), decoded(false) {}
+ TypedResult(Future f, shared_ptr<SessionImpl> s) : Completion(f, s), decoded(false) {}
T& get()
{
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/cluster/ClassifierHandler.cpp Tue Apr 22 05:37:29 2008
@@ -32,10 +32,10 @@
void visit(const ExchangeDeclareBody&) { chosen=&classifier.wiring; }
void visit(const ExchangeDeleteBody&) { chosen=&classifier.wiring; }
- void visit(const QueueBindBody&) { chosen=&classifier.wiring; }
+ void visit(const ExchangeBindBody&) { chosen=&classifier.wiring; }
+ void visit(const ExchangeUnbindBody&) { chosen=&classifier.wiring; }
void visit(const QueueDeclareBody&) { chosen=&classifier.wiring; }
void visit(const QueueDeleteBody&) { chosen=&classifier.wiring; }
- void visit(const QueueUnbindBody&) { chosen=&classifier.wiring; }
void defaultVisit(const AMQBody&) { chosen=&classifier.other; }
using framing::FrameDefaultVisitor::visit;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQFrame.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQFrame.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQFrame.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQFrame.h Tue Apr 22 05:37:29 2008
@@ -105,7 +105,6 @@
void setEos(bool isEos) { eos = isEos; }
static uint32_t frameOverhead();
-
private:
void init() { bof = eof = bos = eos = true; subchannel=0; channel=0; }
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQHeaderBody.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQHeaderBody.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQHeaderBody.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQHeaderBody.h Tue Apr 22 05:37:29 2008
@@ -26,8 +26,8 @@
#include "Buffer.h"
#include "qpid/framing/DeliveryProperties.h"
#include "qpid/framing/MessageProperties.h"
-#include "qpid/framing/DeliveryProperties010.h"
-#include "qpid/framing/MessageProperties010.h"
+#include "qpid/framing/PreviewDeliveryProperties.h"
+#include "qpid/framing/PreviewMessageProperties.h"
#include <iostream>
#include <boost/optional.hpp>
@@ -79,8 +79,8 @@
// Could use boost::mpl::fold to construct a larger set.
typedef PropSet< PropSet< PropSet<PropSet<Empty, DeliveryProperties>,
MessageProperties>,
- DeliveryProperties010>,
- MessageProperties010> Properties;
+ PreviewDeliveryProperties>,
+ PreviewMessageProperties> Properties;
Properties properties;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h Tue Apr 22 05:37:29 2008
@@ -32,8 +32,8 @@
namespace qpid {
namespace framing {
-static ProtocolVersion highestProtocolVersion(99, 0);
-//static ProtocolVersion highestProtocolVersion(0, 10);
+//static ProtocolVersion highestProtocolVersion(99, 0);
+static ProtocolVersion highestProtocolVersion(0, 10);
} /* namespace framing */
} /* namespace qpid */
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.cpp Tue Apr 22 05:37:29 2008
@@ -143,20 +143,24 @@
void SequenceSet::remove(const SequenceNumber& s)
{
- for (Ranges::iterator i = ranges.begin(); i != ranges.end() && s >= i->start; i++) {
- if (i->start == s) {
+ for (Ranges::iterator i = ranges.begin(); i != ranges.end() && s >= i->start; i++) {
+ if (i->contains(s)) {
if (i->start == i->end) {
- ranges.erase(i);
- } else {
+ //range is just a single number, so we can delete the whole range
+ i = ranges.erase(i);
+ } else if (i->start == s) {
+ //move the start forward to exclude s
++(i->start);
+ } else if (i->end == s) {
+ //move the end backward to exclude s
+ --(i->end);
+ } else {
+ //need to split range pointed to by i
+ Range r(i->start, (uint32_t)s - 1);
+ i->start = s + 1;
+ i = ranges.insert(i, r);
}
- } else if (i->end == s) {
- --(i->end);
- } else if (i->contains(s)) {
- //need to split range pointed to by i
- Range r(i->start, (uint32_t)s - 1);
- i->start = s + 1;
- ranges.insert(i, r);
+ break;
}
}
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/framing/SequenceSet.h Tue Apr 22 05:37:29 2008
@@ -69,11 +69,12 @@
bool empty() const;
template <class T>
- void for_each(T& t) const
+ T for_each(T& t) const
{
for (Ranges::const_iterator i = ranges.begin(); i != ranges.end(); i++) {
t(i->start, i->end);
}
+ return t;
}
template <class S> void serialize(S& s) { s.split(*this); s(ranges.begin(), ranges.end()); }
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/management/ManagementAgent.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/management/ManagementAgent.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/management/ManagementAgent.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/management/ManagementAgent.cpp Tue Apr 22 05:37:29 2008
@@ -24,6 +24,7 @@
#include "qpid/log/Statement.h"
#include <qpid/broker/Message.h>
#include <qpid/broker/MessageDelivery.h>
+#include "qpid/framing/MessageXTransferBody.h"
#include <list>
#include <iostream>
#include <fstream>
@@ -217,7 +218,7 @@
return;
intrusive_ptr<Message> msg (new Message ());
- AMQFrame method (in_place<MessageTransferBody>(
+ AMQFrame method (in_place<MessageXTransferBody>(
ProtocolVersion(), 0, exchange->getName (), 0, 0));
AMQFrame header (in_place<AMQHeaderBody>());
AMQFrame content(in_place<AMQContentBody>());
@@ -232,8 +233,8 @@
msg->getFrames().append(method);
msg->getFrames().append(header);
- MessageProperties* props =
- msg->getFrames().getHeaders()->get<MessageProperties>(true);
+ PreviewMessageProperties* props =
+ msg->getFrames().getHeaders()->get<PreviewMessageProperties>(true);
props->setContentLength(length);
msg->getFrames().append(content);
@@ -393,8 +394,8 @@
uint64_t objId = inBuffer.getLongLong ();
string replyToKey;
- const framing::MessageProperties* p =
- msg.getFrames().getHeaders()->get<framing::MessageProperties>();
+ const framing::PreviewMessageProperties* p =
+ msg.getFrames().getHeaders()->get<framing::PreviewMessageProperties>();
if (p && p->hasReplyTo())
{
const framing::ReplyTo& rt = p->getReplyTo ();
@@ -600,8 +601,8 @@
uint32_t sequence;
string replyToKey;
- const framing::MessageProperties* p =
- msg.getFrames().getHeaders()->get<framing::MessageProperties>();
+ const framing::PreviewMessageProperties* p =
+ msg.getFrames().getHeaders()->get<framing::PreviewMessageProperties>();
if (p && p->hasReplyTo())
{
const framing::ReplyTo& rt = p->getReplyTo ();
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/ptr_map.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/ptr_map.h?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/ptr_map.h (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/qpid/ptr_map.h Tue Apr 22 05:37:29 2008
@@ -23,98 +23,35 @@
*/
#include <boost/ptr_container/ptr_map.hpp>
-#include <boost/version.hpp>
+#include <boost/utility/enable_if.hpp>
+#include <boost/type_traits/is_same.hpp>
namespace qpid {
-namespace ptr_map {
/** @file
* Workaround for API change between boost 1.33 and 1.34.
*
* To be portable across these versions, code using boost::ptr_map
- * iterators should use get_pointer(i) to get the pointer from
- * a boost::ptr_map iterator.
+ * iterators should use ptr_map_ptr(i) to get the pointer from
+ * boost::ptr_map::iterator i.
*
* Can be removed when we no longer support platforms on 1.33.
*
* @see http://www.boost.org/libs/ptr_container/doc/ptr_container.html#upgrading-from-boost-v-1-33
*/
-#include <boost/type_traits/remove_const.hpp>
-#if (BOOST_VERSION < 103400)
+typedef boost::is_same<boost::ptr_map<int, int>::iterator::value_type,
+ int> IsOldPtrMap;
-template <class PtrMapIter>
-typename PtrMapIter::pointer get_pointer(const PtrMapIter& i)
-{ return &*i; }
+template <class Iter>
+typename boost::enable_if<IsOldPtrMap, typename Iter::value_type*>::type
+ptr_map_ptr(const Iter& i) { return &*i; }
+
+template <class Iter>
+typename boost::disable_if<IsOldPtrMap, typename Iter::value_type::second_type>::type
+ptr_map_ptr(const Iter& i) { return i->second; }
-#else
-
-template <class PtrMapIter>
-typename boost::remove_const<typename PtrMapIter::value_type::second_type>::type
-get_pointer(const PtrMapIter& i)
-{ return i->second; }
-
-#endif
-
-}} // namespace qpid::ptr_map
-
-#endif /*!QPID_PTR_MAP*/
-#ifndef QPID_PTR_MAP
-#define QPID_PTR_MAP
-
-/*
- *
- * 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/ptr_container/ptr_map.hpp>
-#include <boost/version.hpp>
-
-namespace qpid {
-namespace ptr_map {
-
-/** @file
- * Workaround for API change between boost 1.33 and 1.34.
- *
- * To be portable across these versions, code using boost::ptr_map
- * iterators should use get_pointer(i) to get the pointer from
- * a boost::ptr_map iterator.
- *
- * Can be removed when we no longer support platforms on 1.33.
- *
- * @see http://www.boost.org/libs/ptr_container/doc/ptr_container.html#upgrading-from-boost-v-1-33
- */
-#if (BOOST_VERSION < 103400)
-
-template <class PtrMapIter>
-typename PtrMapIter::pointer get_pointer(const PtrMapIter& i)
-{ return &*i; }
-
-#else
-
-template <class PtrMapIter>
-typename PtrMapIter::value_type::second_type get_pointer(const PtrMapIter& i)
-{ return i->second; }
-
-#endif
-
-}} // namespace qpid::ptr_map
+} // namespace qpid
#endif /*!QPID_PTR_MAP*/
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ClientSessionTest.cpp Tue Apr 22 05:37:29 2008
@@ -20,6 +20,7 @@
*/
#include "unit_test.h"
#include "BrokerFixture.h"
+#include "qpid/client/AckPolicy.h"
#include "qpid/client/Dispatcher.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Thread.h"
@@ -118,13 +119,14 @@
ClientSessionFixture fix;
fix.session=fix.connection.newSession(ASYNC);
fix.declareSubscribe();
- fix.session.messageTransfer(content=TransferContent("my-message", "my-queue"));
+ fix.session.messageTransfer(acceptMode=1, content=TransferContent("my-message", "my-queue"));
//get & test the message:
FrameSet::shared_ptr msg = fix.session.get();
BOOST_CHECK(msg->isA<MessageTransferBody>());
BOOST_CHECK_EQUAL(string("my-message"), msg->getContent());
//confirm receipt:
- fix.session.getExecution().completed(msg->getId(), true, true);
+ AckPolicy autoAck;
+ autoAck.ack(Message(*msg), fix.session);
}
QPID_AUTO_TEST_CASE(testDispatcher)
@@ -161,6 +163,8 @@
}
*/
+/*
+ * GS (18-APR-2008): disabled resume tests until resumption for 0-10 final spec is implemented
QPID_AUTO_TEST_CASE(_FIXTURE)
{
ClientSessionFixture fix;
@@ -195,7 +199,7 @@
FrameSet::shared_ptr msg = fix.session.get();
BOOST_CHECK_EQUAL(string("my-message"), msg->getContent());
}
-
+*/
/**
* Currently broken due to a deadlock in SessionCore
*
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ExchangeTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ExchangeTest.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ExchangeTest.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/ExchangeTest.cpp Tue Apr 22 05:37:29 2008
@@ -30,7 +30,6 @@
#include "qpid/broker/TopicExchange.h"
#include "qpid_test_plugin.h"
#include <iostream>
-#include "qpid/framing/BasicGetBody.h"
#include "MessageUtils.h"
using boost::intrusive_ptr;
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/FramingTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/FramingTest.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/FramingTest.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/FramingTest.cpp Tue Apr 22 05:37:29 2008
@@ -23,8 +23,6 @@
#include "qpid/client/Connection.h"
#include "qpid/client/Connector.h"
#include "qpid/framing/AMQP_HighestVersion.h"
-#include "qpid/framing/BasicGetOkBody.h"
-#include "qpid/framing/ConnectionRedirectBody.h"
#include "qpid/framing/ProtocolVersion.h"
#include "qpid/framing/all_method_bodies.h"
#include "qpid/framing/amqp_framing.h"
@@ -54,16 +52,12 @@
class FramingTest : public CppUnit::TestCase
{
CPPUNIT_TEST_SUITE(FramingTest);
- CPPUNIT_TEST(testBasicQosBody);
+ CPPUNIT_TEST(testMessageTransferBody);
CPPUNIT_TEST(testConnectionSecureBody);
CPPUNIT_TEST(testConnectionRedirectBody);
- CPPUNIT_TEST(testAccessRequestBody);
- CPPUNIT_TEST(testBasicConsumeBody);
+ CPPUNIT_TEST(testQueueDeclareBody);
CPPUNIT_TEST(testConnectionRedirectBodyFrame);
- CPPUNIT_TEST(testBasicConsumeOkBodyFrame);
- CPPUNIT_TEST(testInlineContent);
- CPPUNIT_TEST(testContentReference);
- CPPUNIT_TEST(testContentValidation);
+ CPPUNIT_TEST(testMessageCancelBodyFrame);
CPPUNIT_TEST_SUITE_END();
private:
@@ -74,14 +68,14 @@
FramingTest() : version(highestProtocolVersion) {}
- void testBasicQosBody()
+ void testMessageTransferBody()
{
Buffer wbuff(buffer, sizeof(buffer));
- BasicQosBody in(version, 0xCAFEBABE, 0xABBA, true);
+ MessageTransferBody in(version, "my-exchange", 1, 1);
in.encode(wbuff);
Buffer rbuff(buffer, sizeof(buffer));
- BasicQosBody out(version);
+ MessageTransferBody out(version);
out.decode(rbuff);
CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
}
@@ -104,7 +98,11 @@
Buffer wbuff(buffer, sizeof(buffer));
std::string a = "hostA";
std::string b = "hostB";
- ConnectionRedirectBody in(version, a, b);
+ Array hosts(0x95);
+ hosts.add(boost::shared_ptr<FieldValue>(new Str16Value(a)));
+ hosts.add(boost::shared_ptr<FieldValue>(new Str16Value(b)));
+
+ ConnectionRedirectBody in(version, a, hosts);
in.encode(wbuff);
Buffer rbuff(buffer, sizeof(buffer));
@@ -113,41 +111,28 @@
CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
}
- void testAccessRequestBody()
- {
- Buffer wbuff(buffer, sizeof(buffer));
- std::string s = "text";
- AccessRequestBody in(version, s, true, false, true, false, true);
- in.encode(wbuff);
-
- Buffer rbuff(buffer, sizeof(buffer));
- AccessRequestBody out(version);
- out.decode(rbuff);
- CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
- }
-
- void testBasicConsumeBody()
+ void testQueueDeclareBody()
{
Buffer wbuff(buffer, sizeof(buffer));
- std::string q = "queue";
- std::string t = "tag";
- BasicConsumeBody in(version, 0, q, t, false, true, false, false,
- FieldTable());
+ QueueDeclareBody in(version, "name", "dlq", true, false, true, false, FieldTable());
in.encode(wbuff);
Buffer rbuff(buffer, sizeof(buffer));
- BasicConsumeBody out(version);
+ QueueDeclareBody out(version);
out.decode(rbuff);
CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
}
-
void testConnectionRedirectBodyFrame()
{
Buffer wbuff(buffer, sizeof(buffer));
std::string a = "hostA";
std::string b = "hostB";
- AMQFrame in(in_place<ConnectionRedirectBody>(version, a, b));
+ Array hosts(0x95);
+ hosts.add(boost::shared_ptr<FieldValue>(new Str16Value(a)));
+ hosts.add(boost::shared_ptr<FieldValue>(new Str16Value(b)));
+
+ AMQFrame in(in_place<ConnectionRedirectBody>(version, a, hosts));
in.setChannel(999);
in.encode(wbuff);
@@ -157,11 +142,10 @@
CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
}
- void testBasicConsumeOkBodyFrame()
+ void testMessageCancelBodyFrame()
{
Buffer wbuff(buffer, sizeof(buffer));
- std::string s = "hostA";
- AMQFrame in(in_place<BasicConsumeOkBody>(version, s));
+ AMQFrame in(in_place<MessageCancelBody>(version, "tag"));
in.setChannel(999);
in.encode(wbuff);
@@ -169,56 +153,6 @@
AMQFrame out;
out.decode(rbuff);
CPPUNIT_ASSERT_EQUAL(tostring(in), tostring(out));
- }
-
- void testInlineContent() {
- Buffer wbuff(buffer, sizeof(buffer));
- Content content(INLINE, "MyData");
- CPPUNIT_ASSERT(content.isInline());
- content.encode(wbuff);
-
- Buffer rbuff(buffer, sizeof(buffer));
- Content recovered;
- recovered.decode(rbuff);
- CPPUNIT_ASSERT(recovered.isInline());
- CPPUNIT_ASSERT_EQUAL(content.getValue(), recovered.getValue());
- }
-
- void testContentReference() {
- Buffer wbuff(buffer, sizeof(buffer));
- Content content(REFERENCE, "MyRef");
- CPPUNIT_ASSERT(content.isReference());
- content.encode(wbuff);
-
- Buffer rbuff(buffer, sizeof(buffer));
- Content recovered;
- recovered.decode(rbuff);
- CPPUNIT_ASSERT(recovered.isReference());
- CPPUNIT_ASSERT_EQUAL(content.getValue(), recovered.getValue());
- }
-
- void testContentValidation() {
- try {
- Content content(REFERENCE, "");
- CPPUNIT_ASSERT(false);//fail, expected exception
- } catch (const InvalidArgumentException& e) {}
-
- try {
- Content content(2, "Blah");
- CPPUNIT_ASSERT(false);//fail, expected exception
- } catch (const SyntaxErrorException& e) {}
-
- try {
- Buffer wbuff(buffer, sizeof(buffer));
- wbuff.putOctet(2);
- wbuff.putLongString("blah, blah");
-
- Buffer rbuff(buffer, sizeof(buffer));
- Content content;
- content.decode(rbuff);
- CPPUNIT_FAIL("Expected exception");
- } catch (Exception& e) {}
-
}
};
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/HeaderTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/HeaderTest.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/HeaderTest.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/HeaderTest.cpp Tue Apr 22 05:37:29 2008
@@ -61,16 +61,13 @@
out.castBody<AMQHeaderBody>()->get<MessageProperties>(true);
props1->setContentLength(42);
- props1->setMessageId("messageId");
+ props1->setMessageId(Uuid(true));
props1->setCorrelationId("correlationId");
props1->setReplyTo(ReplyTo("ex","key"));
props1->setContentType("contentType");
props1->setContentEncoding("contentEncoding");
- props1->setType("type");
props1->setUserId("userId");
props1->setAppId("appId");
- props1->setTransactionId("transactionId");
- props1->setSecurityToken("securityToken");
char buff[10000];
Buffer wbuffer(buff, 10000);
@@ -87,11 +84,8 @@
CPPUNIT_ASSERT_EQUAL(props1->getCorrelationId(), props2->getCorrelationId());
CPPUNIT_ASSERT_EQUAL(props1->getContentType(), props2->getContentType());
CPPUNIT_ASSERT_EQUAL(props1->getContentEncoding(), props2->getContentEncoding());
- CPPUNIT_ASSERT_EQUAL(props1->getType(), props2->getType());
CPPUNIT_ASSERT_EQUAL(props1->getUserId(), props2->getUserId());
CPPUNIT_ASSERT_EQUAL(props1->getAppId(), props2->getAppId());
- CPPUNIT_ASSERT_EQUAL(props1->getTransactionId(), props2->getTransactionId());
- CPPUNIT_ASSERT_EQUAL(props1->getSecurityToken(), props2->getSecurityToken());
}
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/InlineVector.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/InlineVector.cpp?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/InlineVector.cpp (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/InlineVector.cpp Tue Apr 22 05:37:29 2008
@@ -66,24 +66,54 @@
}
QPID_AUTO_TEST_CASE(testInsert) {
- Vec v;
- v.push_back(1);
- BOOST_CHECK_EQUAL(v.size(), 1u);
- BOOST_CHECK_EQUAL(v.back(), 1);
- BOOST_CHECK(isInline(v));
-
- v.insert(v.begin(), 2);
- BOOST_CHECK_EQUAL(v.size(), 2u);
- BOOST_CHECK_EQUAL(v.back(), 1);
- BOOST_CHECK(isInline(v));
+ {
+ Vec v;
+ v.push_back(1);
+ BOOST_CHECK_EQUAL(v.size(), 1u);
+ BOOST_CHECK_EQUAL(v.back(), 1);
+ BOOST_CHECK(isInline(v));
+
+ v.insert(v.begin(), 2);
+ BOOST_CHECK_EQUAL(v.size(), 2u);
+ BOOST_CHECK_EQUAL(v.back(), 1);
+ BOOST_CHECK(isInline(v));
- v.push_back(3);
- BOOST_CHECK(isInline(v));
+ v.push_back(3);
+ BOOST_CHECK(isInline(v));
- v.push_back(4);
- BOOST_CHECK_EQUAL(v.size(), 4u);
+ v.push_back(4);
+
+ BOOST_CHECK(!isInline(v));
+ }
+ {
+ Vec v(3,42);
+ v.insert(v.begin(), 9);
+ BOOST_CHECK_EQUAL(v.size(), 4u);
+ BOOST_CHECK(!isInline(v));
+ }
+ {
+ Vec v(3,42);
+ v.insert(v.begin()+1, 9);
+ BOOST_CHECK(!isInline(v));
+ BOOST_CHECK_EQUAL(v.size(), 4u);
+ }
+}
+
+QPID_AUTO_TEST_CASE(testAssign) {
+ Vec v(3,42);
+ Vec u;
+ u = v;
+ BOOST_CHECK(isInline(u));
+ u.push_back(4);
+ BOOST_CHECK(!isInline(u));
+ v = u;
BOOST_CHECK(!isInline(v));
}
+QPID_AUTO_TEST_CASE(testResize) {
+ Vec v;
+ v.resize(5);
+ BOOST_CHECK(!isInline(v));
+}
QPID_AUTO_TEST_SUITE_END()
Modified: incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/Makefile.am?rev=650478&r1=650477&r2=650478&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/cpp/src/tests/Makefile.am Tue Apr 22 05:37:29 2008
@@ -6,7 +6,7 @@
lib_client = $(abs_builddir)/../libqpidclient.la
lib_common = $(abs_builddir)/../libqpidcommon.la
lib_broker = $(abs_builddir)/../libqpidbroker.la
-lib_amqp_0_10 = $(abs_builddir)/../libqpidamqp_0_10.la
+# lib_amqp_0_10 = $(abs_builddir)/../libqpidamqp_0_10.la
#
# Initialize variables that are incremented with +=
@@ -28,7 +28,7 @@
TESTS+=unit_test
check_PROGRAMS+=unit_test
unit_test_LDADD=-lboost_unit_test_framework -lboost_regex \
- $(lib_client) $(lib_broker) $(lib_amqp_0_10)
+ $(lib_client) $(lib_broker) # $(lib_amqp_0_10)
unit_test_SOURCES= unit_test.cpp unit_test.h \
BrokerFixture.h SocketProxy.h \
exception_test.cpp \
@@ -36,16 +36,21 @@
SessionState.cpp Blob.cpp logging.cpp \
Url.cpp Uuid.cpp \
Shlib.cpp FieldValue.cpp FieldTable.cpp Array.cpp \
+ InlineAllocator.cpp \
InlineVector.cpp \
ISList.cpp IList.cpp \
ClientSessionTest.cpp \
SequenceSet.cpp \
- amqp_0_10/serialize.cpp allSegmentTypes.h \
- amqp_0_10/ProxyTemplate.cpp \
- amqp_0_10/apply.cpp \
IncompleteMessageList.cpp \
- amqp_0_10/Map.cpp \
- amqp_0_10/handlers.cpp
+ RangeSet.cpp
+
+# Disabled till we move to amqp_0_10 codec.
+# amqp_0_10/serialize.cpp allSegmentTypes.h \
+# amqp_0_10/ProxyTemplate.cpp \
+# amqp_0_10/apply.cpp \
+# amqp_0_10/Map.cpp \
+# amqp_0_10/handlers.cpp
+
check_LTLIBRARIES += libshlibtest.la
libshlibtest_la_LDFLAGS = -module -rpath $(abs_builddir)