Implement parallel map using Celluloid::Futures

README.md

@@ -488,7 +488,8 @@ Here are some fun facts about this library:
 * This gem works on every major Ruby implementation, including JRuby and
   Rubinius
 * The first version was released on November 26, 2006
-* This gem has just two runtime dependencies: `faraday`, and `simple_oauth`
+* This gem has just three runtime dependencies: `celluloid`, `faraday`, and
+  `simple_oauth`
 * Previous versions of this gem have been [downloaded over a million
   times][stats]
 

lib/twitter/core_ext/enumerable.rb

@@ -1,10 +1,12 @@
+require 'celluloid'
+
 module Enumerable
 
-  def threaded_map
-    threads = map do |object|
-      Thread.new { yield object }
+  def pmap(&block)
+    futures = map do |elem|
+      Celluloid::Future.new(elem, &block)
     end
-    threads.map(&:value)
+    futures.map(&:value)
   end
 
 end

lib/twitter/rest/api/direct_messages.rb

@@ -88,7 +88,7 @@ def direct_messages(*args)
           if arguments.empty?
             direct_messages_received(arguments.options)
           else
-            arguments.flatten.threaded_map do |id|
+            arguments.flatten.pmap do |id|
               direct_message(id, arguments.options)
             end
           end
@@ -108,7 +108,7 @@ def direct_messages(*args)
         #   @param ids [Enumerable<Integer>] A collection of direct message IDs.
         #   @param options [Hash] A customizable set of options.
         def direct_message_destroy(*args)
-          threaded_object_from_response(Twitter::DirectMessage, :post, "/1.1/direct_messages/destroy.json", args)
+          parallel_objects_from_response(Twitter::DirectMessage, :post, "/1.1/direct_messages/destroy.json", args)
         end
 
         # Sends a new direct message to the specified user from the authenticating user

lib/twitter/rest/api/favorites.rb

@@ -51,7 +51,7 @@ def favorites(*args)
         #   @param tweets [Enumerable<Integer, String, URI, Twitter::Tweet>] A collection of Tweet IDs, URIs, or objects.
         #   @param options [Hash] A customizable set of options.
         def unfavorite(*args)
-          threaded_object_from_response(Twitter::Tweet, :post, "/1.1/favorites/destroy.json", args)
+          parallel_objects_from_response(Twitter::Tweet, :post, "/1.1/favorites/destroy.json", args)
         end
         alias favorite_destroy unfavorite
         alias favourite_destroy unfavorite
@@ -71,7 +71,7 @@ def unfavorite(*args)
         #   @param options [Hash] A customizable set of options.
         def favorite(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             begin
               object_from_response(Twitter::Tweet, :post, "/1.1/favorites/create.json", arguments.options.merge(:id => id))
@@ -100,7 +100,7 @@ def favorite(*args)
         #   @param options [Hash] A customizable set of options.
         def favorite!(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             begin
               object_from_response(Twitter::Tweet, :post, "/1.1/favorites/create.json", arguments.options.merge(:id => id))

lib/twitter/rest/api/friends_and_followers.rb

@@ -141,7 +141,7 @@ def follow!(*args)
           # Twitter always turns on notifications if the "follow" option is present, even if it's set to false
           # so only send follow if it's true
           arguments.options[:follow] = true if !!arguments.options.delete(:follow)
-          arguments.flatten.threaded_map do |user|
+          arguments.flatten.pmap do |user|
             begin
               object_from_response(Twitter::User, :post, "/1.1/friendships/create.json", merge_user(arguments.options, user))
             rescue Twitter::Error::Forbidden
@@ -165,7 +165,7 @@ def follow!(*args)
         #   @param users [Enumerable<Integer, String, Twitter::User>] A collection of Twitter user IDs, screen names, or objects.
         #   @param options [Hash] A customizable set of options.
         def unfollow(*args)
-          threaded_user_objects_from_response(:post, "/1.1/friendships/destroy.json", args)
+          parallel_user_objects_from_response(:post, "/1.1/friendships/destroy.json", args)
         end
         alias friendship_destroy unfollow
 

lib/twitter/rest/api/lists.rb

@@ -438,7 +438,7 @@ def list_from_response_with_users(request_method, path, args)
           members = arguments.pop
           merge_list!(arguments.options, arguments.pop)
           merge_owner!(arguments.options, arguments.pop)
-          members.flatten.each_slice(MAX_USERS_PER_REQUEST).threaded_map do |users|
+          members.flatten.each_slice(MAX_USERS_PER_REQUEST).pmap do |users|
             object_from_response(Twitter::List, request_method, path, merge_users(arguments.options, users))
           end.last
         end

lib/twitter/rest/api/saved_searches.rb

@@ -33,7 +33,7 @@ def saved_searches(*args)
           if arguments.empty?
             objects_from_response(Twitter::SavedSearch, :get, "/1.1/saved_searches/list.json", arguments.options)
           else
-            arguments.flatten.threaded_map do |id|
+            arguments.flatten.pmap do |id|
               saved_search(id, arguments.options)
             end
           end
@@ -80,7 +80,7 @@ def saved_search_create(query, options={})
         #   @param options [Hash] A customizable set of options.
         def saved_search_destroy(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |id|
+          arguments.flatten.pmap do |id|
             object_from_response(Twitter::SavedSearch, :post, "/1.1/saved_searches/destroy/#{id}.json", arguments.options)
           end
         end

lib/twitter/rest/api/spam_reporting.rb

@@ -20,7 +20,7 @@ module SpamReporting
         #   @param users [Enumerable<Integer, String, Twitter::User>] A collection of Twitter user IDs, screen names, or objects.
         #   @param options [Hash] A customizable set of options.
         def report_spam(*args)
-          threaded_user_objects_from_response(:post, "/1.1/users/report_spam.json", args)
+          parallel_user_objects_from_response(:post, "/1.1/users/report_spam.json", args)
         end
 
       end

lib/twitter/rest/api/tweets.rb

@@ -80,7 +80,7 @@ def status(tweet, options={})
         #   @param options [Hash] A customizable set of options.
         #   @option options [Boolean, String, Integer] :trim_user Each tweet returned in a timeline will include a user object with only the author's numerical ID when set to true, 't' or 1.
         def statuses(*args)
-          threaded_tweets_from_response(:get, "/1.1/statuses/show", args)
+          parallel_tweets_from_response(:get, "/1.1/statuses/show", args)
         end
 
         # Destroys the specified Tweets
@@ -98,7 +98,7 @@ def statuses(*args)
         #   @param options [Hash] A customizable set of options.
         #   @option options [Boolean, String, Integer] :trim_user Each tweet returned in a timeline will include a user object with only the author's numerical ID when set to true, 't' or 1.
         def status_destroy(*args)
-          threaded_tweets_from_response(:post, "/1.1/statuses/destroy", args)
+          parallel_tweets_from_response(:post, "/1.1/statuses/destroy", args)
         end
         alias tweet_destroy status_destroy
 
@@ -139,7 +139,7 @@ def update(status, options={})
         #   @option options [Boolean, String, Integer] :trim_user Each tweet returned in a timeline will include a user object with only the author's numerical ID when set to true, 't' or 1.
         def retweet(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             begin
               post_retweet(id, arguments.options)
@@ -165,7 +165,7 @@ def retweet(*args)
         #   @option options [Boolean, String, Integer] :trim_user Each tweet returned in a timeline will include a user object with only the author's numerical ID when set to true, 't' or 1.
         def retweet!(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             begin
               post_retweet(id, arguments.options)
@@ -240,7 +240,7 @@ def oembed(tweet, options={})
         #   @option options [String] :lang Language code for the rendered embed. This will affect the text and localization of the rendered HTML.
         def oembeds(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             oembed(id, arguments.options)
           end
@@ -270,9 +270,9 @@ def retweeters_ids(*args)
         # @param path [String]
         # @param args [Array]
         # @return [Array<Twitter::Tweet>]
-        def threaded_tweets_from_response(request_method, path, args)
+        def parallel_tweets_from_response(request_method, path, args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |tweet|
+          arguments.flatten.pmap do |tweet|
             id = extract_id(tweet)
             object_from_response(Twitter::Tweet, request_method, path + "/#{id}.json", arguments.options)
           end

lib/twitter/rest/api/users.rb

@@ -192,7 +192,7 @@ def block?(user, options={})
         #   @param users [Enumerable<Integer, String, Twitter::User>] A collection of Twitter user IDs, screen names, or objects.
         #   @param options [Hash] A customizable set of options.
         def block(*args)
-          threaded_user_objects_from_response(:post, "/1.1/blocks/create.json", args)
+          parallel_user_objects_from_response(:post, "/1.1/blocks/create.json", args)
         end
 
         # Un-blocks the users specified by the authenticating user
@@ -208,7 +208,7 @@ def block(*args)
         #   @param users [Enumerable<Integer, String, Twitter::User>] A collection of Twitter user IDs, screen names, or objects.
         #   @param options [Hash] A customizable set of options.
         def unblock(*args)
-          threaded_user_objects_from_response(:post, "/1.1/blocks/destroy.json", args)
+          parallel_user_objects_from_response(:post, "/1.1/blocks/destroy.json", args)
         end
 
         # Returns extended information for up to 100 users
@@ -228,7 +228,7 @@ def unblock(*args)
         def users(*args)
           arguments = Twitter::REST::API::Arguments.new(args)
           method = arguments.options.delete(:method) || :post
-          arguments.flatten.each_slice(MAX_USERS_PER_REQUEST).threaded_map do |users|
+          arguments.flatten.each_slice(MAX_USERS_PER_REQUEST).pmap do |users|
             objects_from_response(Twitter::User, method, "/1.1/users/lookup.json", merge_users(arguments.options, users))
           end.flatten
         end

lib/twitter/rest/api/utils.rb

@@ -34,9 +34,9 @@ def extract_id(object)
         # @param path [String]
         # @param args [Array]
         # @return [Array<Twitter::User>]
-        def threaded_user_objects_from_response(request_method, path, args)
+        def parallel_user_objects_from_response(request_method, path, args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |user|
+          arguments.flatten.pmap do |user|
             object_from_response(Twitter::User, request_method, path, merge_user(arguments.options, user))
           end
         end
@@ -86,9 +86,9 @@ def objects_from_array(klass, array)
         # @param path [String]
         # @param args [Array]
         # @return [Array]
-        def threaded_object_from_response(klass, request_method, path, args)
+        def parallel_objects_from_response(klass, request_method, path, args)
           arguments = Twitter::REST::API::Arguments.new(args)
-          arguments.flatten.threaded_map do |object|
+          arguments.flatten.pmap do |object|
             id = extract_id(object)
             object_from_response(klass, request_method, path, arguments.options.merge(:id => id))
           end

twitter.gemspec

@@ -4,6 +4,7 @@ $LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
 require 'twitter/version'
 
 Gem::Specification.new do |spec|
+  spec.add_dependency 'celluloid', '~> 0.14.0'
   spec.add_dependency 'faraday', ['~> 0.8', '< 0.10']
   spec.add_dependency 'simple_oauth', '~> 0.2'
   spec.add_development_dependency 'bundler', '~> 1.0'