/* */ #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), _timeout(DHT_MESSAGE_TIMEOUT), _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, _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); } if(_logger->info()) { _logger->info("Message sent: %s", entry->message->toString().c_str()); } } else { return false; } } catch(RecoverableException& e) { _logger->info("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); if(_logger->debug()) { _logger->debug("%lu dht messages remaining in the queue.", static_cast(_messageQueue.size())); } } size_t DHTMessageDispatcherImpl::countMessageInQueue() const { return _messageQueue.size(); } } // namespace aria2