Merge 701cebe into 17880f4

README.md

@@ -296,7 +296,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 half a million
   times][stats]
 

lib/twitter/api/direct_messages.rb

@@ -97,7 +97,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
@@ -119,7 +119,7 @@ def direct_messages(*args)
       #   @param ids [Array<Integer>, Set<Integer>] An array of Tweet 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_object_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/api/favorites.rb

@@ -56,7 +56,7 @@ def favorites(*args)
       #   @param ids [Array<Integer>, Set<Integer>] An array of Tweet IDs.
       #   @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_object_from_response(Twitter::Tweet, :post, "/1.1/favorites/destroy.json", args)
       end
       alias favorite_destroy unfavorite
       alias favourite_destroy unfavorite
@@ -78,7 +78,7 @@ def unfavorite(*args)
       #   @param options [Hash] A customizable set of options.
       def favorite(*args)
         arguments = Twitter::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           begin
             object_from_response(Twitter::Tweet, :post, "/1.1/favorites/create.json", arguments.options.merge(id: id))
           rescue Twitter::Error::Forbidden => error
@@ -108,7 +108,7 @@ def favorite(*args)
       #   @param options [Hash] A customizable set of options.
       def favorite!(*args)
         arguments = Twitter::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           begin
             object_from_response(Twitter::Tweet, :post, "/1.1/favorites/create.json", arguments.options.merge(id: id))
           rescue Twitter::Error::Forbidden => error

lib/twitter/api/friends_and_followers.rb

@@ -162,7 +162,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
@@ -188,7 +188,7 @@ def follow!(*args)
       #   @param users [Array<Integer, String, Twitter::User>, Set<Integer, String, Twitter::User>] An array 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/api/lists.rb

@@ -576,7 +576,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/api/saved_searches.rb

@@ -36,7 +36,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
@@ -89,7 +89,7 @@ def saved_search_create(query, options={})
       #   @param options [Hash] A customizable set of options.
       def saved_search_destroy(*args)
         arguments = Twitter::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/api/spam_reporting.rb

@@ -22,7 +22,7 @@ module SpamReporting
       #   @param users [Array<Integer, String, Twitter::User>, Set<Integer, String, Twitter::User>] An array 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/api/tweets.rb

@@ -84,7 +84,7 @@ def status(id, 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
@@ -104,7 +104,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
 
@@ -147,7 +147,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::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           begin
             post_retweet(id, arguments.options)
           rescue Twitter::Error::Forbidden => error
@@ -174,7 +174,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::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           begin
             post_retweet(id, arguments.options)
           rescue Twitter::Error::Forbidden => error
@@ -252,7 +252,7 @@ def oembed(id_or_url, 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::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id_or_url|
+        arguments.flatten.pmap do |id_or_url|
           oembed(id_or_url, arguments.options)
         end
       end
@@ -285,9 +285,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::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           object_from_response(Twitter::Tweet, request_method, path + "/#{id}.json", arguments.options)
         end
       end

lib/twitter/api/users.rb

@@ -215,7 +215,7 @@ def block?(user, options={})
       #   @param users [Array<Integer, String, Twitter::User>, Set<Integer, String, Twitter::User>] An array 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
@@ -234,7 +234,7 @@ def block(*args)
       #   @param users [Array<Integer, String, Twitter::User>, Set<Integer, String, Twitter::User>] An array 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
@@ -260,7 +260,7 @@ def unblock(*args)
       def users(*args)
         arguments = Twitter::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/api/utils.rb

@@ -14,9 +14,9 @@ module Utils
       # @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::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
@@ -66,9 +66,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_object_from_response(klass, request_method, path, args)
         arguments = Twitter::API::Arguments.new(args)
-        arguments.flatten.threaded_map do |id|
+        arguments.flatten.pmap do |id|
           object_from_response(klass, request_method, path, arguments.options.merge(id: id))
         end
       end

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/stream.rb

@@ -0,0 +1,3 @@
+require 'twitter/stream/client'
+require 'twitter/stream/user_stream_client'
+require 'twitter/stream/site_stream_client'

lib/twitter/stream/client.rb

@@ -0,0 +1,43 @@
+require 'twitter/stream/core/io'
+require 'twitter/stream/core/common'
+
+module Twitter
+  module Stream
+    class Client
+      include Celluloid
+      include Common
+
+      DEFAULT_OPTIONS = {
+        :host         => 'stream.twitter.com',
+        :port         => 443,
+        :method       => 'POST',
+        :content_type => 'application/x-www-form-urlencoded',
+        :headers      => {},
+        :user_agent   => "Twitter Celluloid",
+        :proxy        => nil,
+        :ssl          => {},
+        :timeout      => 0,
+        :path         => '',
+        :params       => {},
+        :oauth        => {},
+        :encoding     => nil,
+      }
+
+      def filter(params={})
+        @options[:path] ='/1.1/statuses/filter.json'
+        super
+      end
+
+      def sample
+        @options[:path] ='/1.1/statuses/sample.json'
+        super
+      end
+
+      def firehose
+        @options[:path] ='/1.1/statuses/firehose.json'
+        super
+      end
+
+    end
+  end
+end