added timeout to fetchers

concurrency/subtasks/divide_and_conquer/divide_and_conquer.go

@@ -30,6 +30,7 @@ func RunDivideAndConquer() {
 
 	evaluators := []Evaluator{
 		EvaluatorFunc(func(inV interface{}) (interface{}, error) {
+			time.Sleep(time.Second * 1)
 			i := inV.(in)
 			r := i.a + i.b
 			return out{"Plus", r}, nil
@@ -83,11 +84,18 @@ func DivideAndConquer(data interface{}, evaluators []Evaluator, timeout time.Dur
 			go func() {
 				result, err := e.Evaluate(data)
 				if err != nil {
-					errors <- err
+					ech <- err
 				} else {
 					ch <- result
 				}
 			}()
+			// Remember that an unbuffered channel pauses the writing goroutine until there’s a read by another
+			// goroutine. If the timeout triggers before the Evaluator finishes executing,
+			// the read will never happen because the only place those channels are read is in the outer
+			// goroutine’s select statement, and the outer goroutine exited after the timeout triggered.
+			// This means that using an unbuffered channel will cause the inner goroutine to wait forever
+			// whenever there is a timeout, leaking the goroutine. Again, the buffered channel proves
+			// useful because we know exactly how many writes we can expect.
 			select {
 			case r := <-ch:
 				gather <- r

concurrency/subtasks/fetchers/fetchers.go

@@ -34,7 +34,7 @@ type BingFetcher struct {
 }
 
 func (b *BingFetcher) Fetch(url string) (string, error) {
-	time.Sleep(time.Second * 3)
+	time.Sleep(time.Second * 10)
 	return fmt.Sprintf("%s is fetching %s", b.Name, url), nil
 }
 
@@ -51,7 +51,7 @@ type DuckDuckGoFetcher struct {
 }
 
 func (d *DuckDuckGoFetcher) Fetch(url string) (string, error) {
-	time.Sleep(time.Second * 2)
+	time.Sleep(time.Second * 10)
 	return fmt.Sprintf("%s is fetching %s", d.Name, url), nil
 }
 
@@ -67,11 +67,23 @@ func FetchResults(url string, fetchers []Fetcher, timeout time.Duration) ([]stri
 	chErr := make(chan error, len(fetchers))
 	for _, f := range fetchers {
 		go func(f Fetcher) {
+			ch := make(chan string, 1)
+			eCh := make(chan error, 1)
+			go func() {
 				s, err := f.Fetch(url)
 				if err != nil {
-				chErr <- err
+					eCh <- err
 				} else {
-				chStr <- s
+					ch <- s
+				}
+			}()
+			select {
+			case r := <-ch:
+				chStr <- r
+			case err := <-eCh:
+				chErr <- err
+			case <-time.After(timeout):
+				chErr <- fmt.Errorf("%s timeout after %v on %v", f.GetName(), timeout, url)
 			}
 		}(f)
 	}
@@ -89,7 +101,7 @@ func FetchResults(url string, fetchers []Fetcher, timeout time.Duration) ([]stri
 }
 
 func RunFetchers() {
-	fetchers := []Fetcher{NewGoogleFetcher("Google"), NewGoogleFetcher("Bing"), NewGoogleFetcher("Duck Duck Go")}
-	r, e := FetchResults("http://www.abc.com", fetchers, time.Millisecond*100)
+	fetchers := []Fetcher{NewGoogleFetcher("Google"), NewBingFetcher("Bing"), NewDuckDuckGoFetcherFetcher("Duck Duck Go")}
+	r, e := FetchResults("http://www.abc.com", fetchers, time.Second*2)
 	spew.Dump(r, e)
 }