Allow to insert throttle into chain

Author: pitr-ch

lib/concurrent/edge/promises.rb

@@ -2031,10 +2031,12 @@ def initialize(max)
       @Queue       = Queue.new
     end
 
-    def limit(ready = nil, &block)
+    def limit(future = nil, &block)
       # TODO (pitr-ch 11-Jun-2016): triggers should allocate resources when they are to be required
+      trigger = future ? future & get_event : get_event
+
       if block_given?
-        block.call(get_event).on_resolution! { done }
+        block.call(trigger).on_resolution! { done }
       else
         get_event
       end
@@ -2067,4 +2069,12 @@ def get_event
       end
     end
   end
+
+  class Promises::AbstractEventFuture < Synchronization::Object
+
+    def throttle(throttle, &throttled_future)
+      throttle.limit(self, &throttled_future)
+    end
+
+  end
 end

spec/concurrent/promises_spec.rb

@@ -496,19 +496,26 @@ def behaves_as_delay(delay, value)
 
   describe 'Throttling' do
     specify do
-      throttle = Concurrent::Throttle.new 3
+      max_tree = Concurrent::Throttle.new 3
       counter  = Concurrent::AtomicFixnum.new
+      testing  = -> do
+        counter.increment
+        sleep 0.01
+        # returns less then 3 since it's throttled
+        counter.decrement
+      end
+
       expect(Concurrent::Promises.zip(
           *12.times.map do |i|
-            throttle.limit do |trigger|
-              trigger.then do
-                counter.increment
-                sleep 0.01
-                counter.decrement
-              end
-            end
+            max_tree.limit { |trigger| trigger.then &testing }
+          end).value.all? { |v| v < 3 }).to be_truthy
+
+      expect(Concurrent::Promises.zip(
+          *12.times.map do |i|
+            Concurrent::Promises.
+                fulfilled_future(i).
+                throttle(max_tree) { |trigger| trigger.then &testing }
           end).value.all? { |v| v < 3 }).to be_truthy
     end
   end
-
 end