rescandirectmsgs: rescan all streams of users we follow for new and old directmessages

libtorrent/include/libtorrent/torrent.hpp

@@ -523,6 +523,8 @@ namespace libtorrent
 			return m_picker->have_piece(index);
 		}
 
+		void recheck_pieces(uint32_t piece_flags);
+
 		// called when we learn that we have a piece
 		// only once per piece
 		void we_have(int index, boost::uint32_t post_flags);

libtorrent/include/libtorrent/torrent_handle.hpp

@@ -170,6 +170,7 @@ namespace libtorrent
 		void read_piece(int piece) const;
 		void get_pieces(std::vector<std::string> &pieces, int count, int max_id, int since_id, uint32_t filter_flags) const;
 		bool have_piece(int piece) const;
+		void recheck_pieces(uint32_t piece_flags) const;
 
 		void get_full_peer_list(std::vector<peer_list_entry>& v) const;
 		void get_peer_info(std::vector<peer_info>& v) const;

libtorrent/src/torrent.cpp

@@ -3073,6 +3073,30 @@ namespace libtorrent
 			m_picker->set_piece_priority(i, 6);
 	}
 
+	void on_disk_read_recheck_piece_complete(int ret, disk_io_job const& j, peer_request r)
+	{
+		// [MF] FIXME: implement cond_wakeup here so that recheck_pieces would wait
+	}
+
+	void torrent::recheck_pieces(uint32_t piece_flags)
+	{
+		TORRENT_ASSERT(m_ses.is_network_thread());
+		TORRENT_ASSERT(m_picker);
+
+		for( int i = 0; i <= last_have(); i++) {
+			if( m_picker->have_piece(i) && m_picker->post_flags(i) == piece_flags ) {
+				peer_request r;
+				r.piece = i;
+				r.start = 0;
+				r.length = torrent_file().piece_size(i);
+
+				filesystem().async_read_and_hash(r,
+				    boost::bind(&on_disk_read_recheck_piece_complete
+						, _1, _2, r), 1);
+			}
+		}
+	}
+
 	void torrent::we_have(int index, boost::uint32_t post_flags)
 	{
 		TORRENT_ASSERT(m_ses.is_network_thread());

libtorrent/src/torrent_handle.cpp

@@ -846,6 +846,12 @@ namespace libtorrent
 		return r;
 	}
 
+	void torrent_handle::recheck_pieces(uint32_t piece_flags) const
+	{
+		INVARIANT_CHECK;
+		TORRENT_SYNC_CALL1(recheck_pieces, piece_flags);
+	}
+
 	storage_interface* torrent_handle::get_storage_impl() const
 	{
 		INVARIANT_CHECK;

src/bitcoinrpc.cpp

@@ -256,6 +256,7 @@ static const CRPCCommand vRPCCommands[] =
     { "listusernamespartial",   &listusernamespartial,   false,     true },
     { "getdefaultuser",         &getdefaultuser,         false,     true },
     { "setdefaultuser",         &setdefaultuser,         false,     true },
+    { "rescandirectmsgs",       &rescandirectmsgs,       false,     true },
 };
 
 CRPCTable::CRPCTable()

src/bitcoinrpc.h

@@ -208,5 +208,6 @@ extern json_spirit::Value getlasthave(const json_spirit::Array& params, bool fHe
 extern json_spirit::Value listusernamespartial(const json_spirit::Array& params, bool fHelp);
 extern json_spirit::Value getdefaultuser(const json_spirit::Array& params, bool fHelp);
 extern json_spirit::Value setdefaultuser(const json_spirit::Array& params, bool fHelp);
+extern json_spirit::Value rescandirectmsgs(const json_spirit::Array& params, bool fHelp);
 
 #endif

src/twister.cpp

@@ -653,11 +653,10 @@ bool processReceivedDM(lazy_entry const* post)
             } else {
                 std::string textOut;
                 if( key.Decrypt(sec, textOut) ) {
-                    /* this printf is good for debug, but bad for security.
+                    // this printf is good for debug, but bad for security.
                     printf("Received DM for user '%s' text = '%s'\n",
                            item.second.username.c_str(),
                            textOut.c_str());
-                    */
 
                     std::string n = post->dict_find_string_value("n");
 
@@ -958,7 +957,8 @@ bool shouldDhtResourceExpire(std::string resource, bool multi, int height)
             if( m_noExpireResources[resourceBasic] == PostNoExpireRecent &&
                 (height + BLOCK_AGE_TO_EXPIRE_DHT_POSTS) < getBestHeight() ) {
 #ifdef DEBUG_EXPIRE_DHT_ITEM
-                printf("shouldDhtResourceExpire: expiring old post resource '%s'\n", resource.c_str());
+                printf("shouldDhtResourceExpire: expiring old post resource '%s' (height %d cur %d)\n",
+                       resource.c_str(), height, getBestHeight());
 #endif
                 return true;
             }
@@ -1594,3 +1594,33 @@ Value listusernamespartial(const Array& params, bool fHelp)
     return ret;
 }
 
+Value rescandirectmsgs(const Array& params, bool fHelp)
+{
+    if (fHelp || (params.size() != 1))
+        throw runtime_error(
+            "rescandirectmsgs <username>\n"
+            "rescan all streams of users we follow for new and old directmessages");
+
+    string localUser = params[0].get_str();
+
+    std::set<std::string> following;
+    {
+        LOCK(cs_twister);
+        following = m_users[localUser].m_following;
+    }
+
+    BOOST_FOREACH(string username, following) {
+        torrent_handle torrent;
+
+        {
+            LOCK(cs_twister);
+            if( username.size() && m_userTorrent.count(username) )
+                torrent = m_userTorrent[username];
+        }
+        if( torrent.is_valid() ){
+            torrent.recheck_pieces(USERPOST_FLAG_DM);
+        }
+    }
+
+    return Value();
+}