Client::Client(epoll::EventPoll& eventPoll, const std::string& socketPath)
: mEventPoll(eventPoll),
- mProcessor("[CLIENT] "),
+ mProcessor(eventPoll, "[CLIENT] "),
mSocketPath(socketPath)
{
LOGS("Client Constructor");
return;
}
LOGS("Client start");
- // Initialize the connection with the server
- auto handleEvent = [&](int, epoll::Events) {
- mProcessor.handleEvent();
- };
- mEventPoll.addFD(mProcessor.getEventFD(), EPOLLIN, handleEvent);
+
mProcessor.start();
LOGD("Connecting to " + mSocketPath);
}
LOGS("Client stop");
mProcessor.stop();
-
- mEventPoll.removeFD(mProcessor.getEventFD());
}
void Client::handle(const FileDescriptor fd, const epoll::Events pollEvents)
std::lock_guard<Mutex> lock(mMutex);
if (mCallbacks.find(fd) != mCallbacks.end()) {
- LOGW("Already added fd: " << fd);
+ LOGE("Already added fd: " << fd);
throw UtilsException("FD already added");
}
if (!addFDInternal(fd, events)) {
+ LOGE("Could not add fd");
throw UtilsException("Could not add fd");
}
{
// No need to lock and check mCallbacks map
if (!modifyFDInternal(fd, events)) {
+ LOGE("Could not modify fd: " << fd);
throw UtilsException("Could not modify fd");
}
}
#include "ipc/internals/acceptor.hpp"
#include "logger/logger.hpp"
+#include <functional>
+
namespace ipc {
-Acceptor::Acceptor(const std::string& socketPath, const NewConnectionCallback& newConnectionCallback)
- : mNewConnectionCallback(newConnectionCallback),
+Acceptor::Acceptor(epoll::EventPoll& eventPoll,
+ const std::string& socketPath,
+ const NewConnectionCallback& newConnectionCallback)
+ : mEventPoll(eventPoll),
+ mNewConnectionCallback(newConnectionCallback),
mSocket(Socket::createSocket(socketPath))
{
LOGT("Creating Acceptor for socket " << socketPath);
+ mEventPoll.addFD(mSocket.getFD(), EPOLLIN, std::bind(&Acceptor::handleConnection, this));
}
Acceptor::~Acceptor()
{
LOGT("Destroyed Acceptor");
+ mEventPoll.removeFD(mSocket.getFD());
}
void Acceptor::handleConnection()
mNewConnectionCallback(tmpSocket);
}
-FileDescriptor Acceptor::getConnectionFD()
-{
- return mSocket.getFD();
-}
-
} // namespace ipc
#include "config.hpp"
#include "ipc/internals/socket.hpp"
+#include "ipc/epoll/event-poll.hpp"
#include "ipc/types.hpp"
#include <string>
/**
* Class for accepting new connections.
*
+ * @param eventPoll dispatcher
* @param socketPath path to the socket
* @param newConnectionCallback called on new connections
*/
- Acceptor(const std::string& socketPath,
+ Acceptor(epoll::EventPoll& eventPoll,
+ const std::string& socketPath,
const NewConnectionCallback& newConnectionCallback);
~Acceptor();
Acceptor(const Acceptor& acceptor) = delete;
Acceptor& operator=(const Acceptor&) = delete;
+private:
+ epoll::EventPoll& mEventPoll;
+ NewConnectionCallback mNewConnectionCallback;
+ Socket mSocket;
+
/**
* Handle one incoming connection.
* Used with external polling
*/
void handleConnection();
-
- /**
- * @return file descriptor for the connection socket
- */
- FileDescriptor getConnectionFD();
-
-private:
- NewConnectionCallback mNewConnectionCallback;
- Socket mSocket;
};
} // namespace ipc
const MethodID Processor::REGISTER_SIGNAL_METHOD_ID = std::numeric_limits<MethodID>::max() - 1;
const MethodID Processor::ERROR_METHOD_ID = std::numeric_limits<MethodID>::max() - 2;
-Processor::Processor(const std::string& logName,
+Processor::Processor(epoll::EventPoll& eventPoll,
+ const std::string& logName,
const PeerCallback& newPeerCallback,
const PeerCallback& removedPeerCallback,
const unsigned int maxNumberOfPeers)
- : mLogPrefix(logName),
+ : mEventPoll(eventPoll),
+ mLogPrefix(logName),
mIsRunning(false),
mNewPeerCallback(newPeerCallback),
mRemovedPeerCallback(removedPeerCallback),
if (!mIsRunning) {
LOGI(mLogPrefix + "Processor start");
mIsRunning = true;
+
+ mEventPoll.addFD(mRequestQueue.getFD(), EPOLLIN, std::bind(&Processor::handleEvent, this));
}
}
std::make_exception_ptr(IPCClosingException()));
}
+ mEventPoll.removeFD(mRequestQueue.getFD());
mIsRunning = false;
-
requestFinisher.conditionPtr->notify_all();
+
return true;
}
#include "ipc/internals/remove-peer-request.hpp"
#include "ipc/internals/send-result-request.hpp"
#include "ipc/internals/finish-request.hpp"
+#include "ipc/epoll/event-poll.hpp"
#include "ipc/exception.hpp"
#include "ipc/method-result.hpp"
#include "ipc/types.hpp"
* @param newPeerCallback called when a new peer arrives
* @param removedPeerCallback called when the Processor stops listening for this peer
*/
- Processor(const std::string& logName = "",
+ Processor(epoll::EventPoll& eventPoll,
+ const std::string& logName = "",
const PeerCallback& newPeerCallback = nullptr,
const PeerCallback& removedPeerCallback = nullptr,
const unsigned int maxNumberOfPeers = DEFAULT_MAX_NUMBER_OF_PEERS);
std::shared_ptr<Socket> socketPtr;
};
+ epoll::EventPoll& mEventPoll;
+
typedef std::vector<PeerInfo> Peers;
std::string mLogPrefix;
const PeerCallback& addPeerCallback,
const PeerCallback& removePeerCallback)
: mEventPoll(eventPoll),
- mProcessor("[SERVICE] "),
- mAcceptor(socketPath, std::bind(&Processor::addPeer, &mProcessor, _1))
+ mProcessor(eventPoll, "[SERVICE] "),
+ mAcceptor(eventPoll, socketPath, std::bind(&Processor::addPeer, &mProcessor, _1))
{
LOGS("Service Constructor");
return;
}
LOGS("Service start");
- auto handleConnection = [&](int, epoll::Events) {
- mAcceptor.handleConnection();
- };
- auto handleProcessorEvent = [&](int, epoll::Events) {
- mProcessor.handleEvent();
- };
- mEventPoll.addFD(mProcessor.getEventFD(), EPOLLIN, handleProcessorEvent);
+
mProcessor.start();
- mEventPoll.addFD(mAcceptor.getConnectionFD(), EPOLLIN, handleConnection);
+
}
bool Service::isStarted()
return;
}
LOGS("Service stop");
- mEventPoll.removeFD(mAcceptor.getConnectionFD());
mProcessor.stop();
- mEventPoll.removeFD(mProcessor.getEventFD());
}
void Service::handle(const FileDescriptor fd, const epoll::Events pollEvents)
LOGS("Service handle");
if (!isStarted()) {
- LOGW("Service stopped");
+ LOGW("Service stopped, but got event: " << pollEvents << " on fd: " << fd);
return;
}