/* */ #include "DHTMessageDispatcherImpl.h" #include "DHTMessage.h" #include "DHTMessageCallback.h" #include "DHTMessageEntry.h" #include "DHTMessageTracker.h" #include "RecoverableException.h" #include "LogFactory.h" #include "Logger.h" #include "DHTConstants.h" #include "StringFormat.h" #include "DHTNode.h" namespace aria2 { DHTMessageDispatcherImpl::DHTMessageDispatcherImpl(const SharedHandle& tracker): _tracker(tracker), _logger(LogFactory::getInstance()) {} DHTMessageDispatcherImpl::~DHTMessageDispatcherImpl() {} void DHTMessageDispatcherImpl::addMessageToQueue(const SharedHandle& message, time_t timeout, const SharedHandle& callback) { SharedHandle e(new DHTMessageEntry(message, timeout, callback)); _messageQueue.push_back(e); } void DHTMessageDispatcherImpl::addMessageToQueue(const SharedHandle& message, const SharedHandle& callback) { addMessageToQueue(message, DHT_MESSAGE_TIMEOUT, callback); } bool DHTMessageDispatcherImpl::sendMessage(const SharedHandle& entry) { try { if(entry->_message->send()) { if(!entry->_message->isReply()) { _tracker->addMessage(entry->_message, entry->_timeout, entry->_callback); } _logger->info("Message sent: %s", entry->_message->toString().c_str()); } else { return false; } } catch(RecoverableException& e) { _logger->error("Failed to send message: %s", e, entry->_message->toString().c_str()); // Add message to DHTMessageTracker with timeout 0 to treat it as // time out. Without this, we have untracked message and some of // DHTTask(such as DHTAbstractNodeLookupTask) don't finish // forever. if(!entry->_message->isReply()) { _tracker->addMessage(entry->_message, 0, entry->_callback); } } return true; } void DHTMessageDispatcherImpl::sendMessages() { // TODO I can't use bind1st and mem_fun here because bind1st cannot bind a // function which takes a reference as an argument.. std::deque >::iterator itr = _messageQueue.begin(); for(; itr != _messageQueue.end(); ++itr) { if(!sendMessage(*itr)) { break; } } _messageQueue.erase(_messageQueue.begin(), itr); _logger->debug("%lu dht messages remaining in the queue.", static_cast(_messageQueue.size())); } size_t DHTMessageDispatcherImpl::countMessageInQueue() const { return _messageQueue.size(); } } // namespace aria2