Implement start method on supplies.

Takes a closure and, for each supplied value, schedules the closure to
run on another thread. It then more's a Supply (resulting in us having
a supply of supplies) that will either have a single value more'd and
then be done if the async work completes successfully, or quit if the
work fails. Useful for kicking off work on the thread pool if you do
not want to block up the thread pushing values at you (maybe 'cus you
are reacting to UI events, but have some long-running work to kick
off).

src/core/Supply.pm

@@ -102,6 +102,7 @@ my role Supply {
     method schedule_on(Scheduler $scheduler) {
         SupplyOperations.schedule_on(self, $scheduler);
     }
+    method start(&startee)     { SupplyOperations.start(self, &startee) }
     method merge(*@s)          { SupplyOperations.merge(self, @s) }
     method zip(*@s,:&with)     { SupplyOperations.zip(self, @s, :&with) }
 

src/core/SupplyOperations.pm

@@ -377,6 +377,32 @@ my class SupplyOperations is repr('Uninstantiable') {
         ScheduleSupply.new(:source($s), :$scheduler)
     }
 
+    method start(Supply $s, &startee) {
+        my class StartSupply does Supply does PrivatePublishing {
+            has $!value;
+            has &!startee;
+
+            submethod BUILD(:$!value, :&!startee) { }
+
+            method tap(|c) {
+                my $sub = self.Supply::tap(|c);
+                Promise.start({ &!startee($!value) }).then({
+                    if .status == Kept {
+                        self!more(.result);
+                        self!done();
+                    }
+                    else {
+                        self!quit(.cause);
+                    }
+                });
+                $sub
+            }
+        }
+        self.map($s, -> \value {
+            StartSupply.new(:value(value), :&startee)
+        })
+    }
+
     method merge(*@s) {
 
         @s.shift unless @s[0].DEFINITE;  # lose if used as class method