diff --git a/tests/connection/tst_pendingreply.cpp b/tests/connection/tst_pendingreply.cpp index 7dcefb6..646aa04 100644 --- a/tests/connection/tst_pendingreply.cpp +++ b/tests/connection/tst_pendingreply.cpp @@ -1,137 +1,133 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "connection.h" #include "../testutil.h" #include #include static void addressMessageToBus(Message *msg) { msg->setType(Message::MethodCallMessage); msg->setDestination("org.freedesktop.DBus"); msg->setInterface("org.freedesktop.DBus"); msg->setPath("/org/freedesktop/DBus"); } class ReplyCheck : public IMessageReceiver { public: - EventDispatcher *m_eventDispatcher; - void handlePendingReplyFinished(PendingReply *pr, Connection *) override + void handlePendingReplyFinished(PendingReply *pr, Connection *connection) override { pr->dumpState(); std::cout << "got it!\n" << pr->reply()->arguments().prettyPrint(); TEST(pr->isFinished()); TEST(!pr->isError()); // This is really a different test, it used to reproduce a memory leak under Valgrind Message reply = pr->takeReply(); - m_eventDispatcher->interrupt(); + connection->eventDispatcher()->interrupt(); } }; static void testBusAddress(bool waitForConnected) { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); Message msg; addressMessageToBus(&msg); msg.setMethod("RequestName"); Arguments::Writer writer; writer.writeString("Bana.nana"); // requested name writer.writeUint32(4); // TODO proper enum or so: 4 == DBUS_NAME_FLAG_DO_NOT_QUEUE msg.setArguments(writer.finish()); if (waitForConnected) { // finish creating the connection while (conn.uniqueName().empty()) { eventDispatcher.poll(); } } PendingReply busNameReply = conn.send(std::move(msg)); ReplyCheck replyCheck; - replyCheck.m_eventDispatcher = &eventDispatcher; busNameReply.setReceiver(&replyCheck); while (eventDispatcher.poll()) { } } class TimeoutCheck : public IMessageReceiver { public: - EventDispatcher *m_eventDispatcher; - void handlePendingReplyFinished(PendingReply *reply, Connection *) override + void handlePendingReplyFinished(PendingReply *reply, Connection *connection) override { TEST(reply->isFinished()); TEST(!reply->hasNonErrorReply()); TEST(reply->error().code() == Error::Timeout); std::cout << "We HAVE timed out.\n"; - m_eventDispatcher->interrupt(); + connection->eventDispatcher()->interrupt(); } }; static void testTimeout() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); // finish creating the connection; we need to know our own name so we can send the message to // ourself so we can make sure that there will be no reply :) while (conn.uniqueName().empty()) { eventDispatcher.poll(); } Message msg = Message::createCall("/some/dummy/path", "org.no_interface", "non_existent_method"); msg.setDestination(conn.uniqueName()); PendingReply neverGonnaGetReply = conn.send(std::move(msg), 200); TimeoutCheck timeoutCheck; - timeoutCheck.m_eventDispatcher = &eventDispatcher; neverGonnaGetReply.setReceiver(&timeoutCheck); while (eventDispatcher.poll()) { } } int main(int, char *[]) { testBusAddress(false); testBusAddress(true); testTimeout(); // TODO testBadCall std::cout << "Passed!\n"; } diff --git a/tests/connection/tst_server.cpp b/tests/connection/tst_server.cpp index 7e9b6dc..da293cd 100644 --- a/tests/connection/tst_server.cpp +++ b/tests/connection/tst_server.cpp @@ -1,338 +1,336 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "connection.h" #include "eventdispatcher.h" #include "iconnectionstatelistener.h" #include "imessagereceiver.h" #include "inewconnectionlistener.h" #include "message.h" #include "pendingreply.h" #include "server.h" #include "../testutil.h" #include #include #include #include #include /* Sequence diagram of successful test runs. There are three runs with three connections each, where what happens in the second connection changes from test run to test run. Events are assumed to be asynchronous between threads, unless indicated as in the following example: Accept connection <- Connect to server Talk to rubber duckie In plain words: The server must accept after the client starts connecting, not earlier. In this case, the ordering is enforced naturally, in other cases, auxiliary synchronization is needed. Note that it WELL POSSIBLE that "talk to rubber duckie" has already happened when the connection is accepted by the server - the only guarantee is that accept happens after connect. TODO: - how to test connections? - review, add further details Server thread . Client thread ###### First connection - always succeeds Set up server -> Create thread Accept connection <- Connect Receive TestMsg <- Send TestMsg Send TestReply -> Receive TestReply ###### Second connection - succeeds (test run 1) or fails due to closing by client (test run 2) ###### or fals due to closing by server (test run 3) Signal next connection -> ... ### test run 1 - connection 2 succeeds Accept <- Connect Receive TestMsg <- Send TestMsg Send TestReply -> Receive TestReply (test checks this) ### test run 2 - connection 2 failed by client Accept <- Connect Receive connection <- Close closed error (check) ### test run 3 - connection 2 failed by server Accept <- Connect Send TestMsg Close -> Receive failed PendingReply (check) ###### Third connection - always succeeds Accept <- Connect Receive TestMsg <- Send TestMsg Send TestReply -> Receive TestReply (check) */ enum TestConstants { BrokenConnectionIndex = 1, ConnectionsPerTestRun = 3, NoFailTestRun = 0, ClientCloseTestRun = 1, ServerCloseTestRun = 2, TestRunCount = 3, ReplyTimeoutMsecs = 25000 // TODO back to 250 }; //////////////////////// client thread (a secondary thread) ///////////////////// class ClientSideHandlers : public IConnectionStateListener, public IMessageReceiver { public: ~ClientSideHandlers() {} // IConnectionStateListener void handleConnectionChanged(Connection *, Connection::State, Connection::State newState) override { if (newState == Connection::Unconnected && m_testRunIndex == ServerCloseTestRun) { std::cerr << "Client thread: handling disconnect" << std::endl; m_serverClosedConnections++; } } // IMessageReceiver - void handlePendingReplyFinished(PendingReply *pr, Connection *) override + void handlePendingReplyFinished(PendingReply *pr, Connection *connection) override { std::cerr << "Client thread: received pong" << " " << pr->hasNonErrorReply() << std::endl; if (pr->hasNonErrorReply()) { m_receivedSuccessReplies++; } else { m_receivedErrorReplies++; } - m_eventDispatcher->interrupt(); + connection->eventDispatcher()->interrupt(); } - EventDispatcher *m_eventDispatcher = nullptr; int m_testRunIndex = 0; int m_serverClosedConnections = 0; int m_receivedSuccessReplies = 0; int m_receivedErrorReplies = 0; }; static void clientThreadRun(ConnectAddress address, int testRunIndex) { EventDispatcher eventDispatcher; ClientSideHandlers clientHandlers; - clientHandlers.m_eventDispatcher = &eventDispatcher; clientHandlers.m_testRunIndex = testRunIndex; // Client-side connections - these call listeners in ClientSideHandlers when closing during // destruction. The easiest way to ensure a non-crashing destruction order is to put them here. std::vector connections; for (int i = 0; i < TestConstants::ConnectionsPerTestRun; i++) { std::cerr << "Client thread: test run " << testRunIndex << " / connection " << i << std::endl; connections.push_back(Connection(&eventDispatcher, address)); connections.back().setConnectionStateListener(&clientHandlers); #if 1 if (i == TestConstants::BrokenConnectionIndex && testRunIndex == ClientCloseTestRun) { std::cerr << "Client thread: closing connection" << std::endl; connections.back().close(); std::cerr << "Client thread: closed connection" << std::endl; continue; } #endif Message ping = Message::createCall("/foo", "org.bar.interface", "serverTest"); PendingReply pendingReply = connections.back().send(std::move(ping), ReplyTimeoutMsecs); std::cerr << "Client thread: sent ping" << std::endl; #if 0 if (i == TestConstants::BrokenConnectionIndex && testRunIndex == ClientCloseTestRun) { std::cerr << "Client thread: closing connection" << std::endl; while (connections.back().sendQueueLength()) { eventDispatcher.poll(); } connections.back().close(); std::cerr << "Client thread: closed connection" << std::endl; continue; } else #endif pendingReply.setReceiver(&clientHandlers); while (eventDispatcher.poll()) { } if (i == TestConstants::BrokenConnectionIndex && testRunIndex == ServerCloseTestRun) { TEST(pendingReply.error().code() == Error::RemoteDisconnect); } else { TEST(!pendingReply.error().isError()); } } if (testRunIndex == NoFailTestRun) { TEST(clientHandlers.m_serverClosedConnections == 0); TEST(clientHandlers.m_receivedSuccessReplies == 3); TEST(clientHandlers.m_receivedErrorReplies == 0); } else if (testRunIndex == ClientCloseTestRun) { TEST(clientHandlers.m_serverClosedConnections == 0); TEST(clientHandlers.m_receivedSuccessReplies == 2); TEST(clientHandlers.m_receivedErrorReplies == 0); } else { TEST(clientHandlers.m_serverClosedConnections == 1); TEST(clientHandlers.m_receivedSuccessReplies == 2); TEST(clientHandlers.m_receivedErrorReplies == 1); } } //////////////////////// server thread (the main thread) ///////////////////// class ServerSideHandlers : public INewConnectionListener, public IConnectionStateListener, public IMessageReceiver { public: // INewConnectionListener void handleNewConnection(Server *server) override { std::unique_ptr conn(server->takeNextClient()); TEST(conn); // for now this is simply not allowed... we could try to check why this // happened, if it ever happens conn->setSpontaneousMessageReceiver(this); conn->setConnectionStateListener(this); const size_t connectionIndex = m_connections.size(); m_connections.push_back(std::move(*conn)); if (connectionIndex == TestConstants::BrokenConnectionIndex && m_testRunIndex == ServerCloseTestRun) { m_connections.back().close(); stopListeningToConnection(&m_connections.back(), "we closed"); } } // IConnectionStateListener void handleConnectionChanged(Connection *conn, Connection::State oldState, Connection::State newState) override { const auto it = std::find_if(m_connections.begin(), m_connections.end(), [conn] (Connection &c) { return conn == &c; }); const int connIndex = std::distance(m_connections.begin(), it); std::cerr << "Server thread: handling state change @ index " << connIndex << " from " << oldState << " to " << newState << std::endl; if (newState != Connection::Unconnected) { return; } std::cerr << "Server thread: handling disconnect @ index " << connIndex << std::endl; if (connIndex == TestConstants::BrokenConnectionIndex) { if (m_testRunIndex == ClientCloseTestRun) { std::cerr << " *** HURZ ***" << std::endl; m_clientClosedConnectionAtTheRightPoint++; stopListeningToConnection(conn, "disconnected"); } } } // IMessageReceiver void handleSpontaneousMessageReceived(Message message, Connection *conn) override { std::cerr << "Server thread: received ping" << std::endl; conn->sendNoReply(Message::createReplyTo(message)); stopListeningToConnection(conn, "ping received"); } int m_testRunIndex = 0; std::vector m_connections; // server-side connections int m_connectionsFullyHandled = 0; int m_clientClosedConnectionAtTheRightPoint = 0; private: void stopListeningToConnection(Connection *conn, const char *reason) { std::cerr << "Server thread: start ignoring connection because " << reason << std::endl; conn->setSpontaneousMessageReceiver(nullptr); conn->setConnectionStateListener(nullptr); m_connectionsFullyHandled++; } }; static void testAcceptMultiple(int testRunIndex) { // Accept multiple connections and run a ping-pong message test on each. If withFail is true, // abort one connection from the client side and check that the rest still works. EventDispatcher eventDispatcher; ConnectAddress addr; addr.setRole(ConnectAddress::Role::PeerServer); #ifdef __unix__ addr.setType(ConnectAddress::Type::TmpDir); addr.setPath("/tmp"); #else addr.setType(ConnectAddress::Type::Tcp); addr.setPort(36816 /* randomly selected ;) */); #endif Server server(&eventDispatcher, addr); ServerSideHandlers serverHandler; serverHandler.m_testRunIndex = testRunIndex; server.setNewConnectionListener(&serverHandler); ConnectAddress clientAddr = server.concreteAddress(); clientAddr.setRole(ConnectAddress::Role::PeerClient); std::thread clientThread(clientThreadRun, clientAddr, testRunIndex); while (serverHandler.m_connectionsFullyHandled < ConnectionsPerTestRun || (serverHandler.m_connections.back().state() != Connection::Unconnected && serverHandler.m_connections.back().sendQueueLength())) { eventDispatcher.poll(); } clientThread.join(); TEST(serverHandler.m_connectionsFullyHandled == ConnectionsPerTestRun); if (testRunIndex == ClientCloseTestRun) { TEST(serverHandler.m_clientClosedConnectionAtTheRightPoint == 1); } else { TEST(serverHandler.m_clientClosedConnectionAtTheRightPoint == 0); } } int main(int, char *[]) { for (int i = 0; i < TestRunCount; i++) { testAcceptMultiple(i); } std::cout << "Passed!\n"; } diff --git a/tests/connection/tst_threads.cpp b/tests/connection/tst_threads.cpp index edb7c6e..9174f63 100644 --- a/tests/connection/tst_threads.cpp +++ b/tests/connection/tst_threads.cpp @@ -1,242 +1,238 @@ /* Copyright (C) 2014 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "stringtools.h" #include "connection.h" #include "../testutil.h" #include #include #include static const char *echoPath = "/echo"; // make the name "fairly unique" because the interface name is our only protection against replying // to the wrong message static const char *echoInterface = "org.example_fb39a8dbd0aa66d2.echo"; static const char *echoMethod = "echo"; //////////////// Multi-thread ping-pong test //////////////// static const char *pingPayload = "-> J. Random PING"; static const char *pongPayload = "<- J. Random Pong"; class PongSender : public IMessageReceiver { public: - Connection *m_connection; - - void handleSpontaneousMessageReceived(Message ping, Connection *) override + void handleSpontaneousMessageReceived(Message ping, Connection *connection) override { if (ping.interface() != echoInterface) { // This is not the ping... it is probably still something from connection setup. // We can possibly receive many things here that we were not expecting. return; } { Arguments args = ping.arguments(); Arguments::Reader reader(args); cstring payload = reader.readString(); TEST(!reader.error().isError()); TEST(reader.isFinished()); std::cout << "we have ping with payload: " << payload.ptr << std::endl; } { Message pong = Message::createReplyTo(ping); Arguments::Writer writer; writer.writeString(pongPayload); pong.setArguments(writer.finish()); std::cout << "\n\nSending pong!\n\n"; - Error replyError = m_connection->sendNoReply(std::move(pong)); + Error replyError = connection->sendNoReply(std::move(pong)); TEST(!replyError.isError()); - m_connection->eventDispatcher()->interrupt(); + connection->eventDispatcher()->interrupt(); } } }; static void pongThreadRun(Connection::CommRef mainConnectionRef, std::atomic *pongThreadReady) { std::cout << " Pong thread starting!\n"; EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, std::move(mainConnectionRef)); PongSender pongSender; - pongSender.m_connection = &conn; - conn.setSpontaneousMessageReceiver(&pongSender); while (eventDispatcher.poll()) { std::cout << " Pong thread waking up!\n"; if (conn.uniqueName().length()) { pongThreadReady->store(true); // HACK: we do this only to wake up the main thread's event loop std::cout << "\n\nSending WAKEUP package!!\n\n"; Message wakey = Message::createCall(echoPath, "org.notexample.foo", echoMethod); wakey.setDestination(conn.uniqueName()); conn.sendNoReply(std::move(wakey)); } else { std::cout << " Pong thread: NO NAME YET!\n"; } // receive ping message // send pong message } std::cout << " Pong thread almost finished!\n"; } class PongReceiver : public IMessageReceiver { public: void handlePendingReplyFinished(PendingReply *pongReply, Connection *) override { TEST(!pongReply->error().isError()); Message pong = pongReply->takeReply(); Arguments args = pong.arguments(); Arguments::Reader reader(args); std::string strPayload = toStdString(reader.readString()); TEST(!reader.error().isError()); TEST(reader.isFinished()); TEST(strPayload == pongPayload); } }; static void testPingPong() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); std::atomic pongThreadReady(false); std::thread pongThread(pongThreadRun, conn.createCommRef(), &pongThreadReady); // finish creating the connection while (conn.uniqueName().empty()) { std::cout << "."; eventDispatcher.poll(); } std::cout << "we have connection! " << conn.uniqueName() << "\n"; // send ping message to other thread Message ping = Message::createCall(echoPath, echoInterface, echoMethod); Arguments::Writer writer; writer.writeString(pingPayload); ping.setArguments(writer.finish()); ping.setDestination(conn.uniqueName()); PongReceiver pongReceiver; PendingReply pongReply; bool sentPing = false; while (!sentPing || !pongReply.isFinished()) { eventDispatcher.poll(); if (pongThreadReady.load() && !sentPing) { std::cout << "\n\nSending ping!!\n\n"; pongReply = conn.send(std::move(ping)); pongReply.setReceiver(&pongReceiver); sentPing = true; } } TEST(pongReply.hasNonErrorReply()); std::cout << "we have pong!\n"; pongThread.join(); } //////////////// Multi-threaded timeout test //////////////// class TimeoutReceiver : public IMessageReceiver { public: void handlePendingReplyFinished(PendingReply *reply, Connection *) override { TEST(reply->isFinished()); TEST(!reply->hasNonErrorReply()); TEST(reply->error().code() == Error::Timeout); std::cout << "We HAVE timed out.\n"; } }; static void timeoutThreadRun(Connection::CommRef mainConnectionRef, std::atomic *done) { // TODO v turn this into proper documentation in Connection // Open a Connection "slaved" to the other Connection - it runs its own event loop in this thread // and has message I/O handled by the Connection in the "master" thread through message passing. // The main purpose of that is to use just one DBus connection per application( module), which is often // more convenient for client programmers and brings some limited ordering guarantees. std::cout << " Other thread starting!\n"; EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, std::move(mainConnectionRef)); while (!conn.uniqueName().length()) { eventDispatcher.poll(); } Message notRepliedTo = Message::createCall(echoPath, echoInterface, echoMethod); notRepliedTo.setDestination(conn.uniqueName()); PendingReply deadReply = conn.send(std::move(notRepliedTo), 50); TimeoutReceiver timeoutReceiver; deadReply.setReceiver(&timeoutReceiver); while (!deadReply.isFinished()) { eventDispatcher.poll(); } *done = true; } static void testThreadedTimeout() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); std::atomic done(false); std::thread timeoutThread(timeoutThreadRun, conn.createCommRef(), &done); while (!done) { eventDispatcher.poll(); } timeoutThread.join(); } // more things to test: // - (do we want to do this, and if so here??) blocking on a reply through other thread's connection // - ping-pong with several messages queued - every message should arrive exactly once and messages // should arrive in sending order (can use serials for that as simplificitaion) int main(int, char *[]) { testPingPong(); testThreadedTimeout(); std::cout << "Passed!\n"; } diff --git a/tests/serialization/tst_message.cpp b/tests/serialization/tst_message.cpp index 981bd3c..11c9347 100644 --- a/tests/serialization/tst_message.cpp +++ b/tests/serialization/tst_message.cpp @@ -1,301 +1,294 @@ /* Copyright (C) 2013 Andreas Hartmetz This library is free software; you can redistribute it and/or modify it under the terms of the GNU Library General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version. This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more details. You should have received a copy of the GNU Library General Public License along with this library; see the file COPYING.LGPL. If not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. Alternatively, this file is available under the Mozilla Public License Version 1.1. You may obtain a copy of the License at http://www.mozilla.org/MPL/ */ #include "arguments.h" #include "connectaddress.h" #include "error.h" #include "eventdispatcher.h" #include "imessagereceiver.h" #include "message.h" #include "pendingreply.h" #include "testutil.h" #include "connection.h" #include #include #include #include static void test_signatureHeader() { Message msg; Arguments::Writer writer; writer.writeByte(123); writer.writeUint64(1); msg.setArguments(writer.finish()); TEST(msg.signature() == "yt"); } class PrintAndTerminateClient : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg, Connection *) override + void handleSpontaneousMessageReceived(Message msg, Connection *connection) override { std::cout << msg.prettyPrint(); - m_eventDispatcher->interrupt(); + connection->eventDispatcher()->interrupt(); } - EventDispatcher *m_eventDispatcher; }; class PrintAndReplyClient : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg, Connection *) override + void handleSpontaneousMessageReceived(Message msg, Connection *connection) override { std::cout << msg.prettyPrint(); - m_connection->sendNoReply(Message::createErrorReplyTo(msg, "Unable to get out of hammock!")); - //m_connection->eventDispatcher()->interrupt(); + connection->sendNoReply(Message::createErrorReplyTo(msg, "Unable to get out of hammock!")); + //connection->eventDispatcher()->interrupt(); } - Connection *m_connection; }; // used during implementation, is supposed to not crash and be valgrind-clean afterwards void testBasic(const ConnectAddress &clientAddress) { EventDispatcher dispatcher; ConnectAddress serverAddress = clientAddress; serverAddress.setRole(ConnectAddress::Role::PeerServer); Connection serverConnection(&dispatcher, serverAddress); std::cout << "Created server connection. " << &serverConnection << std::endl; Connection clientConnection(&dispatcher, clientAddress); std::cout << "Created client connection. " << &clientConnection << std::endl; PrintAndReplyClient printAndReplyClient; - printAndReplyClient.m_connection = &serverConnection; serverConnection.setSpontaneousMessageReceiver(&printAndReplyClient); PrintAndTerminateClient printAndTerminateClient; - printAndTerminateClient.m_eventDispatcher = &dispatcher; clientConnection.setSpontaneousMessageReceiver(&printAndTerminateClient); Message msg = Message::createCall("/foo", "org.foo.interface", "laze"); Arguments::Writer writer; writer.writeString("couch"); msg.setArguments(writer.finish()); clientConnection.sendNoReply(std::move(msg)); while (dispatcher.poll()) { } } void testMessageLength() { static const uint32 bufferSize = Arguments::MaxArrayLength + 1024; byte *buffer = static_cast(malloc(bufferSize)); memset(buffer, 0, bufferSize); for (int i = 0; i < 2; i++) { const bool makeTooLong = i == 1; Arguments::Writer writer; writer.writePrimitiveArray(Arguments::Byte, chunk(buffer, Arguments::MaxArrayLength)); // Our minimal Message is going to have the following variable headers (in that order): // Array: 4 byte length prefix // PathHeader: 4 byte length prefix // MethodHeader: 4 byte length prefix // SignatureHeader: 1 byte length prefix // This is VERY tedious to calculate, so let's just take it as an experimentally determined value uint32 left = Arguments::MaxMessageLength - Arguments::MaxArrayLength - 72; if (makeTooLong) { left += 1; } writer.writePrimitiveArray(Arguments::Byte, chunk(buffer, left)); Message msg = Message::createCall("/a", "x"); msg.setSerial(1); msg.setArguments(writer.finish()); std::vector saved = msg.save(); TEST(msg.error().isError() == makeTooLong); } } enum { // a small integer could be confused with an index into the fd array (in the implementation), // so make it large DummyFdOffset = 1000000 }; static Arguments createArgumentsWithDummyFileDescriptors(uint fdCount) { Arguments::Writer writer; for (uint i = 0; i < fdCount; i++) { writer.writeUnixFd(DummyFdOffset - i); } return writer.finish(); } void testFileDescriptorsInArguments() { // Note: This replaces round-trip tests with file descriptors in tst_arguments. // A full roundtrip test must go through Message due to the out-of-band way that file // descriptors are stored (which is so because they are also transmitted out-of-band). Message msg = Message::createCall("/foo", "org.foo.interface", "doNothing"); for (uint i = 0; i < 4; i++) { msg.setArguments(createArgumentsWithDummyFileDescriptors(i)); { // const ref to arguments const Arguments &args = msg.arguments(); Arguments::Reader reader(args); for (uint j = 0; j < i; j++) { TEST(reader.readUnixFd() == int(DummyFdOffset - j)); TEST(reader.isValid()); } TEST(reader.isFinished()); } { // copy of arguments Arguments args = msg.arguments(); Arguments::Reader reader(args); for (uint j = 0; j < i; j++) { TEST(reader.readUnixFd() == int(DummyFdOffset - j)); TEST(reader.isValid()); } TEST(reader.isFinished()); } } } void testTooManyFileDescriptors() { // TODO re-think what is the best place to catch too many file descriptors... Arguments::Writer writer; } void testFileDescriptorsHeader() { Message msg = Message::createCall("/foo", "org.foo.interface", "doNothing"); for (uint i = 0; i < 4; i++) { msg.setArguments(createArgumentsWithDummyFileDescriptors(i)); TEST(msg.unixFdCount() == i); } } enum { // for pipe2() file descriptor array ReadSide = 0, WriteSide = 1, // how many file descriptors to send in test FdCountToSend = 10 }; class FileDescriptorTestReceiver : public IMessageReceiver { public: - void handleSpontaneousMessageReceived(Message msg, Connection *) override + void handleSpontaneousMessageReceived(Message msg, Connection *connection) override { // we're on the session bus, so we'll receive all kinds of notifications we don't care about here if (msg.type() != Message::MethodCallMessage || msg.method() != "testFileDescriptorsForDataTransfer") { return; } Arguments::Reader reader(msg.arguments()); for (uint i = 0; i < FdCountToSend; i++) { int fd = reader.readUnixFd(); uint readBuf = 12345; ::read(fd, &readBuf, sizeof(uint)); ::close(fd); TEST(readBuf == i); } Message reply = Message::createReplyTo(msg); - m_connection->sendNoReply(std::move(reply)); + connection->sendNoReply(std::move(reply)); } - - Connection *m_connection = nullptr; }; void testFileDescriptorsForDataTransfer() { EventDispatcher eventDispatcher; Connection conn(&eventDispatcher, ConnectAddress::StandardBus::Session); conn.waitForConnectionEstablished(); TEST(conn.isConnected()); int pipeFds[2 * FdCountToSend]; Message msg = Message::createCall("/foo", "org.foo.interface", "testFileDescriptorsForDataTransfer"); msg.setDestination(conn.uniqueName()); Arguments::Writer writer; for (uint i = 0; i < FdCountToSend; i++) { TEST(pipe2(pipeFds + 2 * i, O_NONBLOCK) == 0); // write into write side of the pipe... will be read when the message is received back from bus ::write(pipeFds[2 * i + WriteSide], &i, sizeof(uint)); writer.writeUnixFd(pipeFds[2 * i + ReadSide]); } msg.setArguments(writer.finish()); PendingReply reply = conn.send(std::move(msg), 500 /* fail quickly */); FileDescriptorTestReceiver fdTestReceiver; conn.setSpontaneousMessageReceiver(&fdTestReceiver); - fdTestReceiver.m_connection = &conn; while (!reply.isFinished()) { eventDispatcher.poll(); } TEST(reply.hasNonErrorReply()); // otherwise timeout, the message exchange failed somehow for (uint i = 0; i < FdCountToSend; i++) { ::close(pipeFds[2 * i + WriteSide]); } } int main(int, char *[]) { test_signatureHeader(); #ifdef __linux__ { ConnectAddress clientAddress; clientAddress.setType(ConnectAddress::Type::AbstractUnixPath); clientAddress.setRole(ConnectAddress::Role::PeerClient); clientAddress.setPath("dferry.Test.Message"); testBasic(clientAddress); } #endif // TODO: SocketType::Unix works on any Unix-compatible OS, but we'll need to construct a path { ConnectAddress clientAddress; clientAddress.setType(ConnectAddress::Type::Tcp); clientAddress.setPort(6800); clientAddress.setRole(ConnectAddress::Role::PeerClient); testBasic(clientAddress); } testMessageLength(); testFileDescriptorsInArguments(); testTooManyFileDescriptors(); testFileDescriptorsHeader(); testFileDescriptorsForDataTransfer(); // TODO testSaveLoad(); // TODO testDeepCopy(); std::cout << "\nNote that the hammock error is part of the test.\nPassed!\n"; }