You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2007/10/26 21:48:37 UTC
svn commit: r588761 [2/3] - in /incubator/qpid/trunk/qpid/cpp:
rubygen/templates/ src/ src/qpid/ src/qpid/broker/ src/qpid/client/
src/qpid/framing/ src/qpid/log/ src/qpid/sys/ src/qpid/sys/apr/
src/qpid/sys/posix/ src/tests/
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.cpp Fri Oct 26 12:48:31 2007
@@ -24,105 +24,301 @@
#include "FutureResponse.h"
#include "FutureResult.h"
#include "ConnectionImpl.h"
-
+#include "qpid/framing/FrameSet.h"
#include "qpid/framing/constants.h"
+#include "qpid/framing/ClientInvoker.h"
+#include "qpid/log/Statement.h"
#include <boost/bind.hpp>
-using namespace qpid::client;
+namespace qpid {
+namespace client {
+
using namespace qpid::framing;
-SessionCore::SessionCore(shared_ptr<ConnectionImpl> conn, uint16_t ch, uint64_t maxFrameSize)
- : connection(conn), channel(ch), l2(*this), l3(maxFrameSize),
- uuid(false), sync(false)
+namespace { const std::string OK="ok"; }
+
+typedef sys::Monitor::ScopedLock Lock;
+typedef sys::Monitor::ScopedUnlock UnLock;
+
+inline void SessionCore::invariant() const {
+ switch (state.get()) {
+ case OPENING:
+ assert(!session);
+ assert(code==REPLY_SUCCESS);
+ assert(connection);
+ assert(channel.get());
+ assert(channel.next == connection.get());
+ break;
+ case RESUMING:
+ assert(session);
+ assert(session->getState() == SessionState::RESUMING);
+ assert(code==REPLY_SUCCESS);
+ assert(connection);
+ assert(channel.get());
+ assert(channel.next == connection.get());
+ break;
+ case OPEN:
+ case CLOSING:
+ case SUSPENDING:
+ assert(session);
+ assert(code==REPLY_SUCCESS);
+ assert(connection);
+ assert(channel.get());
+ assert(channel.next == connection.get());
+ break;
+ case SUSPENDED:
+ assert(code==REPLY_SUCCESS);
+ assert(session);
+ assert(!connection);
+ break;
+ case CLOSED:
+ assert(!session);
+ assert(!connection);
+ break;
+ }
+}
+
+inline void SessionCore::setState(State s) {
+ state = s;
+ invariant();
+}
+
+inline void SessionCore::waitFor(State s) {
+ invariant();
+ // We can be CLOSED or SUSPENDED by error at any time.
+ state.waitFor(States(s, CLOSED, SUSPENDED));
+ check();
+ assert(state==s);
+ invariant();
+}
+
+SessionCore::SessionCore(shared_ptr<ConnectionImpl> conn,
+ uint16_t ch, uint64_t maxFrameSize)
+ : l3(maxFrameSize),
+ sync(false),
+ channel(ch),
+ proxy(channel),
+ state(OPENING)
{
- l2.next = &l3;
l3.out = &out;
- out.next = connection.get();
+ attaching(conn);
}
-SessionCore::~SessionCore() {}
+void SessionCore::attaching(shared_ptr<ConnectionImpl> c) {
+ assert(c);
+ assert(channel.get());
+ connection = c;
+ channel.next = connection.get();
+ code = REPLY_SUCCESS;
+ text = OK;
+ state = session ? RESUMING : OPENING;
+ invariant();
+}
+
+SessionCore::~SessionCore() {
+ Lock l(state);
+ invariant();
+ detach(COMMAND_INVALID, "Session deleted");
+ state.waitAll();
+}
+
+void SessionCore::detach(int c, const std::string& t) {
+ connection.reset();
+ channel.next = 0;
+ code=c;
+ text=t;
+}
+
+void SessionCore::doClose(int code, const std::string& text) {
+ if (state != CLOSED) {
+ session.reset();
+ l3.getDemux().close();
+ l3.getCompletionTracker().close();
+ detach(code, text);
+ setState(CLOSED);
+ }
+ invariant();
+}
-ExecutionHandler& SessionCore::getExecution()
-{
- checkClosed();
- return l3;
+void SessionCore::doSuspend(int code, const std::string& text) {
+ if (state != CLOSED) {
+ invariant();
+ detach(code, text);
+ session->suspend();
+ setState(SUSPENDED);
+ }
}
-FrameSet::shared_ptr SessionCore::get()
-{
- checkClosed();
- return l3.getDemux().getDefault().pop();
+ExecutionHandler& SessionCore::getExecution() { // user thread
+ return l3;
}
-void SessionCore::setSync(bool s)
-{
+void SessionCore::setSync(bool s) { // user thread
sync = s;
}
-bool SessionCore::isSync()
-{
+bool SessionCore::isSync() { // user thread
return sync;
}
-namespace {
-struct ClosedOnExit {
- SessionCore& core;
- int code;
- std::string text;
- ClosedOnExit(SessionCore& s, int c, const std::string& t)
- : core(s), code(c), text(t) {}
- ~ClosedOnExit() { core.closed(code, text); }
-};
+FrameSet::shared_ptr SessionCore::get() { // user thread
+ // No lock here: pop does a blocking wait.
+ return l3.getDemux().getDefault().pop();
}
-void SessionCore::close()
-{
- checkClosed();
- ClosedOnExit closer(*this, CHANNEL_ERROR, "Session closed by user.");
- l2.close();
+void SessionCore::open(uint32_t detachedLifetime) { // user thread
+ Lock l(state);
+ check(state==OPENING && !session,
+ COMMAND_INVALID, QPID_MSG("Cannot re-open a session."));
+ proxy.open(detachedLifetime);
+ waitFor(OPEN);
}
-void SessionCore::suspend() {
- checkClosed();
- ClosedOnExit closer(*this, CHANNEL_ERROR, "Client session is suspended");
- l2.suspend();
+void SessionCore::close() { // user thread
+ Lock l(state);
+ check();
+ if (state==OPEN) {
+ setState(CLOSING);
+ proxy.close();
+ waitFor(CLOSED);
+ }
+ else
+ doClose(REPLY_SUCCESS, OK);
}
-void SessionCore::closed(uint16_t code, const std::string& text)
-{
- out.next = 0;
- reason.code = code;
- reason.text = text;
- l2.closed();
- l3.getDemux().close();
- l3.getCompletionTracker().close();
+void SessionCore::suspend() { // user thread
+ Lock l(state);
+ checkOpen();
+ setState(SUSPENDING);
+ proxy.suspend();
+ waitFor(SUSPENDED);
+}
+
+void SessionCore::setChannel(uint16_t ch) { channel=ch; }
+
+void SessionCore::resume(shared_ptr<ConnectionImpl> c) {
+ // user thread
+ {
+ Lock l(state);
+ if (state==OPEN)
+ doSuspend(REPLY_SUCCESS, OK);
+ check(state==SUSPENDED, COMMAND_INVALID, QPID_MSG("Session cannot be resumed."));
+ SequenceNumber sendAck=session->resuming();
+ attaching(c);
+ proxy.resume(getId());
+ waitFor(OPEN);
+ proxy.ack(sendAck, SequenceNumberSet());
+ // FIXME aconway 2007-10-23: Replay inside the lock might be a prolem
+ // for large replay sets.
+ SessionState::Replay replay=session->replay();
+ for (SessionState::Replay::iterator i = replay.begin();
+ i != replay.end(); ++i)
+ {
+ invariant();
+ channel.handle(*i); // Direct to channel.
+ check();
+ }
+ }
}
-void SessionCore::checkClosed() const
+void SessionCore::assertOpen() const {
+ Lock l(state);
+ checkOpen();
+}
+
+// network thread
+void SessionCore::attached(const Uuid& sessionId,
+ uint32_t /*detachedLifetime*/)
{
- // TODO: could have been a connection exception
- if(out.next == 0)
- throw ChannelException(reason.code, reason.text);
+ Lock l(state);
+ invariant();
+ check(state == OPENING || state == RESUMING,
+ COMMAND_INVALID, QPID_MSG("Received unexpected session.attached"));
+ if (state==OPENING) { // New session
+ // FIXME aconway 2007-10-17: arbitrary ack value of 100 for
+ // client, allow configuration.
+ session=in_place<SessionState>(100, sessionId);
+ setState(OPEN);
+ }
+ else { // RESUMING
+ check(sessionId == session->getId(),
+ INVALID_ARGUMENT, QPID_MSG("session.resumed has invalid ID."));
+ // Don't setState yet, wait for first incoming ack.
+ }
+}
+
+void SessionCore::detached() { // network thread
+ Lock l(state);
+ check(state == SUSPENDING,
+ COMMAND_INVALID, QPID_MSG("Received unexpected session.detached."));
+ connection->erase(channel);
+ doSuspend(REPLY_SUCCESS, OK);
+}
+
+void SessionCore::ack(uint32_t ack, const SequenceNumberSet&) {
+ Lock l(state);
+ invariant();
+ check(state==OPEN || state==RESUMING,
+ COMMAND_INVALID, QPID_MSG("Received unexpected session.ack"));
+ session->receivedAck(ack);
+ if (state==RESUMING) {
+ setState(OPEN);
+ }
+ invariant();
}
-void SessionCore::open(uint32_t detachedLifetime) {
- assert(out.next);
- l2.open(detachedLifetime);
+void SessionCore::closed(uint16_t code, const std::string& text)
+{ // network thread
+ Lock l(state);
+ invariant();
+ doClose(code, text);
+}
+
+// closed by connection
+void SessionCore::connectionClosed(uint16_t code, const std::string& text) {
+ Lock l(state);
+ try {
+ doClose(code, text);
+ } catch(...) { assert (0); }
+}
+
+void SessionCore::connectionBroke(uint16_t code, const std::string& text) {
+ Lock l(state);
+ try {
+ doSuspend(code, text);
+ } catch (...) { assert(0); }
+}
+
+void SessionCore::check() const { // Called with lock held.
+ invariant();
+ if (code != REPLY_SUCCESS)
+ throwReplyException(code, text);
+}
+
+void SessionCore::check(bool cond, int newCode, const std::string& msg) const {
+ check();
+ if (!cond) {
+ const_cast<SessionCore*>(this)->doClose(newCode, msg);
+ throwReplyException(code, text);
+ }
}
-void SessionCore::resume(shared_ptr<ConnectionImpl> conn) {
- connection = conn;
- out.next = connection.get();
- l2.resume();
+void SessionCore::checkOpen() const {
+ if (state==SUSPENDED) {
+ std::string cause;
+ if (code != REPLY_SUCCESS)
+ cause=" by :"+text;
+ throw CommandInvalidException(QPID_MSG("Session is suspended" << cause));
+ }
+ check(state==OPEN, COMMAND_INVALID, QPID_MSG("Session is not open"));
}
Future SessionCore::send(const AMQBody& command)
-{
- checkClosed();
-
+{
+ Lock l(state);
+ checkOpen();
command.getMethod()->setSync(sync);
-
Future f;
//any result/response listeners must be set before the command is sent
if (command.getMethod()->resultExpected()) {
@@ -145,21 +341,61 @@
Future SessionCore::send(const AMQBody& command, const MethodContent& content)
{
- checkClosed();
+ Lock l(state);
+ checkOpen();
//content bearing methods don't currently have responses or
//results, if that changes should follow procedure for the other
//send method impl:
return Future(l3.send(command, content));
}
+// Network thread.
void SessionCore::handleIn(AMQFrame& frame) {
- l2.handle(frame);
+ try {
+ // Cast to expose private SessionHandler functions.
+ if (!invoke(static_cast<SessionHandler&>(*this), *frame.getBody())) {
+ session->received(frame);
+ l3.handle(frame);
+ }
+ } catch (const ChannelException& e) {
+ QPID_LOG(error, "Channel exception:" << e.what());
+ doClose(e.code, e.what());
+ }
}
void SessionCore::handleOut(AMQFrame& frame)
{
- checkClosed();
- frame.setChannel(channel);
- out.next->handle(frame);
+ Lock l(state);
+ if (state==OPEN) {
+ if (session->sent(frame))
+ proxy.solicitAck();
+ channel.handle(frame);
+ }
+}
+
+void SessionCore::solicitAck( ) {
+ Lock l(state);
+ checkOpen();
+ proxy.ack(session->sendingAck(), SequenceNumberSet());
+}
+
+void SessionCore::flow(bool) {
+ assert(0); throw NotImplementedException("session.flow");
+}
+
+void SessionCore::flowOk(bool /*active*/) {
+ assert(0); throw NotImplementedException("session.flow");
+}
+
+void SessionCore::highWaterMark(uint32_t /*lastSentMark*/) {
+ // FIXME aconway 2007-10-02: may be removed from spec.
+ assert(0); throw NotImplementedException("session.highWaterMark");
+}
+
+const Uuid SessionCore::getId() const {
+ if (session)
+ return session->getId();
+ throw Exception(QPID_MSG("Closed session, no ID."));
}
+}} // namespace qpid::client
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionCore.h Fri Oct 26 12:48:31 2007
@@ -22,17 +22,25 @@
#ifndef _SessionCore_
#define _SessionCore_
-#include <boost/function.hpp>
-#include <boost/shared_ptr.hpp>
-#include "qpid/framing/AMQMethodBody.h"
+#include "qpid/shared_ptr.h"
#include "qpid/framing/FrameHandler.h"
-#include "qpid/framing/FrameSet.h"
-#include "qpid/framing/MethodContent.h"
-#include "qpid/framing/Uuid.h"
-#include "SessionHandler.h"
+#include "qpid/framing/ChannelHandler.h"
+#include "qpid/framing/SessionState.h"
+#include "qpid/framing/SequenceNumber.h"
+#include "qpid/framing/AMQP_ClientOperations.h"
+#include "qpid/framing/AMQP_ServerProxy.h"
+#include "qpid/sys/StateMonitor.h"
#include "ExecutionHandler.h"
+#include <boost/optional.hpp>
+
namespace qpid {
+namespace framing {
+class FrameSet;
+class MethodContent;
+class SequenceNumberSet;
+}
+
namespace client {
class Future;
@@ -43,60 +51,90 @@
* Attaches to a SessionHandler when active, detaches
* when closed.
*/
-class SessionCore : public framing::FrameHandler::InOutHandler
+class SessionCore : public framing::FrameHandler::InOutHandler,
+ private framing::AMQP_ClientOperations::SessionHandler
{
- struct Reason
- {
- uint16_t code;
- std::string text;
- };
-
- shared_ptr<ConnectionImpl> connection;
- uint16_t channel;
- SessionHandler l2;
- ExecutionHandler l3;
- framing::Uuid uuid;
- volatile bool sync;
- Reason reason;
-
- protected:
- void handleIn(framing::AMQFrame& frame);
- void handleOut(framing::AMQFrame& frame);
-
public:
SessionCore(shared_ptr<ConnectionImpl>, uint16_t channel, uint64_t maxFrameSize);
~SessionCore();
framing::FrameSet::shared_ptr get();
+ const framing::Uuid getId() const;
+ uint16_t getChannel() const { return channel; }
+ void assertOpen() const;
- framing::Uuid getId() const { return uuid; }
- void setId(const framing::Uuid& id) { uuid= id; }
-
- uint16_t getChannel() const { assert(channel); return channel; }
- void setChannel(uint16_t ch) { assert(ch); channel=ch; }
-
+ // NOTE: Public functions called in user thread.
void open(uint32_t detachedLifetime);
-
- /** Closed by client code */
void close();
-
- /** Closed by peer */
- void closed(uint16_t code, const std::string& text);
-
void resume(shared_ptr<ConnectionImpl>);
void suspend();
+ void setChannel(uint16_t channel);
- void setSync(bool);
+ void setSync(bool s);
bool isSync();
ExecutionHandler& getExecution();
- void checkClosed() const;
Future send(const framing::AMQBody& command);
+
Future send(const framing::AMQBody& command, const framing::MethodContent& content);
-};
-}
-}
+ void connectionClosed(uint16_t code, const std::string& text);
+ void connectionBroke(uint16_t code, const std::string& text);
+
+ private:
+ enum State {
+ OPENING,
+ RESUMING,
+ OPEN,
+ CLOSING,
+ SUSPENDING,
+ SUSPENDED,
+ CLOSED
+ };
+ typedef framing::AMQP_ClientOperations::SessionHandler SessionHandler;
+ typedef sys::StateMonitor<State, CLOSED> StateMonitor;
+ typedef StateMonitor::Set States;
+
+ inline void invariant() const;
+ inline void setState(State s);
+ inline void waitFor(State);
+ void doClose(int code, const std::string& text);
+ void doSuspend(int code, const std::string& text);
+
+ /** If there is an error, throw the exception */
+ void check(bool condition, int code, const std::string& text) const;
+ /** Throw if *error */
+ void check() const;
+
+ void handleIn(framing::AMQFrame& frame);
+ void handleOut(framing::AMQFrame& frame);
+
+ // Private functions are called by broker in network thread.
+ void attached(const framing::Uuid& sessionId, uint32_t detachedLifetime);
+ void flow(bool active);
+ void flowOk(bool active);
+ void detached();
+ void ack(uint32_t cumulativeSeenMark,
+ const framing::SequenceNumberSet& seenFrameSet);
+ void highWaterMark(uint32_t lastSentMark);
+ void solicitAck();
+ void closed(uint16_t code, const std::string& text);
+
+ void attaching(shared_ptr<ConnectionImpl>);
+ void detach(int code, const std::string& text);
+ void checkOpen() const;
+
+ int code; // Error code
+ std::string text; // Error text
+ boost::optional<framing::SessionState> session;
+ shared_ptr<ConnectionImpl> connection;
+ ExecutionHandler l3;
+ volatile bool sync;
+ framing::ChannelHandler channel;
+ framing::AMQP_ServerProxy::Session proxy;
+ mutable StateMonitor state;
+};
+}} // namespace qpid::client
#endif
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.cpp Fri Oct 26 12:48:31 2007
@@ -60,7 +60,7 @@
stateLock.notifyAll();
}
-int StateManager::getState()
+int StateManager::getState() const
{
Monitor::ScopedLock l(stateLock);
return state;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/StateManager.h Fri Oct 26 12:48:31 2007
@@ -30,12 +30,12 @@
class StateManager
{
int state;
- sys::Monitor stateLock;
+ mutable sys::Monitor stateLock;
public:
StateManager(int initial);
void setState(int state);
- int getState();
+ int getState() const ;
void waitForStateChange(int current);
void waitFor(std::set<int> states);
void waitFor(int state);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQFrame.cpp Fri Oct 26 12:48:31 2007
@@ -20,9 +20,9 @@
*/
#include "AMQFrame.h"
-#include "qpid/QpidError.h"
#include "qpid/framing/variant.h"
#include "qpid/framing/AMQMethodBody.h"
+#include "qpid/framing/reply_exceptions.h"
#include <boost/format.hpp>
@@ -103,7 +103,7 @@
uint8_t flags = buffer.getOctet();
uint8_t framing_version = (flags & 0xc0) >> 6;
if (framing_version != 0)
- THROW_QPID_ERROR(FRAMING_ERROR, "Framing version unsupported");
+ throw SyntaxErrorException(QPID_MSG("Framing version unsupported"));
bof = flags & 0x08;
eof = flags & 0x04;
bos = flags & 0x02;
@@ -111,7 +111,7 @@
uint8_t type = buffer.getOctet();
uint16_t frame_size = buffer.getShort();
if (frame_size < frameOverhead()-1)
- THROW_QPID_ERROR(FRAMING_ERROR, "Frame size too small");
+ throw SyntaxErrorException(QPID_MSG("Frame size too small"));
uint8_t reserved1 = buffer.getOctet();
uint8_t field1 = buffer.getOctet();
subchannel = field1 & 0x0f;
@@ -121,7 +121,7 @@
// Verify that the protocol header meets current spec
// TODO: should we check reserved2 against zero as well? - the spec isn't clear
if ((flags & 0x30) != 0 || reserved1 != 0 || (field1 & 0xf0) != 0)
- THROW_QPID_ERROR(FRAMING_ERROR, "Reserved bits not zero");
+ throw SyntaxErrorException(QPID_MSG("Reserved bits not zero"));
// TODO: should no longer care about body size and only pass up B,E,b,e flags
uint16_t body_size = frame_size + 1 - frameOverhead();
@@ -133,7 +133,7 @@
uint8_t end = buffer.getOctet();
if (end != 0xCE)
- THROW_QPID_ERROR(FRAMING_ERROR, "Frame end not found");
+ throw SyntaxErrorException(QPID_MSG("Frame end not found"));
return true;
}
@@ -147,9 +147,7 @@
case HEARTBEAT_BODY: body = AMQHeartbeatBody(); break;
default:
- THROW_QPID_ERROR(
- FRAMING_ERROR,
- boost::format("Unknown frame type %d") % type);
+ throw SyntaxErrorException(QPID_MSG("Invalid frame type " << type));
}
boost::apply_visitor(DecodeVisitor(buffer,size), body);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/BodyHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/BodyHandler.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/BodyHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/BodyHandler.cpp Fri Oct 26 12:48:31 2007
@@ -18,14 +18,13 @@
* under the License.
*
*/
-#include "qpid/QpidError.h"
#include "BodyHandler.h"
#include "AMQMethodBody.h"
#include "AMQHeaderBody.h"
#include "AMQContentBody.h"
#include "AMQHeartbeatBody.h"
-
#include <boost/cast.hpp>
+#include "qpid/framing/reply_exceptions.h"
using namespace qpid::framing;
using namespace boost;
@@ -49,7 +48,8 @@
handleHeartbeat(polymorphic_downcast<AMQHeartbeatBody*>(body));
break;
default:
- QPID_ERROR(PROTOCOL_ERROR, "Unknown frame type "+body->type());
+ throw SyntaxErrorException(
+ QPID_MSG("Invalid frame type " << body->type()));
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelAdapter.cpp Fri Oct 26 12:48:31 2007
@@ -15,8 +15,6 @@
* limitations under the License.
*
*/
-#include <boost/format.hpp>
-
#include "ChannelAdapter.h"
#include "OutputHandler.h"
#include "AMQFrame.h"
@@ -26,8 +24,6 @@
#include "AMQMethodBody.h"
#include "qpid/framing/ConnectionOpenBody.h"
-using boost::format;
-
namespace qpid {
namespace framing {
@@ -53,20 +49,20 @@
void ChannelAdapter::assertMethodOk(AMQMethodBody& method) const {
if (getId() != 0 && method.amqpClassId() == ConnectionOpenBody::CLASS_ID)
- throw ConnectionException(
- 504, format("Connection method on non-0 channel %d.")%getId());
+ throw ChannelErrorException(
+ QPID_MSG("Connection method on non-0 channel " << getId()));
}
void ChannelAdapter::assertChannelOpen() const {
if (getId() != 0 && !isOpen())
- throw ConnectionException(
- 504, format("Channel %d is not open.")%getId());
+ throw ChannelErrorException(
+ QPID_MSG("Channel " << getId() << " is not open."));
}
void ChannelAdapter::assertChannelNotOpen() const {
if (getId() != 0 && isOpen())
- throw ConnectionException(
- 504, format("Channel %d is already open.") % getId());
+ throw ChannelErrorException(
+ QPID_MSG("Channel " << getId() << " is already open."));
}
void ChannelAdapter::handle(AMQFrame& f) { handleBody(f.getBody()); }
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h Fri Oct 26 12:48:31 2007
@@ -0,0 +1,53 @@
+#ifndef QPID_FRAMING_CHANNELHANDLER_H
+#define QPID_FRAMING_CHANNELHANDLER_H
+
+/*
+ *
+ * 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 "FrameHandler.h"
+#include "AMQFrame.h"
+
+namespace qpid {
+namespace framing {
+
+/**
+ * Sets the channel number on outgoing frames.
+ */
+class ChannelHandler : public FrameHandler
+{
+ public:
+ ChannelHandler(uint16_t channelId=0, FrameHandler* next=0)
+ : FrameHandler(next), channel(channelId) {}
+ void handle(AMQFrame& frame) {
+ frame.setChannel(channel);
+ next->handle(frame);
+ }
+ uint16_t get() const { return channel; }
+ ChannelHandler& set(uint16_t ch) { channel=ch; return *this; }
+ operator uint16_t() const { return get(); }
+ ChannelHandler& operator=(uint16_t ch) { return set(ch); }
+
+ private:
+ uint16_t channel;
+};
+
+}} // namespace qpid::framing
+
+#endif /*!QPID_FRAMING_CHANNELHANDLER_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ChannelHandler.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldTable.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldTable.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldTable.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldTable.cpp Fri Oct 26 12:48:31 2007
@@ -19,9 +19,10 @@
*
*/
#include "FieldTable.h"
-#include "qpid/QpidError.h"
#include "Buffer.h"
#include "FieldValue.h"
+#include "qpid/Exception.h"
+#include "qpid/framing/reply_exceptions.h"
#include <assert.h>
namespace qpid {
@@ -132,7 +133,7 @@
uint32_t len = buffer.getLong();
uint32_t available = buffer.available();
if (available < len)
- THROW_QPID_ERROR(FRAMING_ERROR, "Not enough data for field table.");
+ throw SyntaxErrorException(QPID_MSG("Not enough data for field table."));
uint32_t leftover = available - len;
while(buffer.available() > leftover){
std::string name;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldValue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldValue.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldValue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FieldValue.cpp Fri Oct 26 12:48:31 2007
@@ -20,8 +20,7 @@
*/
#include "FieldValue.h"
#include "Buffer.h"
-#include "qpid/QpidError.h"
-
+#include "qpid/framing/reply_exceptions.h"
namespace qpid {
namespace framing {
@@ -75,9 +74,7 @@
data.reset(new FixedWidthValue<0>());
break;
default:
- std::stringstream out;
- out << "Unknown field table value type: " << typeOctet;
- THROW_QPID_ERROR(FRAMING_ERROR, out.str());
+ throw SyntaxErrorException(QPID_MSG("Unknown field table value type: " << (int)typeOctet));
}
data->decode(buffer);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/FramingContent.cpp Fri Oct 26 12:48:31 2007
@@ -18,12 +18,10 @@
* under the License.
*
*/
-#include <assert.h>
-
#include "Buffer.h"
#include "FramingContent.h"
-#include "qpid/QpidError.h"
-#include <sstream>
+#include "qpid/Exception.h"
+#include "qpid/framing/reply_exceptions.h"
namespace qpid {
namespace framing {
@@ -37,12 +35,12 @@
void Content::validate() {
if (discriminator == REFERENCE) {
if(value.empty()) {
- THROW_QPID_ERROR(FRAMING_ERROR, "Reference cannot be empty");
+ throw InvalidArgumentException(
+ QPID_MSG("Reference cannot be empty"));
}
}else if (discriminator != INLINE) {
- std::stringstream out;
- out << "Invalid discriminator: " << (int) discriminator;
- THROW_QPID_ERROR(FRAMING_ERROR, out.str());
+ throw SyntaxErrorException(
+ QPID_MSG("Invalid discriminator: " << discriminator));
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Handler.h Fri Oct 26 12:48:31 2007
@@ -33,6 +33,7 @@
struct Handler {
typedef T HandledType;
typedef void handleFptr(T);
+ typedef void result_type; // Compatible with std/boost functors.
Handler(Handler<T>* next_=0) : next(next_) {}
virtual ~Handler() {}
@@ -51,7 +52,7 @@
struct Chain : public Handler<T> {
Chain(Handler<T>* first=0) : Handler(first) {}
void operator=(Handler<T>* h) { next = h; }
- void handle(T t) { (*next)(t); }
+ void handle(T t) { next->handle(t); }
// TODO aconway 2007-08-29: chain modifier ops here.
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SequenceNumber.h Fri Oct 26 12:48:31 2007
@@ -47,6 +47,7 @@
bool operator<=(const SequenceNumber& other) const;
bool operator>=(const SequenceNumber& other) const;
uint32_t getValue() const { return (uint32_t) value; }
+ operator uint32_t() const { return (uint32_t) value; }
friend int32_t operator-(const SequenceNumber& a, const SequenceNumber& b);
};
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp Fri Oct 26 12:48:31 2007
@@ -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
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h Fri Oct 26 12:48:31 2007
@@ -0,0 +1,127 @@
+#ifndef QPID_FRAMING_SESSIONSTATE_H
+#define QPID_FRAMING_SESSIONSTATE_H
+
+/*
+ *
+ * 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 "qpid/framing/SequenceNumber.h"
+#include "qpid/framing/Uuid.h"
+#include "qpid/framing/AMQFrame.h"
+
+#include <boost/optional.hpp>
+
+#include <deque>
+
+namespace qpid {
+namespace framing {
+
+/**
+ * Session state common to client and broker.
+ * Implements session ack/resume protcools.
+ *
+ * A SessionState is always associated with an _open_ session (attached or
+ * suspended) it is destroyed when the session is closed.
+ *
+ * A template to make it protocol independent and easy to test.
+ */
+class SessionState
+{
+ public:
+ typedef std::vector<AMQFrame> Replay;
+
+ /** States of a session. */
+ enum State {
+ SUSPENDED, ///< Suspended, detached from any channel.
+ RESUMING, ///< Resuming: waiting for initial ack from peer.
+ ATTACHED ///< Attached to channel and operating normally.
+ };
+
+ /**
+ *Create a newly opened active session.
+ *@param ackInterval send/solicit an ack whenever N unacked frames
+ * have been received/sent.
+ *@pre ackInterval > 0
+ */
+ SessionState(uint32_t ackInterval=1, const framing::Uuid& id=framing::Uuid(true));
+
+ const framing::Uuid& getId() const { return id; }
+ State getState() const { return state; }
+
+ /** Received incoming L3 frame.
+ * @return SequenceNumber if an ack should be sent, empty otherwise.
+ * SessionState assumes that acks are sent whenever it returns
+ * a seq. number.
+ */
+ boost::optional<SequenceNumber> received(const AMQFrame&);
+
+ /** Sent outgoing L3 frame.
+ *@return true if solicit-ack should be sent. Note the SessionState
+ *assumes that a solicit-ack is sent every time it returns true.
+ */
+ bool sent(const AMQFrame&);
+
+ /** Received normal incoming ack. */
+ void receivedAck(SequenceNumber);
+
+ /** Frames to replay
+ *@pre getState()==ATTACHED
+ */
+ Replay replay();
+
+ /** Suspend the session. */
+ void suspend();
+
+ /** Start resume protocol for the session.
+ *@returns sequence number to ack immediately. */
+ SequenceNumber resuming();
+
+ /** About to send an unscheduled ack, e.g. to respond to a solicit-ack.
+ *
+ * Note: when received() returns a sequence number this function
+ * should not be called. SessionState assumes that the ack is sent
+ * every time received() returns a sequence number.
+ */
+ SequenceNumber sendingAck();
+
+ SequenceNumber getLastSent() const { return lastSent; }
+ SequenceNumber getLastReceived() const { return lastReceived; }
+ private:
+ typedef std::deque<AMQFrame> Unacked;
+
+ bool sendingSolicit();
+
+ State state;
+ framing::Uuid id;
+ Unacked unackedOut;
+ SequenceNumber lastReceived;
+ SequenceNumber lastSent;
+ uint32_t ackInterval;
+ SequenceNumber sendAckAt;
+ SequenceNumber solicitAckAt;
+ bool ackSolicited;
+ bool suspending;
+};
+
+
+}} // namespace qpid::common
+
+
+#endif /*!QPID_FRAMING_SESSIONSTATE_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SessionState.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h Fri Oct 26 12:48:31 2007
@@ -0,0 +1,89 @@
+#ifndef QPID_FRAMING_TEMPLATEVISITOR_H
+#define QPID_FRAMING_TEMPLATEVISITOR_H
+
+/*
+ * 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/mpl/fold.hpp>
+#include <boost/utility/value_init.hpp>
+
+namespace qpid {
+namespace framing {
+
+/**
+ * Metafunction to generate a visitor class derived from Base with a
+ * visit for each type in TypeList calling functor F. TypeList may be
+ * any boost::mpl type collection e.g. mpl::list.
+ *
+ * Generated class is: TemplateVisitor<Base, F, TypeList>::type
+ *
+ * @see make_visitor
+ */
+template <class VisitTemplate, class TypeList, class F>
+class TemplateVisitor
+{
+ struct Base : public VisitorBase {
+ F action;
+ Base(F f) : action(f) {}
+ using VisitorBase::visit;
+ };
+
+ template <class B, class T> struct Visit : public B {
+ Visit(F action) : B(action) {}
+ using B::visit;
+ void visit(const T& body) { action(body); }
+ };
+
+ typedef typename boost::mpl::fold<
+ TypeList, Base, Visit<boost::mpl::placeholders::_1,
+ boost::mpl::placeholders::_2>
+ >::type type;
+};
+
+/**
+ * Construct a TemplateVisitor to perform the given action,
+ * for example:
+ * @code
+ */
+template <class VisitorBase, class TypeList, class F>
+TemplateVisitor<VisitorBase,TypeList,F>::type make_visitor(F action) {
+ return TemplateVisitor<VisitorBase,TypeList,F>::type(action);
+};
+
+/**
+ * For method body classes in TypeList, invoke the corresponding function
+ * on Target and return true. For other body types return false.
+ */
+template <class TypeList, class Target>
+bool invoke(const AMQBody& body, Target& target) {
+ typename InvokeVisitor<TypeList, Target>::type v(target);
+ body.accept(v);
+ return v.target;
+}
+
+}} // namespace qpid::framing
+
+
+#endif /*!QPID_FRAMING_INVOKEVISITOR_H*/
+
+}} // namespace qpid::framing
+
+
+
+#endif /*!QPID_FRAMING_TEMPLATEVISITOR_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TemplateVisitor.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.cpp Fri Oct 26 12:48:31 2007
@@ -24,9 +24,13 @@
namespace qpid {
namespace framing {
-TransferContent::TransferContent(const std::string& _data)
-{
- setData(_data);
+TransferContent::TransferContent(const std::string& data,
+ const std::string& routingKey,
+ const std::string& exchange)
+{
+ setData(data);
+ getDeliveryProperties().setRoutingKey(routingKey);
+ getDeliveryProperties().setExchange(exchange);
}
AMQHeaderBody TransferContent::getHeader() const
@@ -73,14 +77,14 @@
const MessageProperties& TransferContent::getMessageProperties() const
{
const MessageProperties* props = header.get<MessageProperties>();
- if (!props) throw NoSuchPropertiesException();
+ if (!props) throw Exception("No message properties.");
return *props;
}
const DeliveryProperties& TransferContent::getDeliveryProperties() const
{
const DeliveryProperties* props = header.get<DeliveryProperties>();
- if (!props) throw NoSuchPropertiesException();
+ if (!props) throw Exception("No message properties.");
return *props;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/TransferContent.h Fri Oct 26 12:48:31 2007
@@ -30,14 +30,15 @@
namespace qpid {
namespace framing {
-struct NoSuchPropertiesException : public Exception {};
-
class TransferContent : public MethodContent
{
AMQHeaderBody header;
std::string data;
public:
- TransferContent(const std::string& data = "");
+ TransferContent(const std::string& data = std::string(),
+ const std::string& routingKey = std::string(),
+ const std::string& exchange = std::string());
+
AMQHeaderBody getHeader() const;
void setData(const std::string&);
void appendData(const std::string&);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.cpp Fri Oct 26 12:48:31 2007
@@ -17,9 +17,9 @@
*/
#include "Uuid.h"
-
-#include "qpid/QpidError.h"
+#include "qpid/Exception.h"
#include "qpid/framing/Buffer.h"
+#include "qpid/framing/reply_exceptions.h"
namespace qpid {
namespace framing {
@@ -34,7 +34,7 @@
void Uuid::decode(Buffer& buf) {
if (buf.available() < size())
- THROW_QPID_ERROR(FRAMING_ERROR, "Not enough data for UUID.");
+ throw SyntaxErrorException(QPID_MSG("Not enough data for UUID."));
buf.getRawData(c_array(), size());
}
@@ -50,6 +50,12 @@
if (uuid_parse(unparsed, uuid.c_array()) != 0)
in.setstate(ios::failbit);
return in;
+}
+
+std::string Uuid::str() const {
+ std::ostringstream os;
+ os << *this;
+ return os.str();
}
}} // namespace qpid::framing
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Uuid.h Fri Oct 26 12:48:31 2007
@@ -62,6 +62,9 @@
void encode(framing::Buffer& buf) const;
void decode(framing::Buffer& buf);
+
+ /** String value in format 1b4e28ba-2fa1-11d2-883f-b9a761bde3fb */
+ std::string str() const;
};
/** Print in format 1b4e28ba-2fa1-11d2-883f-b9a761bde3fb */
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_framing.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_framing.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_framing.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_framing.h Fri Oct 26 12:48:31 2007
@@ -32,4 +32,3 @@
#include "ProtocolInitiation.h"
#include "BasicHeaderProperties.h"
#include "ProtocolVersion.h"
-#include "ProtocolVersionException.h"
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types.h Fri Oct 26 12:48:31 2007
@@ -61,5 +61,11 @@
const ChannelId CHANNEL_MAX=(ChannelId(~1))>>1;
const ChannelId CHANNEL_HIGH_BIT= ChannelId(~CHANNEL_MAX);
+// Forward declare class types
+class FramingContent;
+class FieldTable;
+class SequenceNumberSet;
+class Uuid;
+
}} // namespace qpid::framing
#endif
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/variant.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/variant.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/variant.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/variant.h Fri Oct 26 12:48:31 2007
@@ -23,7 +23,6 @@
/**@file Tools for using boost::variant */
-#include "qpid/QpidError.h"
#include <boost/variant.hpp>
@@ -39,7 +38,7 @@
struct NoBlankVisitor : public boost::static_visitor<R> {
R foundBlank() const {
assert(0);
- THROW_QPID_ERROR(INTERNAL_ERROR, "Invalid variant value.");
+ throw Exception(QPID_MSG("Invalid variant value."));
}
R operator()(const boost::blank&) const { return foundBlank(); }
R operator()(boost::blank&) const { return foundBlank(); }
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.cpp Fri Oct 26 12:48:31 2007
@@ -190,12 +190,6 @@
statements.insert(&s);
}
-void Logger::remove(Statement& s) {
- ScopedLock l(lock);
- s.enabled = false;
- statements.erase(&s);
-}
-
void Logger::configure(const Options& opts, const std::string& prog)
{
clear();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/log/Logger.h Fri Oct 26 12:48:31 2007
@@ -67,9 +67,6 @@
/** Add a statement. */
void add(Statement& s);
- /** Remove a statement */
- void remove(Statement& s);
-
/** Log a message. */
void log(const Statement&, const std::string&);
@@ -93,7 +90,7 @@
/** Add an output destination for messages */
void output(std::auto_ptr<Output> out);
- /** Reset the logger to it's original state */
+ /** Reset the logger to it's original state. */
void clear();
private:
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.cpp Fri Oct 26 12:48:31 2007
@@ -32,10 +32,6 @@
Logger::instance().add(s);
}
-Statement::Initializer::~Initializer() {
- Logger::instance().remove(statement);
-}
-
namespace {
const char* names[LevelTraits::COUNT] = {
"trace", "debug", "info", "notice", "warning", "error", "critical"
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/log/Statement.h Fri Oct 26 12:48:31 2007
@@ -19,8 +19,9 @@
*
*/
+#include "qpid/Msg.h"
+
#include <boost/current_function.hpp>
-#include <sstream>
namespace qpid {
namespace log {
@@ -69,23 +70,14 @@
struct Initializer {
Initializer(Statement& s);
- ~Initializer();
Statement& statement;
};
};
-///@internal trickery to make QPID_LOG_STRINGSTREAM work.
-inline std::ostream& noop(std::ostream& s) { return s; }
-
///@internal static initializer for a Statement.
#define QPID_LOG_STATEMENT_INIT(level) \
{ 0, __FILE__, __LINE__, BOOST_CURRENT_FUNCTION, (::qpid::log::level) }
-///@internal Stream streamable message and return a string.
-#define QPID_LOG_STRINGSTREAM(message) \
- static_cast<std::ostringstream&>( \
- std::ostringstream() << qpid::log::noop << message).str()
-
/**
* Macro for log statements. Example of use:
* @code
@@ -110,19 +102,9 @@
static ::qpid::log::Statement stmt_= QPID_LOG_STATEMENT_INIT(level); \
static ::qpid::log::Statement::Initializer init_(stmt_); \
if (stmt_.enabled) \
- stmt_.log(QPID_LOG_STRINGSTREAM(message)); \
+ stmt_.log(::qpid::Msg() << message); \
} while(0)
-/**
- * Macro for complicated logging logic that can't fit in a simple QPID_LOG
- * statement. For example:
- * @code
- * QPID_IF_LOG(debug) {
- * message = do_complicated_stuff;
- * QPID_LOG(debug, message);
- * }
- */
-#define QPID_IF_LOG(level)
}} // namespace qpid::log
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/AsynchIOAcceptor.cpp Fri Oct 26 12:48:31 2007
@@ -29,6 +29,7 @@
#include "qpid/sys/ConnectionOutputHandler.h"
#include "qpid/sys/ConnectionInputHandler.h"
#include "qpid/sys/ConnectionInputHandlerFactory.h"
+#include "qpid/framing/reply_exceptions.h"
#include "qpid/framing/Buffer.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/log/Statement.h"
@@ -301,7 +302,7 @@
}
// If frame was egregiously large complain
if (frameSize > buff->byteCount)
- THROW_QPID_ERROR(FRAMING_ERROR, "Could not write frame, too large for buffer.");
+ throw framing::ContentTooLargeException(QPID_MSG("Could not write frame, too large for buffer."));
buff->dataCount = buffUsed;
aio->queueWrite(buff);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConcurrentQueue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConcurrentQueue.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConcurrentQueue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ConcurrentQueue.h Fri Oct 26 12:48:31 2007
@@ -22,7 +22,7 @@
*
*/
-#include "qpid/sys/Monitor.h"
+#include "qpid/sys/Waitable.h"
#include "qpid/sys/ScopedIncrement.h"
#include <boost/bind.hpp>
@@ -39,73 +39,73 @@
*
* Also allows consuming threads to wait until an item is available.
*/
-template <class T> class ConcurrentQueue {
+template <class T> class ConcurrentQueue : public Waitable {
public:
- ConcurrentQueue() : waiters(0), shutdown(false) {}
+ struct ShutdownException {};
+
+ ConcurrentQueue() : shutdownFlag(false) {}
- /** Threads in wait() are woken with ShutdownException before
- * destroying the queue.
- */
- ~ConcurrentQueue() {
- Mutex::ScopedLock l(lock);
- shutdown = true;
- lock.notifyAll();
- while (waiters > 0)
- lock.wait();
+ /** Waiting threads are notified by ~Waitable */
+ ~ConcurrentQueue() { shutdown(); }
+
+ bool shutdown(bool wait=true) {
+ ScopedLock l(lock);
+ if (!shutdownFlag) {
+ shutdownFlag=true;
+ lock.notifyAll();
+ if (wait) lock.waitAll();
+ shutdownFlag=true;
+ return true;
+ }
+ return false;
}
-
+
/** Push a data item onto the back of the queue */
void push(const T& data) {
Mutex::ScopedLock l(lock);
queue.push_back(data);
+ lock.notify();
}
/** If the queue is non-empty, pop the front item into data and
* return true. If the queue is empty, return false
*/
- bool pop(T& data) {
+ bool tryPop(T& data) {
Mutex::ScopedLock l(lock);
- return popInternal(data);
+ if (shutdownFlag || queue.empty())
+ return false;
+ data = queue.front();
+ queue.pop_front();
+ return true;
}
- /** Wait up to deadline for a data item to be available.
- *@return true if data was available, false if timed out.
+ /** Wait up to a timeout for a data item to be available.
+ *@return true if data was available, false if timed out or shut down.
*@throws ShutdownException if the queue is destroyed.
*/
- bool waitPop(T& data, Duration timeout) {
- Mutex::ScopedLock l(lock);
- ScopedIncrement<size_t> w(
- waiters, boost::bind(&ConcurrentQueue::noWaiters, this));
+ bool waitPop(T& data, Duration timeout=TIME_INFINITE) {
+ ScopedLock l(lock);
AbsTime deadline(now(), timeout);
- while (queue.empty() && lock.wait(deadline))
- ;
- return popInternal(data);
- }
-
- private:
-
- bool popInternal(T& data) {
- if (shutdown)
- throw ShutdownException();
+ {
+ ScopedWait(*this);
+ while (!shutdownFlag && queue.empty())
+ if (!lock.wait(deadline))
+ return false;
+ }
if (queue.empty())
return false;
- else {
- data = queue.front();
- queue.pop_front();
- return true;
- }
+ data = queue.front();
+ queue.pop_front();
+ return true;
}
+
+ bool isShutdown() { ScopedLock l(lock); return shutdownFlag; }
- void noWaiters() {
- assert(waiters == 0);
- if (shutdown)
- lock.notify(); // Notify dtor thread.
- }
-
- Monitor lock;
+ protected:
+ Waitable lock;
+ private:
std::deque<T> queue;
- size_t waiters;
- bool shutdown;
+ bool shutdownFlag;
};
}} // namespace qpid::sys
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ScopedIncrement.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ScopedIncrement.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ScopedIncrement.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/ScopedIncrement.h Fri Oct 26 12:48:31 2007
@@ -30,17 +30,17 @@
* Optionally call a function if the decremented counter value is 0.
* Note the function must not throw, it is called in the destructor.
*/
-template <class T>
+template <class T, class F=boost::function<void()> >
class ScopedIncrement : boost::noncopyable
{
public:
- ScopedIncrement(T& c, boost::function0<void> f=0)
+ ScopedIncrement(T& c, F f=0)
: count(c), callback(f) { ++count; }
~ScopedIncrement() { if (--count == 0 && callback) callback(); }
private:
T& count;
- boost::function0<void> callback;
+ F callback;
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h Fri Oct 26 12:48:31 2007
@@ -23,7 +23,7 @@
*
*/
-
+#include "qpid/Exception.h"
#include "qpid/sys/Runnable.h"
#include "qpid/sys/Monitor.h"
#include "qpid/sys/Thread.h"
@@ -41,6 +41,8 @@
{
public:
typedef boost::function<void()> VoidFn0;
+ struct ShutdownException : public Exception {};
+
/** @see Serializer::Serializer */
SerializerBase(bool immediate=true, VoidFn0 notifyDispatch=VoidFn0());
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h Fri Oct 26 12:48:31 2007
@@ -0,0 +1,78 @@
+#ifndef QPID_SYS_STATEMONITOR_H
+#define QPID_SYS_STATEMONITOR_H
+
+/*
+ *
+ * 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 "qpid/sys/Waitable.h"
+
+#include <bitset>
+
+namespace qpid {
+namespace sys {
+
+/**
+ * A monitor with an enum state value.
+ *
+ *@param Enum: enum type to use for states.
+ *@param EnumMax: Highest enum value.
+ */
+template <class Enum, size_t MaxEnum>
+class StateMonitor : public Waitable
+{
+ public:
+ struct Set : public std::bitset<MaxEnum + 1> {
+ Set() {}
+ Set(Enum s) { set(s); }
+ Set(Enum s, Enum t) { set(s).set(t); }
+ Set(Enum s, Enum t, Enum u) { set(s).set(t).set(u); }
+ Set(Enum s, Enum t, Enum u, Enum v) { set(s).set(t).set(u).set(v); }
+ };
+
+
+ StateMonitor(Enum initial) { state=initial; }
+
+ /** @pre Caller holds a ScopedLock. */
+ void set(Enum s) { state=s; notifyAll(); }
+ /** @pre Caller holds a ScopedLock. */
+ StateMonitor& operator=(Enum s) { set(s); return *this; }
+
+ /** @pre Caller holds a ScopedLock. */
+ Enum get() const { return state; }
+ /** @pre Caller holds a ScopedLock. */
+ operator Enum() const { return state; }
+
+ /** @pre Caller holds a ScopedLock */
+ void waitFor(Enum s) { ScopedWait(*this); while (s != state) wait(); }
+ /** @pre Caller holds a ScopedLock */
+ void waitFor(Set s) { ScopedWait(*this); while (!s.test(state)) wait(); }
+ /** @pre Caller holds a ScopedLock */
+ void waitNot(Enum s) { ScopedWait(*this); while (s == state) wait(); }
+ /** @pre Caller holds a ScopedLock */
+ void waitNot(Set s) { ScopedWait(*this); while (s.test(state)) wait(); }
+
+ private:
+ Enum state;
+};
+
+}}
+
+
+#endif /*!QPID_SYS_STATEMONITOR_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/StateMonitor.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h?rev=588761&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h Fri Oct 26 12:48:31 2007
@@ -0,0 +1,73 @@
+#ifndef QPID_SYS_WAITABLE_H
+#define QPID_SYS_WAITABLE_H
+
+/*
+ * 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 "Monitor.h"
+
+#include <assert.h>
+
+namespace qpid {
+namespace sys {
+
+/**
+ * A monitor that keeps track of waiting threads.
+ * Threads that use a WaitLock are counted as waiters, threads that
+ * use a normal ScopedLock are not considered waiters.
+ */
+class Waitable : public Monitor {
+ public:
+ Waitable() : waiters(0) {}
+
+ /** Use this inside a scoped lock around the
+ * call to Monitor::wait to be counted as a waiter
+ */
+ struct ScopedWait {
+ Waitable& w;
+ ScopedWait(Waitable& w_) : w(w_) { ++w.waiters; }
+ ~ScopedWait() { --w.waiters; w.notifyAll(); }
+ };
+
+ /** Block till all waiters have finished waiting.
+ * The calling thread does not count as a waiter.
+ *@pre Must be called inside a ScopedLock but NOT a ScopedWait.
+ */
+ bool waitAll(Duration timeout=TIME_INFINITE) {
+ AbsTime deadline(now(), timeout);
+ while (waiters > 0) {
+ if (!wait(deadline)) {
+ assert(timeout != TIME_INFINITE);
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private:
+ friend struct ScopedWait;
+ size_t waiters;
+};
+
+}} // namespace qpid::sys
+
+
+
+#endif /*!QPID_SYS_WAITABLE_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Waitable.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.cpp Fri Oct 26 12:48:31 2007
@@ -20,7 +20,6 @@
*/
#include <iostream>
#include "qpid/log/Statement.h"
-#include "qpid/QpidError.h"
#include "APRBase.h"
using namespace qpid::sys;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/APRBase.h Fri Oct 26 12:48:31 2007
@@ -24,7 +24,6 @@
#include <string>
#include <apr_thread_mutex.h>
#include <apr_errno.h>
-#include "qpid/QpidError.h"
namespace qpid {
namespace sys {
@@ -64,11 +63,8 @@
// Inlined as it is called *a lot*
void inline qpid::sys::check(apr_status_t status, const char* file, const int line){
if (status != APR_SUCCESS){
- const int size = 50;
- char tmp[size];
- std::string msg(apr_strerror(status, tmp, size));
- throw qpid::QpidError(APR_ERROR + ((int) status), msg,
- qpid::SrcLine(file, line));
+ char tmp[256];
+ throw Exception(QPID_MSG(apr_strerror(status, tmp, size)))
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Shlib.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Shlib.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Shlib.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Shlib.cpp Fri Oct 26 12:48:31 2007
@@ -19,8 +19,7 @@
*/
#include "qpid/sys/Shlib.h"
-
-#include <qpid/QpidError.h>
+#include "qpid/Exception.h"
#include <dlfcn.h>
@@ -32,7 +31,7 @@
handle = ::dlopen(name, RTLD_NOW);
const char* error = ::dlerror();
if (error) {
- THROW_QPID_ERROR(INTERNAL_ERROR, error);
+ throw Exception(QPID_MSG(error));
}
}
@@ -42,7 +41,7 @@
::dlclose(handle);
const char* error = ::dlerror();
if (error) {
- THROW_QPID_ERROR(INTERNAL_ERROR, error);
+ throw Exception(QPID_MSG(error));
}
handle = 0;
}
@@ -53,7 +52,7 @@
void* sym = ::dlsym(handle, name);
const char* error = ::dlerror();
if (error)
- THROW_QPID_ERROR(INTERNAL_ERROR, error);
+ throw Exception(QPID_MSG(error));
return sym;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Socket.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Socket.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Socket.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Socket.cpp Fri Oct 26 12:48:31 2007
@@ -21,7 +21,6 @@
#include "qpid/sys/Socket.h"
-#include "qpid/QpidError.h"
#include "check.h"
#include "PrivatePosix.h"
@@ -60,8 +59,7 @@
result = ::getpeername(fd, (::sockaddr*)&name, &namelen);
}
- if (result < 0)
- throw QPID_POSIX_ERROR(errno);
+ QPID_POSIX_CHECK(result);
char servName[NI_MAXSERV];
char dispName[NI_MAXHOST];
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/check.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/check.h?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/check.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/check.h Fri Oct 26 12:48:31 2007
@@ -22,41 +22,13 @@
*
*/
-#include <cerrno>
-#include <string>
-#include "qpid/QpidError.h"
-
-namespace qpid {
-namespace sys {
-
-/**
- * Exception with message from errno.
- */
-class PosixError : public qpid::QpidError
-{
- public:
- static std::string getMessage(int errNo);
-
- PosixError(int errNo, const qpid::SrcLine& location) throw();
-
- ~PosixError() throw() {}
-
- int getErrNo() { return errNo; }
+#include "qpid/Exception.h"
- Exception::auto_ptr clone() const throw() { return Exception::auto_ptr(new PosixError(*this)); }
-
- void throwSelf() const { throw *this; }
-
- private:
- int errNo;
-};
-
-}}
+#include <cerrno>
-/** Create a PosixError for the current file/line and errno. */
-#define QPID_POSIX_ERROR(errNo) ::qpid::sys::PosixError(errNo, SRCLINE)
+#define QPID_POSIX_ERROR(ERRNO) qpid::Exception(QPID_MSG(qpid::strError(ERRNO)) << " " << ERRNO)
-/** Throw QPID_POSIX_ERROR(errno) if RESULT is less than zero */
+/** THROW QPID_POSIX_ERROR(errno) if RESULT is less than zero */
#define QPID_POSIX_CHECK(RESULT) \
if ((RESULT) < 0) throw QPID_POSIX_ERROR((errno))
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ClientChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ClientChannelTest.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ClientChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ClientChannelTest.cpp Fri Oct 26 12:48:31 2007
@@ -56,7 +56,7 @@
}
};
- InProcessBrokerClient connection; // client::connection + local broker
+ qpid::InProcessBrokerClient connection;
const std::string qname;
const std::string data;
Queue queue;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Fri Oct 26 12:48:31 2007
@@ -18,15 +18,21 @@
* under the License.
*
*/
-#include <list>
#include "qpid_test_plugin.h"
#include "InProcessBroker.h"
#include "qpid/client/Dispatcher.h"
#include "qpid/client/Session.h"
#include "qpid/framing/TransferContent.h"
+#include "qpid/framing/reply_exceptions.h"
+
+#include <boost/optional.hpp>
+
+#include <list>
using namespace qpid::client;
using namespace qpid::framing;
+using namespace qpid;
+using namespace boost;
struct DummyListener : public MessageListener
{
@@ -60,58 +66,77 @@
CPPUNIT_TEST(testQueueQuery);
CPPUNIT_TEST(testTransfer);
CPPUNIT_TEST(testDispatcher);
+ CPPUNIT_TEST(testResumeExpiredError);
+ CPPUNIT_TEST(testUseSuspendedError);
CPPUNIT_TEST(testSuspendResume);
- CPPUNIT_TEST(testSuspendResumeErrors);
+ CPPUNIT_TEST(testDisconnectResume);
+ CPPUNIT_TEST(testAutoDelete);
CPPUNIT_TEST_SUITE_END();
- boost::shared_ptr<Connector> broker;
- Connection connection;
+ shared_ptr<broker::Broker> broker;
Session session;
+ // Defer construction & thread creation to setUp
+ boost::optional<InProcessConnection> c;
+ boost::optional<InProcessConnection> c2;
public:
- ClientSessionTest() : broker(new qpid::broker::InProcessBroker()), connection(broker)
+ void setUp() {
+ broker = broker::Broker::create();
+ c=boost::in_place<InProcessConnection>(broker);
+ c2=boost::in_place<InProcessConnection>(broker);
+ }
+
+ void tearDown() {
+ c2.reset();
+ c.reset();
+ broker.reset();
+ }
+
+ void declareSubscribe(const std::string& q="my-queue",
+ const std::string& dest="my-dest")
{
- connection.open("");
- session = connection.newSession();
+ // FIXME aconway 2007-10-18: autoDelete queues are destroyed on channel close, not session.
+ // Fix & make all test queues exclusive, autoDelete
+ session.queueDeclare_(queue=q); // FIXME aconway 2007-10-01: exclusive=true, autoDelete=true);
+ session.messageSubscribe_(queue=q, destination=dest, acquireMode=1);
+ session.messageFlow_(destination=dest, unit=0, value=0xFFFFFFFF);//messages
+ session.messageFlow_(destination=dest, unit=1, value=0xFFFFFFFF);//bytes
}
+ bool queueExists(const std::string& q) {
+ TypedResult<QueueQueryResult> result = session.queueQuery_(q);
+ return result.get().getQueue() == q;
+ }
+
void testQueueQuery()
{
- std::string name("my-queue");
- std::string alternate("amq.fanout");
- session.queueDeclare((queue=name, alternateExchange=alternate, exclusive=true, autoDelete=true));
- TypedResult<QueueQueryResult> result = session.queueQuery(name);
+ session = c->newSession();
+ session.queueDeclare_(queue="my-queue", alternateExchange="amq.fanout", exclusive=true, autoDelete=true);
+ TypedResult<QueueQueryResult> result = session.queueQuery_(std::string("my-queue"));
CPPUNIT_ASSERT_EQUAL(false, result.get().getDurable());
CPPUNIT_ASSERT_EQUAL(true, result.get().getExclusive());
- CPPUNIT_ASSERT_EQUAL(alternate, result.get().getAlternateExchange());
+ CPPUNIT_ASSERT_EQUAL(std::string("amq.fanout"),
+ result.get().getAlternateExchange());
}
void testTransfer()
{
- std::string queueName("my-queue");
- std::string dest("my-dest");
- std::string data("my message");
- session.queueDeclare_(queue=queueName, exclusive=true, autoDelete=true);
- //subcribe to the queue with confirm_mode = 1:
- session.messageSubscribe_(queue=queueName, destination=dest, acquireMode=1);
- session.messageFlow((destination=dest, unit=0, value=1));//messages
- session.messageFlow((destination=dest, unit=1, value=0xFFFFFFFF));//bytes
- //publish a message:
- TransferContent _content(data);
- _content.getDeliveryProperties().setRoutingKey("my-queue");
- session.messageTransfer_(content=_content);
+ session = c->newSession();
+ declareSubscribe();
+ session.messageTransfer_(content=TransferContent("my-message", "my-queue"));
//get & test the message:
FrameSet::shared_ptr msg = session.get();
CPPUNIT_ASSERT(msg->isA<MessageTransferBody>());
- CPPUNIT_ASSERT_EQUAL(data, msg->getContent());
+ CPPUNIT_ASSERT_EQUAL(std::string("my-message"), msg->getContent());
//confirm receipt:
session.execution().completed(msg->getId(), true, true);
}
void testDispatcher()
{
- session.queueDeclare_(queue="my-queue", exclusive=true, autoDelete=true);
+ session = c->newSession();
+ declareSubscribe();
TransferContent msg1("One");
msg1.getDeliveryProperties().setRoutingKey("my-queue");
@@ -125,9 +150,6 @@
msg3.getDeliveryProperties().setRoutingKey("my-queue");
session.messageTransfer_(content=msg3);
- session.messageSubscribe_(queue="my-queue", destination="my-dest", acquireMode=1);
- session.messageFlow((destination="my-dest", unit=0, value=1));//messages
- session.messageFlow((destination="my-dest", unit=1, value=0xFFFFFFFF));//bytes
DummyListener listener(session, "my-dest", 3);
listener.listen();
CPPUNIT_ASSERT_EQUAL((size_t) 3, listener.messages.size());
@@ -140,29 +162,66 @@
}
- void testSuspendResume() {
- session = connection.newSession(60);
+ void testResumeExpiredError() {
+ session = c->newSession(0);
+ session.suspend(); // session has 0 timeout.
+ try {
+ c->resume(session);
+ CPPUNIT_FAIL("Expected InvalidArgumentException.");
+ } catch(const InvalidArgumentException&) {}
+ }
+
+ void testUseSuspendedError() {
+ session = c->newSession(60);
session.suspend();
try {
session.exchangeQuery_(name="amq.fanout");
CPPUNIT_FAIL("Expected session suspended exception");
- } catch(...) {}
- connection.resume(session);
- session.exchangeQuery_(name="amq.fanout");
- // FIXME aconway 2007-09-25: build up session state and confirm
- //it survives the resume
+ } catch(const CommandInvalidException&) {}
}
- void testSuspendResumeErrors() {
- session.suspend(); // session has 0 timeout.
- try {
- session.exchangeQuery_(name="amq.fanout");
- CPPUNIT_FAIL("Expected suspended session exception");
- } catch(...) {}
- try {
- connection.resume(session);
- CPPUNIT_FAIL("Expected no such session exception.");
- } catch(...) {}
+ void testSuspendResume() {
+ session = c->newSession(60);
+ declareSubscribe();
+ session.suspend();
+ // Make sure we are still subscribed after resume.
+ c->resume(session);
+ session.messageTransfer_(content=TransferContent("my-message", "my-queue"));
+ FrameSet::shared_ptr msg = session.get();
+ CPPUNIT_ASSERT_EQUAL(string("my-message"), msg->getContent());
+ }
+
+ void testDisconnectResume() {
+ session = c->newSession(60);
+ session.queueDeclare_(queue="before");
+ CPPUNIT_ASSERT(queueExists("before"));
+ // Simulate lost frames.
+ c->discard();
+ session.queueDeclare_(queue=string("after"));
+ c->disconnect(); // Simulate disconnect, resume on a new connection.
+ c2->resume(session);
+ CPPUNIT_ASSERT(queueExists("after"));
+ }
+
+ void testAutoDelete() {
+ // Verify that autoDelete queues survive suspend/resume.
+ session = c->newSession(60);
+ session.queueDeclare_(queue="my-queue", exclusive=true, autoDelete=true);
+ CPPUNIT_ASSERT(queueExists("my-queue"));
+ session.suspend();
+ c->resume(session);
+ CPPUNIT_ASSERT(queueExists("my-queue"));
+
+ // Verify they survive disconnect/resume on new Connection
+ c->disconnect();
+ c2->resume(session);
+
+ try {
+ // FIXME aconway 2007-10-23: Negative test, need to
+ // fix auto-delete queues to clean up with session, not channel.
+ CPPUNIT_ASSERT(queueExists("my-queue"));
+ CPPUNIT_FAIL("Negative test passed unexpectedly");
+ } catch(const ChannelException&) {}
}
};
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ConcurrentQueue.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ConcurrentQueue.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ConcurrentQueue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ConcurrentQueue.cpp Fri Oct 26 12:48:31 2007
@@ -61,7 +61,7 @@
/** If the queue is non-empty, pop the front item into data and
* return true. If the queue is empty, return false
*/
- bool pop(T& data) {
+ bool tryPop(T& data) {
Mutex::ScopedLock l(popLock);
if (popIter == popVec.end()) {
popVec.clear();
@@ -109,7 +109,7 @@
struct NullQueue {
NullQueue(int items=0) : npush(items), npop(items) {}
void push(int) { --npush; }
- bool pop(int& n) {
+ bool tryPop(int& n) {
if (npop == 0)
return false;
else {
@@ -144,7 +144,7 @@
void run() {
for (int i=items; i > 0; i--) {
int n;
- if (queue.pop(n))
+ if (queue.tryPop(n))
BOOST_REQUIRE_EQUAL(i,n);
npause();
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/EventChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/EventChannelTest.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/EventChannelTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/EventChannelTest.cpp Fri Oct 26 12:48:31 2007
@@ -117,18 +117,18 @@
CPPUNIT_ASSERT(re.hasError());
try {
re.throwIfError();
- CPPUNIT_FAIL("Expected QpidError.");
+ CPPUNIT_FAIL("Expected Exception.");
}
- catch (const qpid::QpidError&) { }
+ catch (const qpid::Exception&) { }
// Bad file descriptor. Note in this case we fail
// in postEvent and throw immediately.
try {
ReadEvent bad;
ec->postEvent(bad);
- CPPUNIT_FAIL("Expected QpidError.");
+ CPPUNIT_FAIL("Expected Exception.");
}
- catch (const qpid::QpidError&) { }
+ catch (const qpid::Exception&) { }
}
void testWrite() {
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp?rev=588761&r1=588760&r2=588761&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/FramingTest.cpp Fri Oct 26 12:48:31 2007
@@ -18,8 +18,6 @@
* under the License.
*
*/
-#include "InProcessBroker.h"
-#include "qpid/QpidError.h"
#include "qpid/client/Exchange.h"
#include "qpid/client/Queue.h"
#include "qpid/client/Connection.h"
@@ -30,6 +28,7 @@
#include "qpid/framing/ProtocolVersion.h"
#include "qpid/framing/all_method_bodies.h"
#include "qpid/framing/amqp_framing.h"
+#include "qpid/framing/reply_exceptions.h"
#include "qpid_test_plugin.h"
#include <boost/bind.hpp>
@@ -200,18 +199,12 @@
try {
Content content(REFERENCE, "");
CPPUNIT_ASSERT(false);//fail, expected exception
- } catch (QpidError& e) {
- CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
- CPPUNIT_ASSERT_EQUAL(string("Reference cannot be empty"), e.msg);
- }
+ } catch (const InvalidArgumentException& e) {}
try {
Content content(2, "Blah");
CPPUNIT_ASSERT(false);//fail, expected exception
- } catch (QpidError& e) {
- CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
- CPPUNIT_ASSERT_EQUAL(string("Invalid discriminator: 2"), e.msg);
- }
+ } catch (const SyntaxErrorException& e) {}
try {
Buffer wbuff(buffer, sizeof(buffer));
@@ -221,11 +214,8 @@
Buffer rbuff(buffer, sizeof(buffer));
Content content;
content.decode(rbuff);
- CPPUNIT_ASSERT(false);//fail, expected exception
- } catch (QpidError& e) {
- CPPUNIT_ASSERT_EQUAL(FRAMING_ERROR, e.code);
- CPPUNIT_ASSERT_EQUAL(string("Invalid discriminator: 2"), e.msg);
- }
+ CPPUNIT_FAIL("Expected exception");
+ } catch (Exception& e) {}
}