Support multiple parallel functions to StoppableFunc

Author: matteosuppo

cc.go

@@ -47,6 +47,10 @@ func (p *Pool) Run(fn func() error) {
 	}()
 }
 
+// StoppableFunc is a func that receives a stop channel that when closed broadcasts the
+// need to wrap up and stop the function
+type StoppableFunc func(stop chan struct{})
+
 // Stoppable is a function that can be stopped with the method Stop. You can also listen on the Stopped channel to see when it has been stopped.
 // Stoppable is different from a context cancelation because it waits until the function has cleaned up before broadcasting on the Stopped channel
 type Stoppable struct {
@@ -62,18 +66,34 @@ func (s *Stoppable) Stop() {
 	})
 }
 
-// Run creates a new stoppable function from the provided func. When you call the Stop method on the returned Stoppable the stop channel fed to the provided func is closed,
-// signaling the need to stop. When the provided func returns the Stopped channel on the
+// Run creates a new stoppable function from the provided funcs. When you call the Stop method on the returned Stoppable the stop channel fed to the provided funcs is closed,
+// signaling the need to stop. When all the provided func return the Stopped channel on the
 // returned Stoppable is closed as well, broadcasting the message that it has finished
-func Run(fn func(stop chan struct{})) (s *Stoppable) {
+func Run(fns ...StoppableFunc) (s *Stoppable) {
 	s = &Stoppable{
 		Stopped: make(chan struct{}),
 		stop:    make(chan struct{}),
 	}
 
+	stoppedList := []chan struct{}{}
+
+	for _, fn := range fns {
+		stopped := make(chan struct{})
+		stoppedList = append(stoppedList, stopped)
+
+		go func(fn StoppableFunc, stopped chan struct{}) {
+			fn(s.stop)
+			s.Stop()
+			close(stopped)
+		}(fn, stopped)
+	}
+
 	go func() {
-		fn(s.stop)
+		for _, stopped := range stoppedList {
+			<-stopped
+		}
 		close(s.Stopped)
 	}()
+
 	return s
 }

cc_test.go

@@ -66,6 +66,67 @@ func ExampleStoppable() {
 	// stopped finally
 }
 
+func ExampleStoppable_multiple() {
+	stoppable := cc.Run(func(stop chan struct{}) {
+		i := 0
+	L:
+		for {
+			select {
+			case <-stop:
+				fmt.Println("first received stop signal")
+				break L
+			default:
+				i++
+				time.Sleep(250 * time.Millisecond)
+				fmt.Println("first: ", i)
+			}
+		}
+		fmt.Println("first finished with", i)
+	}, func(stop chan struct{}) {
+		time.Sleep(50 * time.Millisecond)
+		i := 0
+	L:
+
+		for {
+			select {
+			case <-stop:
+				fmt.Println("second received stop signal")
+				break L
+			default:
+				i++
+				time.Sleep(250 * time.Millisecond)
+				fmt.Println("second: ", i)
+			}
+		}
+		fmt.Println("second finished with", i)
+	})
+
+	go func() {
+		time.Sleep(1 * time.Second)
+		fmt.Println("send stop signal")
+		stoppable.Stop()
+		stoppable.Stop() // It shouldn't explode even if you attempt to close it multiple times
+	}()
+
+	<-stoppable.Stopped
+	fmt.Println("stopped finally")
+	// Output: first:  1
+	// second:  1
+	// first:  2
+	// second:  2
+	// first:  3
+	// second:  3
+	// send stop signal
+	// first:  4
+	// first received stop signal
+	// first finished with 4
+	// second:  4
+	// second received stop signal
+	// second finished with 4
+	// stopped finally
+
+}
+
 func TestRace(t *testing.T) {
 	p := cc.New(4)