* Enforce an http transfer timeout when libcurl fails to honor

it. Also set a 5-minute timeout for (previously unlimited) torrent
transfers and fixes the argument type for curl_easy_setopt values.

* Allows bandwidth throttles to work without floating point support.

* Adds a d.add_peer=host[:port] command to manually add a peer (not
for torrents marked "private"), port is 6881 by default.

* Allows banning the selected peer with "B". No unbanning is possible
yet.

All the above patches were written by Josef Drexler.

* Differentiate between commands that have no target, and those that
take generic targets, when using XMLRPC.


git-svn-id: svn://rakshasa.no/libtorrent/trunk/rtorrent@1073 e378c898-3ddf-0310-93e7-cc216c733640

src/command_download.cc

@@ -41,9 +41,11 @@
 #include <rak/file_stat.h>
 #include <rak/error_number.h>
 #include <rak/path.h>
+#include <rak/socket_address.h>
 #include <rak/string_manip.h>
 #include <torrent/rate.h>
 #include <torrent/tracker.h>
+#include <torrent/connection_manager.h>
 #include <torrent/data/file.h>
 #include <torrent/data/file_list.h>
 #include <torrent/peer/connection_list.h>
@@ -261,6 +263,42 @@ cmd_d_initialize_logs(core::Download* download) {
   return torrent::Object();
 }
 
+struct call_add_d_peer_t {
+  call_add_d_peer_t(core::Download* d, int port) : m_download(d), m_port(port) { }
+
+  void operator() (const sockaddr* sa, int err) {
+    if (sa == NULL)
+      control->core()->push_log("Could not resolve host.");
+    else
+      m_download->download()->add_peer(sa, m_port);
+  }
+
+  core::Download* m_download;
+  int m_port;
+};
+
+void
+apply_d_add_peer(core::Download* download, const std::string& arg) {
+  int port, ret;
+  char dummy;
+  char host[1024];
+
+  if (download->download()->is_private())
+    throw torrent::input_error("Download is private.");
+
+  ret = std::sscanf(arg.c_str(), "%1023[^:]:%i%c", host, &port, &dummy);
+
+  if (ret == 1)
+    port = 6881;
+  else if (ret != 2)
+    throw torrent::input_error("Could not parse host.");
+
+  if (port < 1 || port > 65535)
+    throw torrent::input_error("Invalid port number.");
+
+  torrent::connection_manager()->resolver()(host, (int)rak::socket_address::pf_inet, SOCK_STREAM, call_add_d_peer_t(download, port));
+}
+
 torrent::Object
 f_multicall(core::Download* download, const torrent::Object& rawArgs) {
   const torrent::Object::list_type& args = rawArgs.as_list();
@@ -368,6 +406,9 @@ p_multicall(core::Download* download, const torrent::Object& rawArgs) {
 #define ADD_CD_LIST(key, slot) \
   ADD_CD_SLOT_PUBLIC("d." key, call_list, slot, "i:", "")
 
+#define ADD_CD_STRING(key, slot) \
+  ADD_CD_SLOT_PUBLIC("d." key, call_string, rpc::object_string_fn<core::Download*>(slot), "i:s", "")
+
 #define ADD_CD_VARIABLE_VALUE(key, firstKey, secondKey) \
   ADD_CD_SLOT_PUBLIC("d.get_" key, call_unknown, rpc::get_variable_d_fn(firstKey, secondKey), "i:", ""); \
   ADD_CD_SLOT       ("d.set_" key, call_value,   rpc::set_variable_d_fn(firstKey, secondKey), "i:i", "");
@@ -449,6 +490,8 @@ initialize_command_download() {
 
   ADD_CD_F_VOID("update_priorities", rak::on(std::mem_fun(&core::Download::download), std::mem_fun(&torrent::Download::update_priorities)));
 
+  ADD_CD_STRING("add_peer",        std::ptr_fun(&apply_d_add_peer));
+
   ADD_CD_VALUE("is_open",          rak::on(std::mem_fun(&core::Download::download), std::mem_fun(&torrent::Download::is_open)));
   ADD_CD_VALUE("is_active",        rak::on(std::mem_fun(&core::Download::download), std::mem_fun(&torrent::Download::is_active)));
   ADD_CD_VALUE("is_hash_checked",  rak::on(std::mem_fun(&core::Download::download), std::mem_fun(&torrent::Download::is_hash_checked)));

src/command_dynamic.cc

@@ -185,11 +185,30 @@ system_method_has_key(__UNUSED rpc::target_type target, const torrent::Object& r
   return torrent::Object((int64_t)(function->find(args.back().as_string().c_str()) != function->end()));
 }
 
+torrent::Object
+system_method_list_keys(__UNUSED rpc::target_type target, const torrent::Object& rawArgs) {
+  rpc::CommandFunctionList* function;
+  rpc::CommandMap::iterator itr = rpc::commands.find(rawArgs.as_string().c_str());
+
+  if (itr == rpc::commands.end() ||
+      (function = dynamic_cast<rpc::CommandFunctionList*>(itr->second.m_variable)) == NULL)
+    throw torrent::input_error("Command is the wrong type.");
+
+  torrent::Object rawResult = torrent::Object::create_list();
+  torrent::Object::list_type& result = rawResult.as_list();
+
+  for (rpc::CommandFunctionList::const_iterator itr = function->begin(), last = function->end(); itr != last; itr++)
+    result.push_back(itr->first);
+
+  return rawResult;
+}
+
 void
 initialize_command_dynamic() {
-  CMD_G("system.method.insert",       rak::ptr_fn(&system_method_insert));
-  CMD_G_STRING("system.method.erase", rak::ptr_fn(&system_method_erase));
-  CMD_G("system.method.set",          rak::ptr_fn(&system_method_set));
-  CMD_G("system.method.set_key",      rak::ptr_fn(&system_method_set_key));
-  CMD_G("system.method.has_key",      rak::ptr_fn(&system_method_has_key));
+  CMD_N       ("system.method.insert",    rak::ptr_fn(&system_method_insert));
+  CMD_N_STRING("system.method.erase",     rak::ptr_fn(&system_method_erase));
+  CMD_N       ("system.method.set",       rak::ptr_fn(&system_method_set));
+  CMD_N       ("system.method.set_key",   rak::ptr_fn(&system_method_set_key));
+  CMD_N       ("system.method.has_key",   rak::ptr_fn(&system_method_has_key));
+  CMD_N_STRING("system.method.list_keys", rak::ptr_fn(&system_method_list_keys));
 }

src/command_helpers.h

@@ -193,6 +193,17 @@ add_variable(key, NULL, NULL, &rpc::CommandVariable::get_string, NULL, std::stri
 #define CMD_G_STRING(key, slot) \
   CMD_G_SLOT(key, call_string, slot, "i:", "")
 
+#define CMD_N_SLOT(key, function, slot, parm, doc)    \
+  commandAnySlotsItr->set_slot(slot); \
+  rpc::commands.insert_type(key, commandAnySlotsItr++, &rpc::CommandSlot<rpc::target_type>::function, \
+                            rpc::CommandMap::flag_dont_delete | rpc::CommandMap::flag_no_target | rpc::CommandMap::flag_public_xmlrpc, parm, doc);
+
+#define CMD_N(key, slot) \
+  CMD_N_SLOT(key, call_unknown, slot, "i:", "")
+
+#define CMD_N_STRING(key, slot) \
+  CMD_N_SLOT(key, call_string, slot, "i:", "")
+
 #define CMD_D_SLOT(key, function, slot, parm, doc)    \
   commandDownloadSlotsItr->set_slot(slot); \
   rpc::commands.insert_type(key, commandDownloadSlotsItr++, &rpc::CommandSlot<core::Download*>::function, rpc::CommandMap::flag_dont_delete | rpc::CommandMap::flag_public_xmlrpc, parm, doc);

src/core/curl_get.cc

@@ -41,6 +41,7 @@
 #include <curl/easy.h>
 #include <torrent/exceptions.h>
 
+#include "globals.h"
 #include "curl_get.h"
 #include "curl_stack.h"
 
@@ -73,14 +74,20 @@ CurlGet::start() {
   curl_easy_setopt(m_handle, CURLOPT_WRITEDATA,      this);
 
   if (m_timeout != 0) {
-    curl_easy_setopt(m_handle, CURLOPT_CONNECTTIMEOUT, 60);
-    curl_easy_setopt(m_handle, CURLOPT_TIMEOUT,        m_timeout);
+    curl_easy_setopt(m_handle, CURLOPT_CONNECTTIMEOUT, (long)60);
+    curl_easy_setopt(m_handle, CURLOPT_TIMEOUT,        (long)m_timeout);
+
+    // Normally libcurl should handle the timeout. But sometimes that doesn't
+    // work right so we do a fallback timeout that just aborts the transfer.
+    m_taskTimeout.set_slot(rak::mem_fn(this, &CurlGet::receive_timeout));
+    priority_queue_erase(&taskScheduler, &m_taskTimeout);
+    priority_queue_insert(&taskScheduler, &m_taskTimeout, cachedTime + rak::timer::from_seconds(m_timeout + 5));
   }
 
-  curl_easy_setopt(m_handle, CURLOPT_FORBID_REUSE,   1);
-  curl_easy_setopt(m_handle, CURLOPT_NOSIGNAL,       1);
-  curl_easy_setopt(m_handle, CURLOPT_FOLLOWLOCATION, 1);
-  curl_easy_setopt(m_handle, CURLOPT_MAXREDIRS,      5);
+  curl_easy_setopt(m_handle, CURLOPT_FORBID_REUSE,   (long)1);
+  curl_easy_setopt(m_handle, CURLOPT_NOSIGNAL,       (long)1);
+  curl_easy_setopt(m_handle, CURLOPT_FOLLOWLOCATION, (long)1);
+  curl_easy_setopt(m_handle, CURLOPT_MAXREDIRS,      (long)5);
   curl_easy_setopt(m_handle, CURLOPT_IPRESOLVE,      CURL_IPRESOLVE_V4);
   curl_easy_setopt(m_handle, CURLOPT_ENCODING,       "");
 
@@ -89,6 +96,7 @@ CurlGet::start() {
 
 void
 CurlGet::close() {
+  priority_queue_erase(&taskScheduler, &m_taskTimeout);
   if (!is_busy())
     return;
 
@@ -99,6 +107,10 @@ CurlGet::close() {
   m_handle = NULL;
 }
 
+void
+CurlGet::receive_timeout() {
+  return m_stack->transfer_done(m_handle, "Timed out");
+}
 
 double
 CurlGet::size_done() {

src/core/curl_get.h

@@ -43,6 +43,8 @@
 #include <sigc++/signal.h>
 #include <torrent/http.h>
 
+#include "rak/priority_queue_default.h"
+
 namespace core {
 
 class CurlStack;
@@ -69,8 +71,12 @@ private:
   CurlGet(const CurlGet&);
   void operator = (const CurlGet&);
 
+  void               receive_timeout();
+
   bool               m_active;
 
+  rak::priority_item m_taskTimeout;
+  
   CURL*              m_handle;
   CurlStack*         m_stack;
 };

src/core/curl_stack.cc

@@ -111,15 +111,7 @@ CurlStack::receive_action(CurlSocket* socket, int events) {
         if (msg->msg != CURLMSG_DONE)
           throw torrent::internal_error("CurlStack::receive_action() msg->msg != CURLMSG_DONE.");
 
-        iterator itr = std::find_if(begin(), end(), rak::equal(msg->easy_handle, std::mem_fun(&CurlGet::handle)));
-
-        if (itr == end())
-          throw torrent::internal_error("Could not find CurlGet with the right easy_handle.");
-
-        if (msg->data.result == CURLE_OK)
-          (*itr)->signal_done().emit();
-        else
-          (*itr)->signal_failed().emit(curl_easy_strerror(msg->data.result));
+	transfer_done(msg->easy_handle, msg->data.result == CURLE_OK ? NULL : curl_easy_strerror(msg->data.result));
       }
 
       if (empty())
@@ -129,6 +121,19 @@ CurlStack::receive_action(CurlSocket* socket, int events) {
   } while (code == CURLM_CALL_MULTI_PERFORM);
 }
 
+void
+CurlStack::transfer_done(void* handle, const char* msg) {
+  iterator itr = std::find_if(begin(), end(), rak::equal(handle, std::mem_fun(&CurlGet::handle)));
+
+  if (itr == end())
+    throw torrent::internal_error("Could not find CurlGet with the right easy_handle.");
+
+  if (msg == NULL)
+    (*itr)->signal_done().emit();
+  else
+    (*itr)->signal_failed().emit(msg);
+}
+
 void
 CurlStack::receive_timeout() {
   receive_action(NULL, 0);

src/core/curl_stack.h

@@ -111,6 +111,8 @@ class CurlStack : std::deque<CurlGet*> {
 
   static int          set_timeout(void* handle, long timeout_ms, void* userp);
 
+  void                transfer_done(void* handle, const char* msg);
+
  protected:
   void                add_get(CurlGet* get);
   void                remove_get(CurlGet* get);

src/core/http_queue.cc

@@ -54,6 +54,7 @@ HttpQueue::insert(const std::string& url, std::iostream* s) {
   
   h->set_url(url);
   h->set_stream(s);
+  h->set_timeout(5 * 60);
 
   iterator itr = Base::insert(end(), h.get());
 

src/main.cc

@@ -233,7 +233,6 @@ main(int argc, char** argv) {
        "view_sort_new     = seeding,less=d.get_state_changed=\n"
        "view_sort_current = seeding,less=d.get_state_changed=\n"
 
-       // Changing these will bork the (non-existant) scheduler.
        "schedule = view_main,10,10,\"view_sort=main,20\"\n"
        "schedule = view_name,10,10,\"view_sort=name,20\"\n"
 

src/rpc/command_function.cc

@@ -77,7 +77,7 @@ CommandFunctionList::find(const char* key) {
 
 void
 CommandFunctionList::insert(const std::string& key, const std::string& cmd) {
-  base_type::iterator itr = std::find_if(begin(), end(), rak::greater_equal(key, rak::mem_ref(&base_type::value_type::first)));
+  base_type::iterator itr = std::find_if(begin(), end(), rak::less_equal(key, rak::mem_ref(&base_type::value_type::first)));
 
   if (itr != end() && itr->first == key)
     itr->second = cmd;

src/rpc/command_map.h

@@ -102,7 +102,8 @@ public:
   static const int flag_dont_delete   = 0x1;
   static const int flag_delete_key    = 0x2;
   static const int flag_public_xmlrpc = 0x4;
-  static const int flag_modifiable    = 0x8;
+  static const int flag_no_target     = 0x8;
+  static const int flag_modifiable    = 0x10;
 
   CommandMap() {}
   ~CommandMap();

src/rpc/xmlrpc.cc

@@ -417,7 +417,10 @@ xmlrpc_call_command(xmlrpc_env* env, xmlrpc_value* args, void* voidServerInfo) {
     torrent::Object object;
     rpc::target_type target = rpc::make_target();
 
-    xmlrpc_to_object(env, args, itr->second.target(), &target).swap(object);
+    if (itr->second.m_flags & CommandMap::flag_no_target)
+      xmlrpc_to_object(env, args, XmlRpc::call_generic, &target).swap(object);
+    else
+      xmlrpc_to_object(env, args, itr->second.target(), &target).swap(object);
 
     if (env->fault_occurred)
       return NULL;

src/ui/element_peer_list.cc

@@ -75,6 +75,7 @@ ElementPeerList::ElementPeerList(core::Download* d) :
 
   m_bindings['k']       = sigc::mem_fun(this, &ElementPeerList::receive_disconnect_peer);
   m_bindings['*']       = sigc::mem_fun(this, &ElementPeerList::receive_snub_peer);
+  m_bindings['B']       = sigc::mem_fun(this, &ElementPeerList::receive_ban_peer);
   m_bindings[KEY_LEFT] = m_bindings['B' - '@']  = sigc::mem_fun(&m_slotExit, &slot_type::operator());  
   m_bindings[KEY_RIGHT] = m_bindings['F' - '@'] = sigc::bind(sigc::mem_fun(this, &ElementPeerList::activate_display), DISPLAY_INFO);
 
@@ -244,6 +245,17 @@ ElementPeerList::receive_snub_peer() {
   update_itr();
 }
 
+void
+ElementPeerList::receive_ban_peer() {
+  if (m_listItr == m_list.end())
+    return;
+
+  (*m_listItr)->set_banned();
+  m_download->download()->connection_list()->erase(*m_listItr, torrent::ConnectionList::disconnect_quick);
+
+  update_itr();
+}
+
 void
 ElementPeerList::update_itr() {
   m_windowList->mark_dirty();

src/ui/element_peer_list.h

@@ -75,6 +75,7 @@ private:
   void                receive_peer_disconnected(torrent::Peer* p);
 
   void                receive_snub_peer();
+  void                receive_ban_peer();
 
   void                update_itr();