Implement schedule_on.

This allows a supply's more/done/quit to be scheduled on another
scheduler. Useful in GUI situations, for example, where the final
stage of some work needs to be done on some UI scheduler in order to
have UI updates run on the UI thread.

src/core/Supply.pm

@@ -99,6 +99,9 @@ my role Supply {
     method batch( :$elems, :$seconds ) {
         SupplyOperations.batch( self, :$elems, :$seconds)
     }
+    method schedule_on(Scheduler $scheduler) {
+        SupplyOperations.schedule_on(self, $scheduler);
+    }
     method merge(*@s)          { SupplyOperations.merge(self, @s) }
     method zip(*@s,:&with)     { SupplyOperations.zip(self, @s, :&with) }
 

src/core/SupplyOperations.pm

@@ -355,6 +355,27 @@ my class SupplyOperations is repr('Uninstantiable') {
         }
         BatchSupply.new(:source($s), :$elems, :$seconds)
     }
+
+    method schedule_on(Supply $s, Scheduler $scheduler) {
+        my class ScheduleSupply does Supply does PrivatePublishing {
+            has $!source;
+            has $!scheduler;
+
+            submethod BUILD(:$!source, :$!scheduler) { }
+
+            method tap(|c) {
+                my $source_tap;
+                my $sub = self.Supply::tap(|c, closing => { $source_tap.close() });
+                $source_tap = $!source.tap( -> \val {
+                      $!scheduler.cue: { self!more(val) }
+                  },
+                  done => { $!scheduler.cue: { self!done(); } },
+                  quit => -> $ex { $!scheduler.cue: { self!quit($ex) } });
+                $sub
+            }
+        }
+        ScheduleSupply.new(:source($s), :$scheduler)
+    }
 
     method merge(*@s) {