From 8f7dfa139e58830caae6e2b66d831378b8a8c0f4 Mon Sep 17 00:00:00 2001
From: Petr Chalupa <git@pitr.ch>
Date: Tue, 29 Sep 2015 18:02:14 +0200
Subject: [PATCH 1/2] Fix Future#flat when failures happen

---
 lib/concurrent/edge/future.rb       | 17 ++++++++++++++---
 spec/concurrent/edge/future_spec.rb | 10 ++++++++++
 2 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/lib/concurrent/edge/future.rb b/lib/concurrent/edge/future.rb
index b9b265532..86e6eab58 100644
--- a/lib/concurrent/edge/future.rb
+++ b/lib/concurrent/edge/future.rb
@@ -1141,17 +1141,24 @@ def blocked_by
 
       def process_on_done(future)
         countdown = super(future)
-        value     = future.value!
         if countdown.nonzero?
+          internal_state = future.internal_state
+
+          unless internal_state.success?
+            complete_with internal_state
+            return countdown
+          end
+
+          value = internal_state.value
           case value
           when Future
             @BlockedBy.push value
             value.add_callback :pr_callback_notify_blocked, self
             @Countdown.value
           when Event
-            raise TypeError, 'cannot flatten to Event'
+            evaluate_to(lambda { raise TypeError, 'cannot flatten to Event' })
           else
-            raise TypeError, "returned value #{value.inspect} is not a Future"
+            evaluate_to(lambda { raise TypeError, "returned value #{value.inspect} is not a Future" })
           end
         end
         countdown
@@ -1174,6 +1181,10 @@ def clear_blocked_by!
         @BlockedBy.clear
         nil
       end
+
+      def completable?(countdown)
+        !@Future.internal_state.completed? && super(countdown)
+      end
     end
 
     # @!visibility private
diff --git a/spec/concurrent/edge/future_spec.rb b/spec/concurrent/edge/future_spec.rb
index 4476ab350..1c2a12556 100644
--- a/spec/concurrent/edge/future_spec.rb
+++ b/spec/concurrent/edge/future_spec.rb
@@ -312,6 +312,16 @@
     it 'has flat map' do
       f = Concurrent.future { Concurrent.future { 1 } }.flat.then(&:succ)
       expect(f.value!).to eq 2
+
+      err = StandardError.new('boo')
+      f   = Concurrent.future { Concurrent.failed_future(err) }.flat
+      expect(f.reason).to eq err
+
+      f   = Concurrent.future { raise 'boo' }.flat
+      expect(f.reason.message).to eq 'boo'
+
+      f   = Concurrent.future { 'boo' }.flat
+      expect(f.reason).to be_an_instance_of TypeError
     end
   end
 

From 4f08f90cbc57ba5a8bcb92f2d1b95143dfaa9353 Mon Sep 17 00:00:00 2001
From: Petr Chalupa <git@pitr.ch>
Date: Tue, 29 Sep 2015 18:34:51 +0200
Subject: [PATCH 2/2] split examples

---
 spec/concurrent/edge/future_spec.rb | 33 +++++++++++++++++++----------
 1 file changed, 22 insertions(+), 11 deletions(-)

diff --git a/spec/concurrent/edge/future_spec.rb b/spec/concurrent/edge/future_spec.rb
index 1c2a12556..27cc5698f 100644
--- a/spec/concurrent/edge/future_spec.rb
+++ b/spec/concurrent/edge/future_spec.rb
@@ -194,7 +194,7 @@
   describe 'Future' do
     it 'has sync and async callbacks' do
       callbacks_tester = ->(future) do
-        queue  = Queue.new
+        queue = Queue.new
         future.on_completion(:io) { |result| queue.push("async on_completion #{ result.inspect }") }
         future.on_completion! { |result| queue.push("sync on_completion #{ result.inspect }") }
         future.on_success(:io) { |value| queue.push("async on_success #{ value.inspect }") }
@@ -309,19 +309,30 @@
       expect(Concurrent.zip(branch1, branch2).value!).to eq [2, 3]
     end
 
-    it 'has flat map' do
-      f = Concurrent.future { Concurrent.future { 1 } }.flat.then(&:succ)
-      expect(f.value!).to eq 2
+    describe '#flat' do
+      it 'returns value of inner future' do
+        f = Concurrent.future { Concurrent.future { 1 } }.flat.then(&:succ)
+        expect(f.value!).to eq 2
+      end
 
-      err = StandardError.new('boo')
-      f   = Concurrent.future { Concurrent.failed_future(err) }.flat
-      expect(f.reason).to eq err
+      it 'propagates failure of inner future' do
+        err = StandardError.new('boo')
+        f   = Concurrent.future { Concurrent.failed_future(err) }.flat
+        expect(f.reason).to eq err
+      end
 
-      f   = Concurrent.future { raise 'boo' }.flat
-      expect(f.reason.message).to eq 'boo'
+      it 'it propagates failure of the future which was suppose to provide inner future' do
+        f = Concurrent.future { raise 'boo' }.flat
+        expect(f.reason.message).to eq 'boo'
+      end
 
-      f   = Concurrent.future { 'boo' }.flat
-      expect(f.reason).to be_an_instance_of TypeError
+      it 'fails if inner value is not a future' do
+        f = Concurrent.future { 'boo' }.flat
+        expect(f.reason).to be_an_instance_of TypeError
+
+        f = Concurrent.future { Concurrent.completed_event }.flat
+        expect(f.reason).to be_an_instance_of TypeError
+      end
     end
   end