diff --git a/CMakeLists.txt b/CMakeLists.txt index ea3747c..4dae741 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,175 +1,179 @@ project(dferry) cmake_minimum_required(VERSION 3.1.0 FATAL_ERROR) list(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake) if (CMAKE_COMPILER_IS_GNUCXX) set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wpedantic -Wextra -Werror -Wno-error=unused-result") set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fvisibility=hidden -fvisibility-inlines-hidden") endif() set(CMAKE_CXX_STANDARD 11) if (WIN32 AND CMAKE_SYSTEM_VERSION VERSION_LESS 6.0) message(FATAL_ERROR "Windows Vista or later is required.") endif() include(TestBigEndian) if (BIGENDIAN) add_definitions(-DBIGENDIAN) endif() if (UNIX) add_definitions(-D__unix__) # help for platforms that don't define this standard macro endif() option(DFERRY_BUILD_ANALYZER "Build the dfer-analyzer bus analyzer GUI" TRUE) include(GNUInstallDirs) if (WIN32) # Windows doesn't have an RPATH equivalent, so just make sure that all .dll and .exe files # are located together, so that the .exes find the .dlls at runtime set (CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}) else() set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib) set(CMAKE_INSTALL_RPATH ${CMAKE_INSTALL_FULL_LIBDIR}) # add libdfer install dir to rpath set(CMAKE_INSTALL_RPATH_USE_LINK_PATH TRUE) # add Qt (etc.) dir to rpath, if necessary endif() include_directories(${CMAKE_SOURCE_DIR}/client ${CMAKE_SOURCE_DIR}/connection ${CMAKE_SOURCE_DIR}/events ${CMAKE_SOURCE_DIR}/serialization ${CMAKE_SOURCE_DIR}/transport ${CMAKE_SOURCE_DIR}/util) set(DFER_SOURCES connection/authclient.cpp connection/connectaddress.cpp connection/connection.cpp connection/imessagereceiver.cpp + connection/inewconnectionlistener.cpp connection/pendingreply.cpp + connection/server.cpp events/event.cpp events/eventdispatcher.cpp events/foreigneventloopintegrator.cpp events/ieventpoller.cpp events/iioeventlistener.cpp events/platformtime.cpp events/timer.cpp serialization/arguments.cpp serialization/argumentsreader.cpp serialization/argumentswriter.cpp serialization/message.cpp transport/ipserver.cpp transport/ipsocket.cpp transport/iserver.cpp transport/itransport.cpp transport/itransportlistener.cpp transport/stringtools.cpp util/error.cpp util/icompletionlistener.cpp util/types.cpp) if (UNIX) list(APPEND DFER_SOURCES transport/localserver.cpp transport/localsocket.cpp) endif() set(DFER_PUBLIC_HEADERS connection/connectaddress.h connection/connection.h connection/imessagereceiver.h + connection/inewconnectionlistener.h connection/pendingreply.h + connection/server.h client/introspection.h events/eventdispatcher.h events/foreigneventloopintegrator.h events/timer.h serialization/message.h serialization/arguments.h util/commutex.h util/error.h util/export.h util/icompletionlistener.h util/types.h util/valgrind-noop.h) set(DFER_PRIVATE_HEADERS connection/authclient.h events/event.h events/ieventpoller.h events/iioeventlistener.h events/platformtime.h serialization/basictypeio.h transport/ipserver.h transport/ipsocket.h transport/iserver.h transport/itransport.h transport/itransportlistener.h transport/platform.h transport/stringtools.h) if (UNIX) list(APPEND DFER_PRIVATE_HEADERS transport/localserver.h transport/localsocket.h) endif() if (CMAKE_SYSTEM_NAME STREQUAL "Linux") list(APPEND DFER_SOURCES events/epolleventpoller.cpp) list(APPEND DFER_PRIVATE_HEADERS events/epolleventpoller.h) elseif(CMAKE_SYSTEM_NAME STREQUAL "Windows") list(APPEND DFER_PRIVATE_HEADERS events/selecteventpoller_win32.h util/winutil.h) list(APPEND DFER_SOURCES events/selecteventpoller_win32.cpp util/winutil.cpp) elseif(UNIX) list(APPEND DFER_PRIVATE_HEADERS events/selecteventpoller_unix.h) list(APPEND DFER_SOURCES events/selecteventpoller_unix.cpp) else() message(FATAL_ERROR "This operating system is not supported.") endif() set(DFER_HEADERS ${DFER_PUBLIC_HEADERS} ${DFER_PRIVATE_HEADERS}) add_library(dfer SHARED ${DFER_SOURCES} ${DFER_HEADERS}) target_include_directories(dfer INTERFACE "$") if (WIN32) target_link_libraries(dfer PRIVATE ws2_32) endif() find_package(LibTinyxml2 REQUIRED) # for the introspection parser in dferclient include_directories(${LIBTINYXML2_INCLUDE_DIRS}) find_package(Valgrind) # for checking homemade multithreading primitives if (VALGRIND_INCLUDE_DIR) add_definitions(-DHAVE_VALGRIND) include_directories(${VALGRIND_INCLUDE_DIR}) endif() # for small_vector, optional; small_vector appeared in 1.58 find_package(Boost 1.58) if (BOOST_FOUND) add_definitions(-DHAVE_BOOST) endif() set(DFERCLIENT_SOURCES client/introspection.h) set(DFERCLIENT_HEADERS client/introspection.cpp) add_library(dferclient SHARED ${DFERCLIENT_SOURCES} ${DFERCLIENT_HEADERS}) target_include_directories(dferclient INTERFACE "$") target_link_libraries(dferclient PUBLIC dfer PRIVATE ${LIBTINYXML2_LIBRARIES}) install(TARGETS dfer dferclient EXPORT dferryExports DESTINATION ${CMAKE_INSTALL_LIBDIR}) install(FILES ${DFER_PUBLIC_HEADERS} DESTINATION include/dferry) enable_testing() # need this here to get the "test" target in the toplevel build dir add_subdirectory(tests) add_subdirectory(applications) set(configModuleLocation "lib/cmake/dferry") install(EXPORT dferryExports DESTINATION "${configModuleLocation}" FILE dferryTargets.cmake) file(WRITE ${PROJECT_BINARY_DIR}/dferryConfig.cmake "include(\"\${CMAKE_CURRENT_LIST_DIR}/dferryTargets.cmake\")") install(FILES "${PROJECT_BINARY_DIR}/dferryConfig.cmake" DESTINATION "${configModuleLocation}") diff --git a/connection/connection.cpp b/connection/connection.cpp index 7e85ab7..a2b89b5 100644 --- a/connection/connection.cpp +++ b/connection/connection.cpp @@ -1,733 +1,758 @@ /* 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 "connection.h" #include "connection_p.h" #include "arguments.h" #include "authclient.h" #include "event.h" #include "eventdispatcher_p.h" #include "icompletionlistener.h" #include "imessagereceiver.h" #include "iserver.h" #include "localsocket.h" #include "message.h" #include "message_p.h" #include "pendingreply.h" #include "pendingreply_p.h" #include "stringtools.h" #include #include #include class HelloReceiver : public IMessageReceiver { public: void handlePendingReplyFinished(PendingReply *pr) override { assert(pr == &m_helloReply); (void) pr; m_parent->handleHelloReply(); } PendingReply m_helloReply; // keep it here so it conveniently goes away when it's done ConnectionPrivate *m_parent; }; class ClientConnectedHandler : public ICompletionListener { public: ~ClientConnectedHandler() override { delete m_server; } void handleCompletion(void *) override { m_parent->handleClientConnected(); } IServer *m_server; ConnectionPrivate *m_parent; }; ConnectionPrivate::ConnectionPrivate(Connection *connection, EventDispatcher *dispatcher) : m_state(Unconnected), m_connection(connection), m_client(nullptr), m_receivingMessage(nullptr), m_transport(nullptr), m_helloReceiver(nullptr), m_clientConnectedHandler(nullptr), m_eventDispatcher(dispatcher), m_authClient(nullptr), m_defaultTimeout(25000), m_sendSerial(1), m_mainThreadConnection(nullptr) { } Connection::Connection(EventDispatcher *dispatcher, const ConnectAddress &ca) : d(new ConnectionPrivate(this, dispatcher)) { d->m_connectAddress = ca; assert(d->m_eventDispatcher); EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; if (ca.type() == ConnectAddress::Type::None || ca.role() == ConnectAddress::Role::None) { std::cerr << "\nConnection: connection constructor Exit A\n\n"; return; } if (ca.role() == ConnectAddress::Role::PeerServer) { // this sets up a server that will be destroyed after accepting exactly one connection d->m_clientConnectedHandler = new ClientConnectedHandler; - d->m_clientConnectedHandler->m_server = IServer::create(ca); + ConnectAddress dummyClientAddress; + d->m_clientConnectedHandler->m_server = IServer::create(ca, &dummyClientAddress); d->m_clientConnectedHandler->m_server->setEventDispatcher(dispatcher); d->m_clientConnectedHandler->m_server->setNewConnectionListener(d->m_clientConnectedHandler); d->m_clientConnectedHandler->m_parent = d; d->m_state = ConnectionPrivate::ServerWaitingForClient; } else { d->m_transport = ITransport::create(ca); d->m_transport->setEventDispatcher(dispatcher); if (ca.role() == ConnectAddress::Role::BusClient) { d->startAuthentication(); d->m_state = ConnectionPrivate::Authenticating; } else { assert(ca.role() == ConnectAddress::Role::PeerClient); // get ready to receive messages right away d->receiveNextMessage(); d->m_state = ConnectionPrivate::Connected; } } } Connection::Connection(EventDispatcher *dispatcher, CommRef mainConnectionRef) : d(new ConnectionPrivate(this, dispatcher)) { EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; d->m_mainThreadLink = std::move(mainConnectionRef.commutex); CommutexLocker locker(&d->m_mainThreadLink); assert(locker.hasLock()); Commutex *const id = d->m_mainThreadLink.id(); if (!id) { assert(false); std::cerr << "\nConnection: slave constructor Exit A\n\n"; return; // stay in Unconnected state } // TODO how do we handle m_state? d->m_mainThreadConnection = mainConnectionRef.connection; ConnectionPrivate *mainD = d->m_mainThreadConnection; // get the current values - if we got them from e.g. the CommRef they could be outdated // and we don't want to wait for more event ping-pong SpinLocker mainLocker(&mainD->m_lock); d->m_connectAddress = mainD->m_connectAddress; // register with the main Connection SecondaryConnectionConnectEvent *evt = new SecondaryConnectionConnectEvent(); evt->connection = d; evt->id = id; EventDispatcherPrivate::get(mainD->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } +Connection::Connection(ITransport *transport, const ConnectAddress &address) + : d(new ConnectionPrivate(this, transport->eventDispatcher())) +{ + // TODO FULLY validate address, also in the other constructors and in ITransport::create() + // and in IServer::create()! + assert(address.role() == ConnectAddress::Role::PeerServer); + assert(d->m_eventDispatcher); + d->m_transport = transport; + d->m_connectAddress = address; + EventDispatcherPrivate::get(d->m_eventDispatcher)->m_connectionToNotify = d; + +#if 0 + // TODO make the client authenticate itself, roughly along these lines + // this sets up a server that will be destroyed after accepting exactly one connection + d->m_clientConnectedHandler = new ClientConnectedHandler; + d->m_clientConnectedHandler->m_server = IServer::create(ca); + d->m_clientConnectedHandler->m_server->setEventDispatcher(dispatcher); + d->m_clientConnectedHandler->m_server->setNewConnectionListener(d->m_clientConnectedHandler); + d->m_clientConnectedHandler->m_parent = d; +#endif + d->receiveNextMessage(); + d->m_state = ConnectionPrivate::Connected; +} + Connection::~Connection() { d->close(); delete d->m_transport; delete d->m_authClient; delete d->m_helloReceiver; delete d->m_receivingMessage; delete d; d = nullptr; } void Connection::close() { d->close(); } void ConnectionPrivate::close() { // Can't be main and secondary at the main time - it could be made to work, but what for? assert(m_secondaryThreadLinks.empty() || !m_mainThreadConnection); if (m_mainThreadConnection) { CommutexUnlinker unlinker(&m_mainThreadLink); if (unlinker.hasLock()) { SecondaryConnectionDisconnectEvent *evt = new SecondaryConnectionDisconnectEvent(); evt->connection = this; EventDispatcherPrivate::get(m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } } // Destroy whatever is suitable and available at a given time, in order to avoid things like // one secondary thread blocking another indefinitely and smaller dependency-related slowdowns. while (!m_secondaryThreadLinks.empty()) { for (auto it = m_secondaryThreadLinks.begin(); it != m_secondaryThreadLinks.end(); ) { CommutexUnlinker unlinker(&it->second, false); if (unlinker.willSucceed()) { if (unlinker.hasLock()) { MainConnectionDisconnectEvent *evt = new MainConnectionDisconnectEvent(); EventDispatcherPrivate::get(it->first->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } unlinker.unlinkNow(); // don't access the element after erasing it, finish it now it = m_secondaryThreadLinks.erase(it); } else { ++it; // don't block, try again next iteration } } } cancelAllPendingReplies(); EventDispatcherPrivate::get(m_eventDispatcher)->m_connectionToNotify = nullptr; } void ConnectionPrivate::startAuthentication() { m_authClient = new AuthClient(m_transport); m_authClient->setCompletionListener(this); } void ConnectionPrivate::handleHelloReply() { if (!m_helloReceiver->m_helloReply.hasNonErrorReply()) { delete m_helloReceiver; m_helloReceiver = nullptr; m_state = Unconnected; // TODO set an error, provide access to it, also set it on messages when trying to send / receive them return; } Arguments argList = m_helloReceiver->m_helloReply.reply()->arguments(); delete m_helloReceiver; m_helloReceiver = nullptr; Arguments::Reader reader(argList); assert(reader.state() == Arguments::String); cstring busName = reader.readString(); assert(reader.state() == Arguments::Finished); m_uniqueName = toStdString(busName); // tell current secondaries UniqueNameReceivedEvent evt; evt.uniqueName = m_uniqueName; for (auto &it : m_secondaryThreadLinks) { CommutexLocker otherLocker(&it.second); if (otherLocker.hasLock()) { EventDispatcherPrivate::get(it.first->m_eventDispatcher) ->queueEvent(std::unique_ptr(new UniqueNameReceivedEvent(evt))); } } m_state = Connected; } void ConnectionPrivate::handleClientConnected() { m_transport = m_clientConnectedHandler->m_server->takeNextClient(); delete m_clientConnectedHandler; m_clientConnectedHandler = nullptr; assert(m_transport); m_transport->setEventDispatcher(m_eventDispatcher); receiveNextMessage(); m_state = Connected; } void Connection::setDefaultReplyTimeout(int msecs) { d->m_defaultTimeout = msecs; } int Connection::defaultReplyTimeout() const { return d->m_defaultTimeout; } uint32 ConnectionPrivate::takeNextSerial() { uint32 ret; do { ret = m_sendSerial.fetch_add(1, std::memory_order_relaxed); } while (unlikely(ret == 0)); return ret; } Error ConnectionPrivate::prepareSend(Message *msg) { if (msg->serial() == 0) { if (!m_mainThreadConnection) { msg->setSerial(takeNextSerial()); } else { // we take a serial from the other Connection and then serialize locally in order to keep the CPU // expense of serialization local, even though it's more complicated than doing everything in the // other thread / Connection. CommutexLocker locker(&m_mainThreadLink); if (locker.hasLock()) { msg->setSerial(m_mainThreadConnection->takeNextSerial()); } else { return Error::LocalDisconnect; } } } MessagePrivate *const mpriv = MessagePrivate::get(msg); // this is unchanged by move()ing the owning Message. if (!mpriv->serialize()) { return mpriv->m_error; } return Error::NoError; } void ConnectionPrivate::sendPreparedMessage(Message msg) { MessagePrivate *const mpriv = MessagePrivate::get(&msg); mpriv->setCompletionListener(this); m_sendQueue.push_back(std::move(msg)); if (m_state == ConnectionPrivate::Connected && m_sendQueue.size() == 1) { // first in queue, don't wait for some other event to trigger sending mpriv->send(m_transport); } } PendingReply Connection::send(Message m, int timeoutMsecs) { if (timeoutMsecs == DefaultTimeout) { timeoutMsecs = d->m_defaultTimeout; } Error error = d->prepareSend(&m); PendingReplyPrivate *pendingPriv = new PendingReplyPrivate(d->m_eventDispatcher, timeoutMsecs); pendingPriv->m_connectionOrReply.connection = d; pendingPriv->m_receiver = nullptr; pendingPriv->m_serial = m.serial(); // even if we're handing off I/O to a main Connection, keep a record because that simplifies // aborting all pending replies when we disconnect from the main Connection, no matter which // side initiated the disconnection. d->m_pendingReplies.emplace(m.serial(), pendingPriv); if (error.isError()) { // Signal the error asynchronously, in order to get the same delayed completion callback as in // the non-error case. This should make the behavior more predictable and client code harder to // accidentally get wrong. To detect errors immediately, PendingReply::error() can be used. pendingPriv->m_error = error; pendingPriv->m_replyTimeout.start(0); } else { if (!d->m_mainThreadConnection) { d->sendPreparedMessage(std::move(m)); } else { CommutexLocker locker(&d->m_mainThreadLink); if (locker.hasLock()) { std::unique_ptr evt(new SendMessageWithPendingReplyEvent); evt->message = std::move(m); evt->connection = d; EventDispatcherPrivate::get(d->m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::move(evt)); } else { pendingPriv->m_error = Error::LocalDisconnect; } } } return PendingReply(pendingPriv); } Error Connection::sendNoReply(Message m) { // ### (when not called from send()) warn if sending a message without the noreply flag set? // doing that is wasteful, but might be common. needs investigation. Error error = d->prepareSend(&m); if (error.isError()) { return error; } // pass ownership to the send queue now because if the IO system decided to send the message without // going through an event loop iteration, handleCompletion would be called and expects the message to // be in the queue if (!d->m_mainThreadConnection) { d->sendPreparedMessage(std::move(m)); } else { CommutexLocker locker(&d->m_mainThreadLink); if (locker.hasLock()) { std::unique_ptr evt(new SendMessageEvent); evt->message = std::move(m); EventDispatcherPrivate::get(d->m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::move(evt)); } else { return Error::LocalDisconnect; } } return Error::NoError; } void Connection::waitForConnectionEstablished() { if (d->m_state != ConnectionPrivate::Authenticating) { return; } while (d->m_state == ConnectionPrivate::Authenticating) { d->m_authClient->handleTransportCanRead(); } if (d->m_state != ConnectionPrivate::AwaitingUniqueName) { return; } // Send the hello message assert(!d->m_sendQueue.empty()); // the hello message should be in the queue MessagePrivate *helloPriv = MessagePrivate::get(&d->m_sendQueue.front()); helloPriv->handleTransportCanWrite(); // Receive the hello reply while (d->m_state == ConnectionPrivate::AwaitingUniqueName) { MessagePrivate::get(d->m_receivingMessage)->handleTransportCanRead(); } } ConnectAddress Connection::connectAddress() const { return d->m_connectAddress; } std::string Connection::uniqueName() const { return d->m_uniqueName; } bool Connection::isConnected() const { return d->m_transport && d->m_transport->isOpen(); } EventDispatcher *Connection::eventDispatcher() const { return d->m_eventDispatcher; } IMessageReceiver *Connection::spontaneousMessageReceiver() const { return d->m_client; } void Connection::setSpontaneousMessageReceiver(IMessageReceiver *receiver) { d->m_client = receiver; } void ConnectionPrivate::handleCompletion(void *task) { switch (m_state) { case Authenticating: { assert(task == m_authClient); if (!m_authClient->isAuthenticated()) { m_state = Unconnected; } delete m_authClient; m_authClient = nullptr; if (m_state == Unconnected) { break; } m_state = AwaitingUniqueName; // Announce our presence to the bus and have it send some introductory information of its own Message hello = Message::createCall("/org/freedesktop/DBus", "org.freedesktop.DBus", "Hello"); hello.setExpectsReply(false); hello.setDestination(std::string("org.freedesktop.DBus")); MessagePrivate *const helloPriv = MessagePrivate::get(&hello); m_helloReceiver = new HelloReceiver; m_helloReceiver->m_helloReply = m_connection->send(std::move(hello)); // Small hack: Connection::send() refuses to really start sending if the connection isn't in // Connected state. So force the sending here to actually get to Connected state. helloPriv->send(m_transport); // Also ensure that the hello message is sent before any other messages that may have been // already enqueued by an API client hello = std::move(m_sendQueue.back()); m_sendQueue.pop_back(); m_sendQueue.push_front(std::move(hello)); m_helloReceiver->m_helloReply.setReceiver(m_helloReceiver); m_helloReceiver->m_parent = this; // get ready to receive the first message, the hello reply receiveNextMessage(); break; } case AwaitingUniqueName: // the code paths for these two states only diverge in the PendingReply handler case Connected: { assert(!m_authClient); if (!m_sendQueue.empty() && task == &m_sendQueue.front()) { m_sendQueue.pop_front(); if (!m_sendQueue.empty()) { MessagePrivate::get(&m_sendQueue.front())->send(m_transport); } } else { assert(task == m_receivingMessage); Message *const receivedMessage = m_receivingMessage; receiveNextMessage(); if (!maybeDispatchToPendingReply(receivedMessage)) { if (m_client) { m_client->handleSpontaneousMessageReceived(Message(std::move(*receivedMessage))); } // dispatch to other threads listening to spontaneous messages, if any for (auto it = m_secondaryThreadLinks.begin(); it != m_secondaryThreadLinks.end(); ) { SpontaneousMessageReceivedEvent *evt = new SpontaneousMessageReceivedEvent(); evt->message = *receivedMessage; CommutexLocker otherLocker(&it->second); if (otherLocker.hasLock()) { EventDispatcherPrivate::get(it->first->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); ++it; } else { ConnectionPrivate *connection = it->first; it = m_secondaryThreadLinks.erase(it); discardPendingRepliesForSecondaryThread(connection); delete evt; } } delete receivedMessage; } } break; } default: // ### decide what to do here break; }; } bool ConnectionPrivate::maybeDispatchToPendingReply(Message *receivedMessage) { if (receivedMessage->type() != Message::MethodReturnMessage && receivedMessage->type() != Message::ErrorMessage) { return false; } auto it = m_pendingReplies.find(receivedMessage->replySerial()); if (it == m_pendingReplies.end()) { return false; } if (PendingReplyPrivate *pr = it->second.asPendingReply()) { m_pendingReplies.erase(it); assert(!pr->m_isFinished); pr->handleReceived(receivedMessage); } else { // forward to other thread's Connection ConnectionPrivate *connection = it->second.asConnection(); m_pendingReplies.erase(it); assert(connection); PendingReplySuccessEvent *evt = new PendingReplySuccessEvent; evt->reply = std::move(*receivedMessage); delete receivedMessage; EventDispatcherPrivate::get(connection->m_eventDispatcher)->queueEvent(std::unique_ptr(evt)); } return true; } void ConnectionPrivate::receiveNextMessage() { m_receivingMessage = new Message; MessagePrivate *const mpriv = MessagePrivate::get(m_receivingMessage); mpriv->setCompletionListener(this); mpriv->receive(m_transport); } void ConnectionPrivate::unregisterPendingReply(PendingReplyPrivate *p) { if (m_mainThreadConnection) { CommutexLocker otherLocker(&m_mainThreadLink); if (otherLocker.hasLock()) { PendingReplyCancelEvent *evt = new PendingReplyCancelEvent; evt->serial = p->m_serial; EventDispatcherPrivate::get(m_mainThreadConnection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } } #ifndef NDEBUG auto it = m_pendingReplies.find(p->m_serial); assert(it != m_pendingReplies.end()); if (!m_mainThreadConnection) { assert(it->second.asPendingReply()); assert(it->second.asPendingReply() == p); } #endif m_pendingReplies.erase(p->m_serial); } void ConnectionPrivate::cancelAllPendingReplies() { // No locking because we should have no connections to other threads anymore at this point. // No const iteration followed by container clear because that has different semantics - many // things can happen in a callback... // In case we have pending replies for secondary threads, and we cancel all pending replies, // that is because we're shutting down, which we told the secondary thread, and it will deal // with bulk cancellation of replies. We just throw away our records about them. for (auto it = m_pendingReplies.begin() ; it != m_pendingReplies.end(); ) { PendingReplyPrivate *pendingPriv = it->second.asPendingReply(); it = m_pendingReplies.erase(it); if (pendingPriv) { // if from this thread pendingPriv->handleError(Error::LocalDisconnect); } } } void ConnectionPrivate::discardPendingRepliesForSecondaryThread(ConnectionPrivate *connection) { for (auto it = m_pendingReplies.begin() ; it != m_pendingReplies.end(); ) { if (it->second.asConnection() == connection) { it = m_pendingReplies.erase(it); // notification and deletion are handled on the event's source thread } else { ++it; } } } void ConnectionPrivate::processEvent(Event *evt) { // std::cerr << "ConnectionPrivate::processEvent() with event type " << evt->type << std::endl; switch (evt->type) { case Event::SendMessage: sendPreparedMessage(std::move(static_cast(evt)->message)); break; case Event::SendMessageWithPendingReply: { SendMessageWithPendingReplyEvent *pre = static_cast(evt); m_pendingReplies.emplace(pre->message.serial(), pre->connection); sendPreparedMessage(std::move(pre->message)); break; } case Event::SpontaneousMessageReceived: if (m_client) { SpontaneousMessageReceivedEvent *smre = static_cast(evt); m_client->handleSpontaneousMessageReceived(Message(std::move(smre->message))); } break; case Event::PendingReplySuccess: maybeDispatchToPendingReply(&static_cast(evt)->reply); break; case Event::PendingReplyFailure: { PendingReplyFailureEvent *prfe = static_cast(evt); const auto it = m_pendingReplies.find(prfe->m_serial); if (it == m_pendingReplies.end()) { // not a disaster, but when it happens in debug mode I want to check it out assert(false); break; } PendingReplyPrivate *pendingPriv = it->second.asPendingReply(); m_pendingReplies.erase(it); pendingPriv->handleError(prfe->m_error); break; } case Event::PendingReplyCancel: // This comes from a secondary thread, which handles PendingReply notification itself. m_pendingReplies.erase(static_cast(evt)->serial); break; case Event::SecondaryConnectionConnect: { SecondaryConnectionConnectEvent *sce = static_cast(evt); const auto it = find_if(m_unredeemedCommRefs.begin(), m_unredeemedCommRefs.end(), [sce](const CommutexPeer &item) { return item.id() == sce->id; } ); assert(it != m_unredeemedCommRefs.end()); const auto emplaced = m_secondaryThreadLinks.emplace(sce->connection, std::move(*it)).first; m_unredeemedCommRefs.erase(it); // "welcome package" - it's done (only) as an event to avoid locking order issues CommutexLocker locker(&emplaced->second); if (locker.hasLock()) { UniqueNameReceivedEvent *evt = new UniqueNameReceivedEvent; evt->uniqueName = m_uniqueName; EventDispatcherPrivate::get(sce->connection->m_eventDispatcher) ->queueEvent(std::unique_ptr(evt)); } break; } case Event::SecondaryConnectionDisconnect: { SecondaryConnectionDisconnectEvent *sde = static_cast(evt); // delete our records to make sure we don't call into it in the future! const auto found = m_secondaryThreadLinks.find(sde->connection); if (found == m_secondaryThreadLinks.end()) { // looks like we've noticed the disappearance of the other thread earlier return; } m_secondaryThreadLinks.erase(found); discardPendingRepliesForSecondaryThread(sde->connection); break; } case Event::MainConnectionDisconnect: // since the main thread *sent* us the event, it already knows to drop all our PendingReplies m_mainThreadConnection = nullptr; cancelAllPendingReplies(); break; case Event::UniqueNameReceived: // We get this when the unique name became available after we were linked up with the main thread m_uniqueName = static_cast(evt)->uniqueName; break; } } Connection::CommRef Connection::createCommRef() { // TODO this is a good time to clean up "dead" CommRefs, where the counterpart was destroyed. CommRef ret; ret.connection = d; std::pair link = CommutexPeer::createLink(); { SpinLocker mainLocker(&d->m_lock); d->m_unredeemedCommRefs.emplace_back(std::move(link.first)); } ret.commutex = std::move(link.second); return ret; } bool Connection::supportsPassingFileDescriptors() const { return d->m_transport && d->m_transport->supportsPassingFileDescriptors(); } diff --git a/connection/connection.h b/connection/connection.h index d8c805e..0546477 100644 --- a/connection/connection.h +++ b/connection/connection.h @@ -1,107 +1,112 @@ /* 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/ */ #ifndef CONNECTION_H #define CONNECTION_H #include "commutex.h" #include "types.h" #include class ConnectAddress; class ConnectionPrivate; class Error; class EventDispatcher; class IMessageReceiver; +class ITransport; class Message; class PendingReply; +class Server; class DFERRY_EXPORT Connection { public: enum ThreadAffinity { MainConnection = 0, ThreadLocalConnection }; // Reference for passing to another thread; it guarantees that the target Connection // either exists or not, but is not currently being destroyed. Yes, the data is all private. class CommRef { friend class Connection; ConnectionPrivate *connection; CommutexPeer commutex; }; // for connecting to the session or system bus Connection(EventDispatcher *dispatcher, const ConnectAddress &connectAddress); // for reusing the connection of a Connection in another thread Connection(EventDispatcher *dispatcher, CommRef otherConnection); ~Connection(); Connection(Connection &other) = delete; Connection &operator=(Connection &other) = delete; void close(); CommRef createCommRef(); bool supportsPassingFileDescriptors() const; void setDefaultReplyTimeout(int msecs); int defaultReplyTimeout() const; enum TimeoutSpecialValues { DefaultTimeout = -1, NoTimeout = -2 }; // if a message expects no reply, that is not absolutely binding; this method allows to send a message that // does not expect (request) a reply, but we get it if it comes - not terribly useful in most cases // NOTE: this takes ownership of the message! The message will be deleted after sending in some future // event loop iteration, so it is guaranteed to stay valid before the next event loop iteration. PendingReply send(Message m, int timeoutMsecs = DefaultTimeout); // Mostly same as above. // This one ignores the reply, if any. Reports any locally detectable errors in the return value. Error sendNoReply(Message m); void waitForConnectionEstablished(); ConnectAddress connectAddress() const; std::string uniqueName() const; bool isConnected() const; EventDispatcher *eventDispatcher() const; // TODO matching patterns for subscription; note that a signal requires path, interface and // "method" (signal name) of sender void subscribeToSignal(); IMessageReceiver *spontaneousMessageReceiver() const; void setSpontaneousMessageReceiver(IMessageReceiver *receiver); private: + friend class Server; + Connection(ITransport *transport, const ConnectAddress &address); // called from Server + friend class ConnectionPrivate; ConnectionPrivate *d; }; #endif // CONNECTION_H diff --git a/connection/inewconnectionlistener.cpp b/connection/inewconnectionlistener.cpp new file mode 100644 index 0000000..b684a61 --- /dev/null +++ b/connection/inewconnectionlistener.cpp @@ -0,0 +1,28 @@ +/* + Copyright (C) 2017 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 "inewconnectionlistener.h" + +INewConnectionListener::~INewConnectionListener() +{ +} diff --git a/connection/inewconnectionlistener.h b/connection/inewconnectionlistener.h new file mode 100644 index 0000000..7728282 --- /dev/null +++ b/connection/inewconnectionlistener.h @@ -0,0 +1,40 @@ +/* + Copyright (C) 2017 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/ +*/ + +#ifndef INEWCONNECTIONLISTENER_H +#define INEWCONNECTIONLISTENER_H + +#include "export.h" + +class Server; + +class DFERRY_EXPORT INewConnectionListener +{ +public: + virtual ~INewConnectionListener(); + // This is called when a new client has connected and receives ownership of the Connection. + // Usually you want to call server->takeNextConnection() in a loop until it returns nullptr. + virtual void handleNewConnection(Server *server) = 0; +}; + +#endif // INEWCONNECTIONLISTENER_H diff --git a/connection/server.cpp b/connection/server.cpp new file mode 100644 index 0000000..741cafc --- /dev/null +++ b/connection/server.cpp @@ -0,0 +1,124 @@ +/* + Copyright (C) 2017 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 "server.h" + +#include "connectaddress.h" +#include "connection.h" +#include "icompletionlistener.h" +#include "inewconnectionlistener.h" +#include "iserver.h" +#include "itransport.h" + +#include + +#include + +class ServerPrivate : public ICompletionListener +{ +public: + void handleCompletion(void *transportServer) override; + + ConnectAddress listenAddress; + ConnectAddress concreteAddress; + Server *server; + INewConnectionListener *newConnectionListener; + IServer *transportServer; +}; + +Server::Server(EventDispatcher *dispatcher, const ConnectAddress &listenAddress) + : d(new ServerPrivate) +{ +#if 0 + if (ca.bus() == ConnectAddress::Bus::None || ca.socketType() == ConnectAddress::AddressType::None || + ca.role() == ConnectAddress::Role::None || + (ca.role() != ConnectAddress::Role::Server && ca.isServerOnly())) { + cerr << "\nConnection: connection constructor Exit A\n\n"; + return; + } +#endif + d->listenAddress = listenAddress; + d->server = this; + d->newConnectionListener = nullptr; + d->transportServer = IServer::create(listenAddress, &d->concreteAddress); + if (d->transportServer) { + d->transportServer->setEventDispatcher(dispatcher); + d->transportServer->setNewConnectionListener(d); + } +} + +Server::~Server() +{ + delete d->transportServer; + + delete d; + d = nullptr; +} + +void Server::setNewConnectionListener(INewConnectionListener *listener) +{ + d->newConnectionListener = listener; +} + +INewConnectionListener *Server::newConnectionListener() const +{ + return d->newConnectionListener; +} + +Connection *Server::takeNextClient() +{ + // TODO proper error handling / propagation + if (!d->transportServer) { + return nullptr; + } + ITransport *newTransport = d->transportServer->takeNextClient(); + if (!newTransport) { + return nullptr; + } + newTransport->setEventDispatcher(d->transportServer->eventDispatcher()); + return new Connection(newTransport, d->concreteAddress); +} + +bool Server::isListening() const +{ + return d->transportServer ? d->transportServer->isListening() : false; +} + +ConnectAddress Server::listenAddress() const +{ + return d->listenAddress; +} + +ConnectAddress Server::concreteAddress() const +{ + return d->concreteAddress; +} + +void ServerPrivate::handleCompletion(void *task) +{ + assert(task == transportServer); + (void) task; + if (newConnectionListener) { + newConnectionListener->handleNewConnection(server); + } +} diff --git a/transport/iserver.h b/connection/server.h similarity index 52% copy from transport/iserver.h copy to connection/server.h index 64f521c..9afe5ce 100644 --- a/transport/iserver.h +++ b/connection/server.h @@ -1,67 +1,59 @@ /* 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/ */ -#ifndef ISERVER_H -#define ISERVER_H +#ifndef SERVER_H +#define SERVER_H -#include "iioeventlistener.h" -#include "platform.h" -#include "types.h" - -#include +#include "export.h" +class ServerPrivate; class ConnectAddress; +class Connection; +class Error; class EventDispatcher; -class ITransport; -class ICompletionListener; +class INewConnectionListener; -class IServer : public IioEventListener +class DFERRY_EXPORT Server { public: - IServer(); // TODO event dispatcher as constructor argument? - ~IServer() override; - - virtual bool isListening() const = 0; - - void setNewConnectionListener(ICompletionListener *listener); // notified once on every new connection - - ITransport *takeNextClient(); - virtual void close() = 0; - - void setEventDispatcher(EventDispatcher *ed) override; - EventDispatcher *eventDispatcher() const override; + Server(EventDispatcher *dispatcher, const ConnectAddress &listenAddress); + ~Server(); - static IServer *create(const ConnectAddress &connectAddress); + Connection *takeNextClient(); -protected: - friend class EventDispatcher; - // handleCanRead() and handleCanWrite() from IioEventListener stay pure virtual + bool isListening() const; + // The listenAddress passed in + ConnectAddress listenAddress() const; + // The address clients can connect to, which may be (usually is...) different from serverAddress + ConnectAddress concreteAddress() const; + Error error() const; // TODO - std::deque m_incomingConnections; - ICompletionListener *m_newConnectionListener; + void setNewConnectionListener(INewConnectionListener *listener); + INewConnectionListener *newConnectionListener() const; private: - EventDispatcher *m_eventDispatcher; + friend class ServerPrivate; + ServerPrivate *d; }; -#endif // ISERVER_H +#endif // SERVER_H diff --git a/transport/iserver.cpp b/transport/iserver.cpp index 03dfbc5..dc2499a 100644 --- a/transport/iserver.cpp +++ b/transport/iserver.cpp @@ -1,107 +1,181 @@ /* 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 "iserver.h" #include "connectaddress.h" #include "eventdispatcher_p.h" #include "itransport.h" #include "ipserver.h" #ifdef __unix__ #include "localserver.h" #endif #include +#ifdef __unix__ + +#include +#include "stringtools.h" + +static std::string randomDbusSocketName() +{ + std::random_device rd; + std::mt19937 mt(rd()); + char randomData[16]; + // OK dead code elimination, show us what you can! + if (sizeof(size_t) >= 8) { + std::uniform_int_distribution dist; + for (size_t i = 0; i < (sizeof(randomData) / sizeof(uint64)); i++) { + reinterpret_cast(randomData)[i] = dist(mt); + } + } else { + std::uniform_int_distribution dist; + for (size_t i = 0; i < (sizeof(randomData) / sizeof(uint32)); i++) { + reinterpret_cast(randomData)[i] = dist(mt); + } + } + // Good that std::string knows nothing about valid utf-8 encoding! + const std::string pseudoString(randomData, sizeof(randomData)); + return std::string("/dbus-") + hexEncode(pseudoString); +} + +static std::string xdgRuntimeDir() +{ + return std::string(getenv("XDG_RUNTIME_DIR")); +} +#endif + IServer::IServer() : m_newConnectionListener(nullptr), m_eventDispatcher(nullptr) { } IServer::~IServer() { for (ITransport *c : m_incomingConnections) { delete c; } } //static -IServer *IServer::create(const ConnectAddress &ca) +IServer *IServer::create(const ConnectAddress &listenAddr, ConnectAddress *concreteAddr) { - if (ca.role() != ConnectAddress::Role::PeerServer) { + if (listenAddr.role() != ConnectAddress::Role::PeerServer) { return nullptr; } - switch (ca.type()) { +#ifdef __unix__ + bool isLocalSocket = true; + bool isAbstract = false; + std::string unixSocketPath; +#endif + + switch (listenAddr.type()) { #ifdef __unix__ case ConnectAddress::Type::UnixPath: - return new LocalServer(ca.path()); + unixSocketPath = listenAddr.path(); + break; + case ConnectAddress::Type::UnixDir: + unixSocketPath = listenAddr.path() + randomDbusSocketName(); + break; + case ConnectAddress::Type::RuntimeDir: + unixSocketPath = xdgRuntimeDir() + randomDbusSocketName(); + break; + case ConnectAddress::Type::TmpDir: + unixSocketPath = listenAddr.path() + randomDbusSocketName(); +#ifdef __linux__ + isAbstract = true; +#endif + break; +#ifdef __linux__ case ConnectAddress::Type::AbstractUnixPath: - return new LocalServer(std::string(1, '\0') + ca.path()); + unixSocketPath = listenAddr.path(); + isAbstract = true; + break; +#endif #endif case ConnectAddress::Type::Tcp: case ConnectAddress::Type::Tcp4: case ConnectAddress::Type::Tcp6: - return new IpServer(ca); +#ifdef __unix__ + isLocalSocket = false; +#endif + break; default: return nullptr; } + + *concreteAddr = listenAddr; + +#ifdef __unix__ + if (isLocalSocket) { + concreteAddr->setType(isAbstract ? ConnectAddress::Type::AbstractUnixPath + : ConnectAddress::Type::UnixPath); + concreteAddr->setPath(unixSocketPath); + if (isAbstract) { + unixSocketPath.insert(0, 1, '\0'); + } + return new LocalServer(unixSocketPath); + } else +#endif + return new IpServer(listenAddr); } ITransport *IServer::takeNextClient() { if (m_incomingConnections.empty()) { return nullptr; } ITransport *ret = m_incomingConnections.front(); m_incomingConnections.pop_front(); return ret; } void IServer::setNewConnectionListener(ICompletionListener *listener) { m_newConnectionListener = listener; } void IServer::setEventDispatcher(EventDispatcher *ed) { if (m_eventDispatcher == ed) { return; } if (m_eventDispatcher) { EventDispatcherPrivate *const ep = EventDispatcherPrivate::get(m_eventDispatcher); ep->removeIoEventListener(this); } m_eventDispatcher = ed; if (m_eventDispatcher) { EventDispatcherPrivate *const ep = EventDispatcherPrivate::get(m_eventDispatcher); ep->addIoEventListener(this); ep->setReadWriteInterest(this, true, false); } } EventDispatcher *IServer::eventDispatcher() const { return m_eventDispatcher; } diff --git a/transport/iserver.h b/transport/iserver.h index 64f521c..3350d8e 100644 --- a/transport/iserver.h +++ b/transport/iserver.h @@ -1,67 +1,70 @@ /* 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/ */ #ifndef ISERVER_H #define ISERVER_H #include "iioeventlistener.h" #include "platform.h" #include "types.h" #include class ConnectAddress; class EventDispatcher; class ITransport; class ICompletionListener; class IServer : public IioEventListener { public: IServer(); // TODO event dispatcher as constructor argument? ~IServer() override; virtual bool isListening() const = 0; void setNewConnectionListener(ICompletionListener *listener); // notified once on every new connection ITransport *takeNextClient(); virtual void close() = 0; void setEventDispatcher(EventDispatcher *ed) override; EventDispatcher *eventDispatcher() const override; - static IServer *create(const ConnectAddress &connectAddress); + // listenAddress may be a concrete address (in which case *concreteAddress will be set to a copy of it) + // or it may be a "listen-only address", which is an underspecified or wildcard address. In the latter + // case, *concreteAddress will be set to a concrete address generated according to listenAddress. + static IServer *create(const ConnectAddress &listenAddress, ConnectAddress *concreteAddress); protected: friend class EventDispatcher; // handleCanRead() and handleCanWrite() from IioEventListener stay pure virtual std::deque m_incomingConnections; ICompletionListener *m_newConnectionListener; private: EventDispatcher *m_eventDispatcher; }; #endif // ISERVER_H diff --git a/transport/localserver.cpp b/transport/localserver.cpp index 6811105..3effb97 100644 --- a/transport/localserver.cpp +++ b/transport/localserver.cpp @@ -1,110 +1,110 @@ /* 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 "localserver.h" #include "icompletionlistener.h" #include "localsocket.h" #include #include #include #include #include #include LocalServer::LocalServer(const std::string &socketFilePath) : m_listenFd(-1) { const int fd = socket(PF_UNIX, SOCK_STREAM, 0); if (fd < 0) { return; } // don't let forks inherit the file descriptor - just in case fcntl(fd, F_SETFD, FD_CLOEXEC); struct sockaddr_un addr; addr.sun_family = PF_UNIX; bool ok = socketFilePath.length() + 1 <= sizeof(addr.sun_path); if (ok) { memcpy(addr.sun_path, socketFilePath.c_str(), socketFilePath.length() + 1); } if (!socketFilePath.empty() && socketFilePath[0] != '\0') { // not a so-called abstract socket (weird but useful Linux specialty) unlink(socketFilePath.c_str()); } ok = ok && (bind(fd, (struct sockaddr *)&addr, sizeof(sa_family_t) + socketFilePath.length()) == 0); ok = ok && (::listen(fd, /* max queued incoming connections */ 64) == 0); if (ok) { m_listenFd = fd; } else { ::close(fd); } } LocalServer::~LocalServer() { close(); } void LocalServer::handleCanRead() { - setEventDispatcher(nullptr); int connFd = accept(m_listenFd, nullptr, nullptr); if (connFd < 0) { return; } fcntl(connFd, F_SETFD, FD_CLOEXEC); m_incomingConnections.push_back(new LocalSocket(connFd)); if (m_newConnectionListener) { m_newConnectionListener->handleCompletion(this); } } void LocalServer::handleCanWrite() { // We never registered this to be called, so... assert(false); } bool LocalServer::isListening() const { return m_listenFd >= 0; } void LocalServer::close() { + setEventDispatcher(nullptr); if (m_listenFd >= 0) { ::close(m_listenFd); m_listenFd = -1; } } FileDescriptor LocalServer::fileDescriptor() const { return m_listenFd; }