/* */ #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 "fmt.h" #include "DHTNode.h" namespace aria2 { DHTMessageDispatcherImpl::DHTMessageDispatcherImpl (const SharedHandle& tracker) : tracker_(tracker), timeout_(DHT_MESSAGE_TIMEOUT) {} 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); } A2_LOG_INFO(fmt("Message sent: %s", entry->message->toString().c_str())); } else { return false; } } catch(RecoverableException& e) { A2_LOG_INFO_EX(fmt("Failed to send message: %s", entry->message->toString().c_str()), e); // 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); A2_LOG_DEBUG(fmt("%lu dht messages remaining in the queue.", static_cast(messageQueue_.size()))); } size_t DHTMessageDispatcherImpl::countMessageInQueue() const { return messageQueue_.size(); } } // namespace aria2