diff options
| -rw-r--r-- | src/packetSource.cpp | 6 | ||||
| -rw-r--r-- | src/syncServer.cpp | 62 | ||||
| -rw-r--r-- | src/syncServer.h | 13 | ||||
| -rw-r--r-- | src/syncTcpConnection.cpp | 8 |
4 files changed, 65 insertions, 24 deletions
diff --git a/src/packetSource.cpp b/src/packetSource.cpp index fa4abd0..06a7500 100644 --- a/src/packetSource.cpp +++ b/src/packetSource.cpp @@ -79,10 +79,8 @@ void UDPPacketSource::onResolve(PacketSourceResolverIt& it) sock.sock_->open(e.protocol()); #ifndef _MSC_VER - if(e.protocol() == proto::v6()) { - boost::asio::ip::v6_only option(true); - sock.sock_->set_option(option); - } + if(e.protocol() == proto::v6()) + sock.sock_->set_option(boost::asio::ip::v6_only(true)); #endif sock.sock_->bind(e); sockets_.push_back(sock); diff --git a/src/syncServer.cpp b/src/syncServer.cpp index 893ceba..130d5b6 100644 --- a/src/syncServer.cpp +++ b/src/syncServer.cpp @@ -33,26 +33,54 @@ #include "syncServer.h" #include "resolver.h" #include "log.h" +#include "anytunError.h" //using asio::ip::tcp; SyncServer::SyncServer(std::string localaddr, std::string port, ConnectCallback onConnect) - : acceptor_(io_service_), onConnect_(onConnect) + : onConnect_(onConnect) { gResolver.resolveTcp(localaddr, port, boost::bind(&SyncServer::onResolve, this, _1), boost::bind(&SyncServer::onResolvError, this, _1)); } +SyncServer::~SyncServer() +{ + std::list<AcceptorsElement>::iterator it = acceptors_.begin(); + for(;it != acceptors_.end(); ++it) { +/// this might be a needed by a running thread, TODO cleanup + delete(it->acceptor_); + } +} + void SyncServer::onResolve(SyncTcpConnection::proto::resolver::iterator& it) { - SyncTcpConnection::proto::endpoint e = *it; + while(it != SyncTcpConnection::proto::resolver::iterator()) { + SyncTcpConnection::proto::endpoint e = *it; + + AcceptorsElement acceptor; + acceptor.acceptor_ = new SyncTcpConnection::proto::acceptor(io_service_); + if(!acceptor.acceptor_) + AnytunError::throwErr() << "memory error"; + + acceptor.acceptor_->open(e.protocol()); +#ifndef _MSC_VER + if(e.protocol() == boost::asio::ip::tcp::v6()) + acceptor.acceptor_->set_option(boost::asio::ip::v6_only(true)); +#endif + acceptor.acceptor_->set_option(boost::asio::socket_base::reuse_address(true)); + acceptor.acceptor_->bind(e); + acceptor.acceptor_->listen(); + acceptor.started_ = false; + + acceptors_.push_back(acceptor); + + cLog.msg(Log::PRIO_NOTICE) << "sync server listening on " << e; + + it++; + } - acceptor_.open(e.protocol()); - acceptor_.set_option(boost::asio::socket_base::reuse_address(true)); - acceptor_.bind(e); - acceptor_.listen(); start_accept(); ready_sem_.up(); - cLog.msg(Log::PRIO_NOTICE) << "sync server listening on " << e; } void SyncServer::onResolvError(const std::runtime_error& e) @@ -77,17 +105,27 @@ void SyncServer::send(std::string message) void SyncServer::start_accept() { Lock lock(mutex_); - SyncTcpConnection::pointer new_connection = SyncTcpConnection::create(acceptor_.io_service()); - conns_.push_back(new_connection); - acceptor_.async_accept(new_connection->socket(), - boost::bind(&SyncServer::handle_accept, this, new_connection, boost::asio::placeholders::error)); + + std::list<AcceptorsElement>::iterator it = acceptors_.begin(); + for(;it != acceptors_.end(); ++it) { + if(!it->started_) { + SyncTcpConnection::pointer new_connection = SyncTcpConnection::create(it->acceptor_->io_service()); + conns_.push_back(new_connection); + it->acceptor_->async_accept(new_connection->socket(), + boost::bind(&SyncServer::handle_accept, this, new_connection, boost::asio::placeholders::error, it)); + it->started_ = true; + } + } } -void SyncServer::handle_accept(SyncTcpConnection::pointer new_connection, const boost::system::error_code& error) +void SyncServer::handle_accept(SyncTcpConnection::pointer new_connection, const boost::system::error_code& error, std::list<AcceptorsElement>::iterator it) { if (!error) { + cLog.msg(Log::PRIO_INFO) << "new sync client connected from " << new_connection->socket().remote_endpoint(); + new_connection->onConnect = onConnect_; new_connection->start(); + it->started_ = false; start_accept(); } } diff --git a/src/syncServer.h b/src/syncServer.h index 3157d15..032e975 100644 --- a/src/syncServer.h +++ b/src/syncServer.h @@ -50,6 +50,7 @@ class SyncServer { public: SyncServer(std::string localaddr, std::string port, ConnectCallback onConnect); + ~SyncServer(); void onResolve(SyncTcpConnection::proto::resolver::iterator& it); void onResolvError(const std::runtime_error& e); @@ -59,14 +60,18 @@ public: std::list<SyncTcpConnection::pointer> conns_; private: - void start_accept(); - void handle_accept(SyncTcpConnection::pointer new_connection, const boost::system::error_code& error); - Mutex mutex_; //Mutex for list conns_ boost::asio::io_service io_service_; - SyncTcpConnection::proto::acceptor acceptor_; + typedef struct { + SyncTcpConnection::proto::acceptor* acceptor_; + bool started_; + } AcceptorsElement; + std::list<AcceptorsElement> acceptors_; ConnectCallback onConnect_; Semaphore ready_sem_; + + void start_accept(); + void handle_accept(SyncTcpConnection::pointer new_connection, const boost::system::error_code& error, std::list<AcceptorsElement>::iterator it); }; #endif diff --git a/src/syncTcpConnection.cpp b/src/syncTcpConnection.cpp index d243633..08875c7 100644 --- a/src/syncTcpConnection.cpp +++ b/src/syncTcpConnection.cpp @@ -48,10 +48,10 @@ void SyncTcpConnection::start() void SyncTcpConnection::Send(std::string message) { - boost::asio::async_write(socket_, boost::asio::buffer(message), - boost::bind(&SyncTcpConnection::handle_write, shared_from_this(), - boost::asio::placeholders::error, - boost::asio::placeholders::bytes_transferred)); + boost::asio::async_write(socket_, boost::asio::buffer(message), + boost::bind(&SyncTcpConnection::handle_write, shared_from_this(), + boost::asio::placeholders::error, + boost::asio::placeholders::bytes_transferred)); } SyncTcpConnection::SyncTcpConnection(boost::asio::io_service& io_service) : socket_(io_service) |