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 2006/12/20 18:54:07 UTC
svn commit: r489159 [2/2] - in
/incubator/qpid/branches/event-queue-2006-12-20/cpp: ./ lib/broker/
lib/client/ lib/common/ lib/common/framing/ lib/common/sys/
lib/common/sys/apr/ lib/common/sys/posix/ tests/
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.cpp Wed Dec 20 09:54:05 2006
@@ -32,8 +32,8 @@
return std::string(strerror_r(errNo, buf, sizeof(buf)));
}
-PosixError::PosixError(int errNo, const qpid::SrcLine& loc) throw()
- : qpid::QpidError(INTERNAL_ERROR + errNo, getMessage(errNo), loc)
+PosixError::PosixError(int errNo, const qpid::QpidError::Location& l) throw()
+ : qpid::QpidError(INTERNAL_ERROR + errNo, getMessage(errNo), l)
{ }
}}
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.h
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.h?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.h (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/lib/common/sys/posix/check.h Wed Dec 20 09:54:05 2006
@@ -37,13 +37,15 @@
public:
static std::string getMessage(int errNo);
- PosixError(int errNo, const qpid::SrcLine& location) throw();
+ PosixError(int errNo, const qpid::QpidError::Location& location) throw();
~PosixError() throw() {}
int getErrNo() { return errNo; }
- Exception* clone() const throw() { return new PosixError(*this); }
+ Exception::auto_ptr clone() const throw() {
+ return Exception::auto_ptr(new PosixError(*this));
+ }
void throwSelf() { throw *this; }
@@ -54,9 +56,17 @@
}}
/** 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::sys::PosixError((ERRNO), QPID_ERROR_LOCATION)
-/** Throw a posix error if errNo is non-zero */
-#define QPID_POSIX_THROW_IF(ERRNO) \
- if ((ERRNO) != 0) throw QPID_POSIX_ERROR((ERRNO))
+/** Throw QPID_POSIX_ERROR(errno) if RESULT is less than zero */
+#define QPID_POSIX_CHECK(RESULT) \
+ if ((RESULT) < 0) throw QPID_POSIX_ERROR((errno))
+
+/** Throw QPID_POSIX_ERROR(ERRNO) if ERRNO is non zero */
+#define QPID_POSIX_THROW_IF(ERRNO) \
+ do { int e = (ERRNO); if (e) throw QPID_POSIX_ERROR(e); } while(0)
+
+
+
#endif /*!_posix_check_h*/
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelTest.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelTest.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelTest.cpp Wed Dec 20 09:54:05 2006
@@ -47,8 +47,9 @@
class EventChannelTest : public CppUnit::TestCase
{
CPPUNIT_TEST_SUITE(EventChannelTest);
- CPPUNIT_TEST(testEvent);
+ CPPUNIT_TEST(testDispatch);
CPPUNIT_TEST(testRead);
+ CPPUNIT_TEST(testPartialRead);
CPPUNIT_TEST(testFailedRead);
CPPUNIT_TEST(testWrite);
CPPUNIT_TEST(testFailedWrite);
@@ -72,26 +73,26 @@
signal(SIGPIPE, SIG_IGN);
}
- // Verify that calling getEvent returns event.
+ // Verify that calling wait returns event.
template <class T> bool isNextEvent(T& event)
{
- return &event == dynamic_cast<T*>(ec->getEvent());
+ return &event == dynamic_cast<T*>(ec->wait(5*TIME_SEC));
}
template <class T> bool isNextEventOk(T& event)
{
- Event* next = ec->getEvent();
- if (next) next->throwIfError();
+ Event* next = ec->wait(TIME_SEC);
+ if (next) next->throwIfException();
return &event == next;
}
- void testEvent()
+ void testDispatch()
{
RunMe runMe;
CPPUNIT_ASSERT(!runMe.ran);
// Instances of Event just pass thru the channel immediately.
- Event e(runMe.functor());
- ec->postEvent(e);
+ DispatchEvent e(runMe.functor());
+ ec->post(e);
CPPUNIT_ASSERT(isNextEventOk(e));
e.dispatch();
CPPUNIT_ASSERT(runMe.ran);
@@ -99,42 +100,64 @@
void testRead() {
ReadEvent re(pipe[0], readBuf, size);
- ec->postEvent(re);
+ ec->post(re);
CPPUNIT_ASSERT_EQUAL(ssize_t(size), ::write(pipe[1], hello, size));
CPPUNIT_ASSERT(isNextEventOk(re));
- CPPUNIT_ASSERT_EQUAL(size, re.getSize());
+ CPPUNIT_ASSERT_EQUAL(size, re.getBytesRead());
CPPUNIT_ASSERT_EQUAL(std::string(hello), std::string(readBuf));
}
+ void testPartialRead() {
+ ReadEvent re(pipe[0], readBuf, size, 0, true);
+ ec->post(re);
+ CPPUNIT_ASSERT_EQUAL(ssize_t(size/2), ::write(pipe[1], hello, size/2));
+ CPPUNIT_ASSERT(isNextEventOk(re));
+ CPPUNIT_ASSERT_EQUAL(size/2, re.getBytesRead());
+ CPPUNIT_ASSERT_EQUAL(std::string(hello, size/2),
+ std::string(readBuf, size/2));
+ }
+
+
void testFailedRead()
{
ReadEvent re(pipe[0], readBuf, size);
- ec->postEvent(re);
+ ec->post(re);
// EOF before all data read.
::close(pipe[1]);
CPPUNIT_ASSERT(isNextEvent(re));
- CPPUNIT_ASSERT(re.hasError());
+ CPPUNIT_ASSERT(re.getException());
try {
- re.throwIfError();
+ re.throwIfException();
CPPUNIT_FAIL("Expected QpidError.");
}
catch (const qpid::QpidError&) { }
+
+ // Try to read from closed file descriptor.
+ try {
+ ec->post(re);
+ CPPUNIT_ASSERT(isNextEvent(re));
+ re.throwIfException();
+ CPPUNIT_FAIL("Expected an exception.");
+ }
+ catch (const qpid::QpidError&) {}
+
// Bad file descriptor. Note in this case we fail
- // in postEvent and throw immediately.
+ // in post and throw immediately.
try {
- ReadEvent bad;
- ec->postEvent(bad);
+ ReadEvent bad(-1, readBuf, size);
+ ec->post(bad);
CPPUNIT_FAIL("Expected QpidError.");
}
- catch (const qpid::QpidError&) { }
+ catch (const qpid::QpidError&) {}
}
void testWrite() {
WriteEvent wr(pipe[1], hello, size);
- ec->postEvent(wr);
+ ec->post(wr);
CPPUNIT_ASSERT(isNextEventOk(wr));
+ CPPUNIT_ASSERT_EQUAL(size, wr.getBytesWritten());
CPPUNIT_ASSERT_EQUAL(ssize_t(size), ::read(pipe[0], readBuf, size));;
CPPUNIT_ASSERT_EQUAL(std::string(hello), std::string(readBuf));
}
@@ -142,42 +165,54 @@
void testFailedWrite() {
WriteEvent wr(pipe[1], hello, size);
::close(pipe[0]);
- ec->postEvent(wr);
+ ec->post(wr);
CPPUNIT_ASSERT(isNextEvent(wr));
- CPPUNIT_ASSERT(wr.hasError());
+ CPPUNIT_ASSERT(wr.getException());
}
void testReadWrite()
{
ReadEvent re(pipe[0], readBuf, size);
WriteEvent wr(pipe[1], hello, size);
- ec->postEvent(re);
- ec->postEvent(wr);
- ec->getEvent();
- ec->getEvent();
+ ec->post(re);
+ ec->post(wr);
+ ec->wait(TIME_SEC);
+ ec->wait(TIME_SEC);
CPPUNIT_ASSERT_EQUAL(std::string(hello), std::string(readBuf));
}
- void testAccept() {
- Socket s = Socket::createTcp();
- int port = s.listen(0, 10);
- CPPUNIT_ASSERT(port != 0);
-
- AcceptEvent ae(s.fd());
- ec->postEvent(ae);
- Socket client = Socket::createTcp();
+ void connectSendRead(AcceptEvent& ae, int port, Socket client)
+ {
+ ec->post(ae);
+ // Connect a client, send some data, read the data.
client.connect("localhost", port);
CPPUNIT_ASSERT(isNextEvent(ae));
- ae.dispatch();
+ ae.throwIfException();
- // Verify client writes are read by the accepted descriptor.
char readBuf[size];
ReadEvent re(ae.getAcceptedDesscriptor(), readBuf, size);
- ec->postEvent(re);
- CPPUNIT_ASSERT_EQUAL(ssize_t(size), client.send(hello, sizeof(hello)));
+ ec->post(re);
+ CPPUNIT_ASSERT_EQUAL(ssize_t(size),
+ client.send(hello, sizeof(hello)));
CPPUNIT_ASSERT(isNextEvent(re));
- re.dispatch();
+ re.throwIfException();
CPPUNIT_ASSERT_EQUAL(std::string(hello), std::string(readBuf));
+ }
+
+ void testAccept() {
+ Socket s = Socket::createTcp();
+ int port = s.listen(0, 10);
+ CPPUNIT_ASSERT(port != 0);
+
+ AcceptEvent ae(s.fd());
+ Socket client = Socket::createTcp();
+ connectSendRead(ae, port, client);
+ Socket client2 = Socket::createTcp();
+ connectSendRead(ae, port, client2);
+ client.close();
+ client2.close();
+ Socket client3 = Socket::createTcp();
+ connectSendRead(ae, port, client3);
}
};
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelThreadsTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelThreadsTest.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelThreadsTest.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/EventChannelThreadsTest.cpp Wed Dec 20 09:54:05 2006
@@ -42,7 +42,7 @@
* We count the total number of events, and the
* number of reads and writes for each message number.
*/
-class TestResults : public Monitor {
+class TestResults {
public:
TestResults() : isShutdown(false), nEventsRemaining(totalEvents) {}
@@ -62,20 +62,21 @@
}
void shutdown(const std::string& exceptionMsg = std::string()) {
- ScopedLock lock(*this);
+ Monitor::ScopedLock lock(monitor);
exception = exceptionMsg;
isShutdown = true;
- notifyAll();
+ monitor.notifyAll();
}
void wait() {
- ScopedLock lock(*this);
+ Monitor::ScopedLock lock(monitor);
Time deadline = now() + 10*TIME_SEC;
while (!isShutdown) {
- CPPUNIT_ASSERT(Monitor::wait(deadline));
+ CPPUNIT_ASSERT(monitor.wait(deadline));
}
}
+ Monitor monitor;
bool isShutdown;
std::string exception;
AtomicCount reads[nMessages];
@@ -113,30 +114,34 @@
};
/** Repost an event N times. */
-class Repost {
+template <class T>
+class Reposter {
public:
- Repost(int n) : count (n) {}
- virtual ~Repost() {}
+ Reposter(T* event_, int n) : event(event_), original(*event_), count (n) {}
+ virtual ~Reposter() {}
- void repost(Event* event) {
+ void repost() {
if (--count==0) {
delete event;
} else {
- threads->postEvent(event);
+ *event = original;
+ threads->post(event);
}
}
private:
+ T* event;
+ T original;
int count;
};
/** Repeating read event. */
-class TestReadEvent : public ReadEvent, public Runnable, private Repost {
+class TestReadEvent : public ReadEvent, public Runnable {
public:
explicit TestReadEvent(int fd=-1) :
ReadEvent(fd, &value, sizeof(value), SafeCallback(*this)),
- Repost(nMessages)
+ reposter(this, nMessages)
{}
void run() {
@@ -144,47 +149,50 @@
CPPUNIT_ASSERT(0 <= value);
CPPUNIT_ASSERT(value < nMessages);
results.countRead(value);
- repost(this);
+ reposter.repost();
}
private:
int value;
- ReadEvent original;
+ Reposter<ReadEvent> reposter;
};
/** Fire and forget write event */
-class TestWriteEvent : public WriteEvent, public Runnable, private Repost {
+class TestWriteEvent : public WriteEvent, public Runnable {
public:
TestWriteEvent(int fd=-1) :
WriteEvent(fd, &value, sizeof(value), SafeCallback(*this)),
- Repost(nMessages),
+ reposter(this, nMessages),
value(0)
{}
void run() {
CPPUNIT_ASSERT_EQUAL(sizeof(int), getSize());
results.countWrite(value++);
- repost(this);
+ reposter.repost();
}
private:
+ Reposter<WriteEvent> reposter;
int value;
};
/** Fire-and-forget Accept event, posts reads on the accepted connection. */
-class TestAcceptEvent : public AcceptEvent, public Runnable, private Repost {
+class TestAcceptEvent : public AcceptEvent, public Runnable {
public:
TestAcceptEvent(int fd=-1) :
AcceptEvent(fd, SafeCallback(*this)),
- Repost(nConnections)
+ reposter(this, nConnections)
{}
void run() {
- threads->postEvent(new TestReadEvent(getAcceptedDesscriptor()));
+ threads->post(new TestReadEvent(getAcceptedDesscriptor()));
results.countEvent();
- repost(this);
+ reposter.repost();
}
+ private:
+ Reposter<AcceptEvent> reposter;
};
class EventChannelThreadsTest : public CppUnit::TestCase
@@ -207,10 +215,10 @@
{
Socket listener = Socket::createTcp();
int port = listener.listen();
-
+
// Post looping accept events, will repost nConnections times.
// The accept event will automatically post read events.
- threads->postEvent(new TestAcceptEvent(listener.fd()));
+ threads->post(new TestAcceptEvent(listener.fd()));
// Make connections.
Socket connections[nConnections];
@@ -221,7 +229,7 @@
// Post looping write events.
for (int i = 0; i < nConnections; ++i) {
- threads->postEvent(new TestWriteEvent(connections[i].fd()));
+ threads->post(new TestWriteEvent(connections[i].fd()));
}
// Wait for all events to be dispatched.
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/Makefile.am?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/Makefile.am (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/Makefile.am Wed Dec 20 09:54:05 2006
@@ -16,9 +16,10 @@
topicall \
topictest \
qpid_test_plugin.h \
- APRBaseTest.cpp
+ MockSessionHandler.h
-client_tests = \
+
+client_exe_tests = \
client_test \
echo_service \
topic_listener \
@@ -41,7 +42,8 @@
TxAckTest \
TxBufferTest \
TxPublishTest \
- ValueTest
+ ValueTest \
+ AcceptorTest
framing_tests = \
BodyHandlerTest \
@@ -54,17 +56,21 @@
posix_tests = \
EventChannelTest \
- EventChannelThreadsTest
+ EventChannelThreadsTest \
+ EventChannelConnectionTest
+
+apr_tests = APRBaseTest.cpp
unit_tests = \
$(broker_tests) \
$(framing_tests) \
- $(misc_tests)
+ $(misc_tests) \
+ $(posix_tests)
-noinst_PROGRAMS = $(client_tests)
+noinst_PROGRAMS = $(client_exe_tests)
-TESTS = run-unit-tests run-python-tests
+TESTS = run-unit-tests run-system-tests
EXTRA_DIST += $(TESTS)
include gen.mk
@@ -77,7 +83,7 @@
gen.mk: Makefile.am
( \
- for i in $(client_tests); do \
+ for i in $(client_exe_tests); do \
echo $${i}_SOURCES = $$i.cpp; \
echo $${i}_LDADD = '$$(lib_client) $$(lib_common) $$(extra_libs)'; \
done; \
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/run-python-tests
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/run-python-tests?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/run-python-tests (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/run-python-tests Wed Dec 20 09:54:05 2006
@@ -1,15 +0,0 @@
-#!/bin/sh
-
-set -e
-log=`pwd`/qpidd.log
-# Start the daemon, recording its PID.
-../src/qpidd > $log 2>&1 & pid=$!
-
-# Arrange to kill the daemon upon any type of termination.
-trap 'status=$?; kill $pid; exit $status' 0
-trap '(exit $?); exit $?' 1 2 13 15
-
-# Run the tests.
-cd ../../python && ./run-tests -v -I cpp_failing.txt
-
-rm -f $log
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topic_publisher.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topic_publisher.cpp?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topic_publisher.cpp (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topic_publisher.cpp Wed Dec 20 09:54:05 2006
@@ -138,13 +138,15 @@
int64_t sum(0);
for(int i = 0; i < batchSize; i++){
if(i > 0 && args.getDelay()) sleep(args.getDelay());
- Time time = publisher.publish(
- args.getMessages(), args.getSubscribers(), args.getSize());
- if(!max || time > max) max = time;
- if(!min || time < min) min = time;
- sum += time;
+ int64_t msecs =
+ publisher.publish(args.getMessages(),
+ args.getSubscribers(),
+ args.getSize()) / TIME_MSEC;
+ if(!max || msecs > max) max = msecs;
+ if(!min || msecs < min) min = msecs;
+ sum += msecs;
std::cout << "Completed " << (i+1) << " of " << batchSize
- << " in " << time/TIME_MSEC << "ms" << std::endl;
+ << " in " << msecs << "ms" << std::endl;
}
publisher.terminate();
int64_t avg = sum / batchSize;
Modified: incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topictest
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topictest?view=diff&rev=489159&r1=489158&r2=489159
==============================================================================
--- incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topictest (original)
+++ incubator/qpid/branches/event-queue-2006-12-20/cpp/tests/topictest Wed Dec 20 09:54:05 2006
@@ -1,42 +1,41 @@
#!/bin/bash
-# Run the c++ or java topic test
+# Run the c++ or topic test
-. `dirname $0`/env
-
-# Edit parameters here:
-
-# Big test:
-# LISTENERS=10
-# MESSAGES=10000
-# BATCHES=20
-
-LISTENERS=10
+# Defaults
+SUBSCRIBERS=10
MESSAGES=2000
BATCHES=10
-cppcmds() {
- LISTEN_CMD=topic_listener
- PUBLISH_CMD="topic_publisher -messages $MESSAGES -batches $BATCHES -subscribers $LISTENERS"
-}
+while getopts "s:m:b:" opt ; do
+ case $opt in
+ s) SUBSCRIBERS=$OPTARG ;;
+ m) MESSAGES=$OPTARG ;;
+ b) BATCHES=$OPTARG ;;
+ ?)
+ echo "Usage: %0 [-l <subscribers>] [-m <messages.] [-b <batches>]"
+ exit 1
+ ;;
+ esac
+done
-javacmds() {
- DEF=-Damqj.logging.level="error"
- LISTEN_CMD="qpid-run $DEF org.apache.qpid.topic.Listener"
- PUBLISH_CMD="qpid-run $DEF org.apache.qpid.topic.Publisher -messages $MESSAGES -batch $BATCHES -clients $LISTENERS"
+subscribe() {
+ ID=$1
+ echo "subscriber $ID"
+ ./topic_listener > subscriber.$ID 2>&1 || {
+ echo "SUBSCRIBER %ID FAILED: " ;
+ cat subscriber.$ID
+ }
+ rm subscriber.$ID
}
-case $1 in
- c) cppcmds ;;
- j) javacmds ;;
- *) cppcmds ;;
-esac
+publish() {
+ ./topic_publisher -messages $MESSAGES -batches $BATCHES -subscribers $SUBSCRIBERS
+}
-for ((i=$LISTENERS ; i--; )); do
- $LISTEN_CMD > /dev/null 2>&1 &
+for ((i=$SUBSCRIBERS ; i--; )); do
+ subscribe $i &
done
sleep 1
-echo $PUBLISH_CMD $OPTIONS
-
STATS=~/bin/topictest.times
echo "---- topictest `date`" >> $STATS
-$PUBLISH_CMD $OPTIONS | tee -a $STATS
+publish | tee -a $STATS