diff options
Diffstat (limited to 'qpid/cpp/src/tests/DispatcherTest.cpp')
-rw-r--r-- | qpid/cpp/src/tests/DispatcherTest.cpp | 47 |
1 files changed, 27 insertions, 20 deletions
diff --git a/qpid/cpp/src/tests/DispatcherTest.cpp b/qpid/cpp/src/tests/DispatcherTest.cpp index c619a36438..17b3b4e3e6 100644 --- a/qpid/cpp/src/tests/DispatcherTest.cpp +++ b/qpid/cpp/src/tests/DispatcherTest.cpp @@ -7,9 +7,9 @@ * 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 @@ -39,6 +39,9 @@ using namespace std; using namespace qpid::sys; +namespace qpid { +namespace tests { + int writeALot(int fd, const string& s) { int bytesWritten = 0; do { @@ -46,7 +49,7 @@ int writeALot(int fd, const string& s) { int lastWrite = ::write(fd, s.c_str(), s.size()); if ( lastWrite >= 0) { bytesWritten += lastWrite; - } + } } while (errno != EAGAIN); return bytesWritten; } @@ -54,13 +57,13 @@ int writeALot(int fd, const string& s) { int readALot(int fd) { int bytesRead = 0; char buf[10240]; - + do { errno = 0; int lastRead = ::read(fd, buf, sizeof(buf)); if ( lastRead >= 0) { bytesRead += lastRead; - } + } } while (errno != EAGAIN); return bytesRead; } @@ -83,7 +86,7 @@ void rInterrupt(DispatchHandle&) { } void wInterrupt(DispatchHandle&) { - cerr << "W"; + cerr << "W"; } DispatchHandle::Callback rcb = rInterrupt; @@ -108,15 +111,19 @@ void timer_handler(int /*signo*/, siginfo_t* /*info*/, void* /*context*/) { wh->call(wcb); } else { phase1finished = true; - assert(::timer_delete(timer) == 0); + assert(::timer_delete(timer) == 0); } } +}} // namespace qpid::tests + +using namespace qpid::tests; + int main(int /*argc*/, char** /*argv*/) { // Create poller Poller::shared_ptr poller(new Poller); - + // Create dispatcher thread Dispatcher d(poller); Dispatcher d1(poller); @@ -131,14 +138,14 @@ int main(int /*argc*/, char** /*argv*/) int sv[2]; int rc = ::socketpair(AF_UNIX, SOCK_STREAM, 0, sv); assert(rc >= 0); - + // Set non-blocking rc = ::fcntl(sv[0], F_SETFL, O_NONBLOCK); assert(rc >= 0); rc = ::fcntl(sv[1], F_SETFL, O_NONBLOCK); assert(rc >= 0); - + // Make up a large string string testString = "This is only a test ... 1,2,3,4,5,6,7,8,9,10;"; for (int i = 0; i < 8; i++) @@ -148,19 +155,19 @@ int main(int /*argc*/, char** /*argv*/) PosixIOHandle f1(sv[1]); rh = new DispatchHandleRef(f0, boost::bind(reader, _1, sv[0]), 0, 0); - wh = new DispatchHandleRef(f1, 0, boost::bind(writer, _1, sv[1], testString), 0); + wh = new DispatchHandleRef(f1, 0, boost::bind(writer, _1, sv[1], testString), 0); rh->startWatch(poller); wh->startWatch(poller); // Set up a regular itimer interupt - + // Ignore signal in this thread ::sigset_t sm; ::sigemptyset(&sm); ::sigaddset(&sm, SIGRTMIN); ::pthread_sigmask(SIG_BLOCK, &sm, 0); - + // Signal handling struct ::sigaction sa; sa.sa_sigaction = timer_handler; @@ -168,18 +175,18 @@ int main(int /*argc*/, char** /*argv*/) ::sigemptyset(&sa.sa_mask); rc = ::sigaction(SIGRTMIN, &sa,0); assert(rc == 0); - + ::sigevent se; ::memset(&se, 0, sizeof(se)); // Clear to make valgrind happy (this *is* the neatest way to do this portably - sigh) se.sigev_notify = SIGEV_SIGNAL; se.sigev_signo = SIGRTMIN; rc = ::timer_create(CLOCK_REALTIME, &se, &timer); assert(rc == 0); - + itimerspec ts = { /*.it_value = */ {2, 0}, // s, ns /*.it_interval = */ {2, 0}}; // s, ns - + rc = ::timer_settime(timer, 0, &ts, 0); assert(rc == 0); @@ -200,11 +207,11 @@ int main(int /*argc*/, char** /*argv*/) sa.sa_sigaction = stop_handler; rc = ::sigaction(SIGRTMIN, &sa,0); assert(rc == 0); - + itimerspec nts = { /*.it_value = */ {30, 0}, // s, ns /*.it_interval = */ {30, 0}}; // s, ns - + rc = ::timer_create(CLOCK_REALTIME, &se, &timer); assert(rc == 0); rc = ::timer_settime(timer, 0, &nts, 0); @@ -228,7 +235,7 @@ int main(int /*argc*/, char** /*argv*/) rc = ::timer_delete(timer); assert(rc == 0); - + poller->shutdown(); dt.join(); dt1.join(); @@ -237,6 +244,6 @@ int main(int /*argc*/, char** /*argv*/) cout << "\nWrote: " << writtenBytes << "\n"; cout << "Read: " << readBytes << "\n"; - + return 0; } |