Merge pull request #127 from lovoo/feature/run-context

Feature/run context

README.md

@@ -65,6 +65,7 @@ To locally start a dockerized Zookeeper and Kafka instances, execute `make start
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"os"
@@ -124,16 +125,20 @@ func runProcessor() {
 	if err != nil {
 		log.Fatalf("error creating processor: %v", err)
 	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan bool)
 	go func() {
-		if err = p.Start(); err != nil {
+		defer close(done)
+		if err = p.Run(ctx); err != nil {
 			log.Fatalf("error running processor: %v", err)
 		}
 	}()
 
 	wait := make(chan os.Signal, 1)
 	signal.Notify(wait, syscall.SIGINT, syscall.SIGTERM)
 	<-wait   // wait for SIGINT/SIGTERM
-	p.Stop() // gracefully stop processor
+	cancel() // gracefully stop processor
+	<-done
 }
 
 func main() {

examples/1-simplest/main.go

@@ -1,11 +1,9 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
-	"os"
-	"os/signal"
-	"syscall"
 
 	"github.com/lovoo/goka"
 	"github.com/lovoo/goka/codec"
@@ -60,16 +58,9 @@ func runProcessor() {
 	if err != nil {
 		log.Fatalf("error creating processor: %v", err)
 	}
-	go func() {
-		if err = p.Start(); err != nil {
-			log.Fatalf("error running processor: %v", err)
-		}
-	}()
-
-	wait := make(chan os.Signal, 1)
-	signal.Notify(wait, syscall.SIGINT, syscall.SIGTERM)
-	<-wait   // wait for SIGINT/SIGTERM
-	p.Stop() // gracefully stop processor
+	if err = p.Run(context.Background()); err != nil {
+		log.Fatalf("error running processor: %v", err)
+	}
 }
 
 func main() {

examples/2-clicks/README.md

@@ -77,7 +77,7 @@ update it in our group table (10) by calling `ctx.SetValue(u)`.
 We conclude process() with a print statement showing message’s content, the
 current count of the user, and the user ID fetched with ctx.Key().
 
-The context interface never returns errors to the callbacks. Instead, if an error is encountered while executing the context functions, the processor instance is stopped and its Start() method returns an error.
+The context interface never returns errors to the callbacks. Instead, if an error is encountered while executing the context functions, the processor instance is stopped and its Run() method returns an error.
 
 
 We configure the processor using `goka.DefineGroup`, which we later
@@ -128,8 +128,7 @@ func runView() {
 	if err != nil {
 		panic(err)
 	}
-	go view.Start()
-	defer view.Stop()
+	go view.Run(context.Background())
 
 	root := mux.NewRouter()
 	root.HandleFunc("/{key}", func(w http.ResponseWriter, r *http.Request) {

examples/2-clicks/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -91,12 +92,7 @@ func runProcessor() {
 		panic(err)
 	}
 
-	err = p.Start()
-	if err != nil {
-		panic(err)
-	} else {
-		fmt.Println("Processor stopped without errors")
-	}
+	p.Run(context.Background())
 }
 
 func runView() {
@@ -107,8 +103,6 @@ func runView() {
 	if err != nil {
 		panic(err)
 	}
-	go view.Start()
-	defer view.Stop()
 
 	root := mux.NewRouter()
 	root.HandleFunc("/{key}", func(w http.ResponseWriter, r *http.Request) {
@@ -117,7 +111,9 @@ func runView() {
 		w.Write(data)
 	})
 	fmt.Println("View opened at http://localhost:9095/")
-	http.ListenAndServe(":9095", root)
+	go http.ListenAndServe(":9095", root)
+
+	view.Run(context.Background())
 }
 
 func main() {

examples/3-messaging/README.md

@@ -119,8 +119,6 @@ g := goka.DefineGroup(goka.Group("collector"),
 	goka.Input(messaging.ReceivedStream, new(messaging.MessageCodec), collect),
 )
 p, _ := goka.NewProcessor(brokers, g)
-go p.Start()
-
 ```
 
 ### Feed endpoint
@@ -165,7 +163,6 @@ view, _ := goka.NewView(
 	collector.Table,
 	new(collector.MessageListCodec),
 )
-go view.Start()
 router.HandleFunc("/{user}/feed", feed(view)).Methods("GET")
 ```
 
@@ -289,8 +286,6 @@ g := goka.DefineGroup(goka.Group("filter"),
 )
 
 p, _ := goka.NewProcessor(brokers, g)
-_ = p.Start()
-
 ```
 
 Nothing has to be changed in the collector processor or in the feed endpoint.
@@ -493,7 +488,6 @@ g := goka.DefineGroup(goka.Group("detector"),
 )
 
 p, _ := goka.NewProcessor(brokers, g)
-_ = p.Start()
 ```
 
 ### Running the example

examples/3-messaging/blocker/blocker.go

@@ -1,6 +1,7 @@
 package blocker
 
 import (
+	"context"
 	"encoding/json"
 
 	"github.com/lovoo/goka"
@@ -57,14 +58,16 @@ func block(ctx goka.Context, msg interface{}) {
 	ctx.SetValue(s)
 }
 
-func Run(brokers []string) {
-	g := goka.DefineGroup(group,
-		goka.Input(Stream, new(BlockEventCodec), block),
-		goka.Persist(new(BlockValueCodec)),
-	)
-	if p, err := goka.NewProcessor(brokers, g); err != nil {
-		panic(err)
-	} else if err = p.Start(); err != nil {
-		panic(err)
+func Run(ctx context.Context, brokers []string) func() error {
+	return func() error {
+		g := goka.DefineGroup(group,
+			goka.Input(Stream, new(BlockEventCodec), block),
+			goka.Persist(new(BlockValueCodec)),
+		)
+		p, err := goka.NewProcessor(brokers, g)
+		if err != nil {
+			return err
+		}
+		return p.Run(ctx)
 	}
 }

examples/3-messaging/cmd/processor/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"log"
 	"os"
@@ -12,6 +13,7 @@ import (
 	"github.com/lovoo/goka/examples/3-messaging/detector"
 	"github.com/lovoo/goka/examples/3-messaging/filter"
 	"github.com/lovoo/goka/examples/3-messaging/translator"
+	"golang.org/x/sync/errgroup"
 )
 
 var (
@@ -26,33 +28,40 @@ var (
 
 func main() {
 	flag.Parse()
+	ctx, cancel := context.WithCancel(context.Background())
+	grp, ctx := errgroup.WithContext(ctx)
+
 	if *runCollector {
 		log.Println("starting collector")
-		go collector.Run(brokers)
+		grp.Go(collector.Run(ctx, brokers))
 	}
 	if *runFilter {
 		log.Println("starting filter")
-		go filter.Run(brokers)
+		grp.Go(filter.Run(ctx, brokers))
 	}
 	if *runBlocker {
 		log.Println("starting blocker")
-		go blocker.Run(brokers)
+		grp.Go(blocker.Run(ctx, brokers))
 	}
 	if *runDetector {
 		log.Println("starting detector")
-		go detector.Run(brokers)
+		grp.Go(detector.Run(ctx, brokers))
 	}
 	if *runTranslator {
 		log.Println("starting translator")
-		go translator.Run(brokers)
+		grp.Go(translator.Run(ctx, brokers))
 	}
 
 	// Wait for SIGINT/SIGTERM
 	waiter := make(chan os.Signal, 1)
 	signal.Notify(waiter, syscall.SIGINT, syscall.SIGTERM)
-
 	select {
-	case signal := <-waiter:
-		log.Printf("Got interrupted by %v", signal)
+	case <-waiter:
+	case <-ctx.Done():
+	}
+	cancel()
+	if err := grp.Wait(); err != nil {
+		log.Println(err)
 	}
+	log.Println("done")
 }

examples/3-messaging/collector/collector.go

@@ -1,6 +1,7 @@
 package collector
 
 import (
+	"context"
 	"encoding/json"
 
 	"github.com/lovoo/goka"
@@ -41,14 +42,16 @@ func collect(ctx goka.Context, msg interface{}) {
 	ctx.SetValue(ml)
 }
 
-func Run(brokers []string) {
-	g := goka.DefineGroup(group,
-		goka.Input(messaging.ReceivedStream, new(messaging.MessageCodec), collect),
-		goka.Persist(new(MessageListCodec)),
-	)
-	if p, err := goka.NewProcessor(brokers, g); err != nil {
-		panic(err)
-	} else if err = p.Start(); err != nil {
-		panic(err)
+func Run(ctx context.Context, brokers []string) func() error {
+	return func() error {
+		g := goka.DefineGroup(group,
+			goka.Input(messaging.ReceivedStream, new(messaging.MessageCodec), collect),
+			goka.Persist(new(MessageListCodec)),
+		)
+		p, err := goka.NewProcessor(brokers, g)
+		if err != nil {
+			return err
+		}
+		return p.Run(ctx)
 	}
 }

examples/3-messaging/detector/detector.go

@@ -1,6 +1,7 @@
 package detector
 
 import (
+	"context"
 	"encoding/json"
 
 	"github.com/lovoo/goka"
@@ -48,38 +49,41 @@ func detectSpammer(ctx goka.Context, c *Counters) bool {
 	return total >= minMessages && rate >= maxRate
 }
 
-func Run(brokers []string) {
-	g := goka.DefineGroup(group,
-		goka.Input(messaging.SentStream, new(messaging.MessageCodec), func(ctx goka.Context, msg interface{}) {
-			c := getValue(ctx)
-			c.Sent++
-			ctx.SetValue(c)
+func Run(ctx context.Context, brokers []string) func() error {
+	return func() error {
+		g := goka.DefineGroup(group,
+			goka.Input(messaging.SentStream, new(messaging.MessageCodec), func(ctx goka.Context, msg interface{}) {
+				c := getValue(ctx)
+				c.Sent++
+				ctx.SetValue(c)
 
-			// check if sender is a spammer
-			if detectSpammer(ctx, c) {
-				ctx.Emit(blocker.Stream, ctx.Key(), new(blocker.BlockEvent))
-			}
+				// check if sender is a spammer
+				if detectSpammer(ctx, c) {
+					ctx.Emit(blocker.Stream, ctx.Key(), new(blocker.BlockEvent))
+				}
 
-			// Loop to receiver
-			m := msg.(*messaging.Message)
-			ctx.Loopback(m.To, m)
-		}),
-		goka.Loop(new(messaging.MessageCodec), func(ctx goka.Context, msg interface{}) {
-			c := getValue(ctx)
-			c.Received++
-			ctx.SetValue(c)
+				// Loop to receiver
+				m := msg.(*messaging.Message)
+				ctx.Loopback(m.To, m)
+			}),
+			goka.Loop(new(messaging.MessageCodec), func(ctx goka.Context, msg interface{}) {
+				c := getValue(ctx)
+				c.Received++
+				ctx.SetValue(c)
 
-			// check if receiver is a spammer
-			if detectSpammer(ctx, c) {
-				ctx.Emit(blocker.Stream, ctx.Key(), new(blocker.BlockEvent))
-			}
-		}),
-		goka.Output(blocker.Stream, new(blocker.BlockEventCodec)),
-		goka.Persist(new(CountersCodec)),
-	)
-	if p, err := goka.NewProcessor(brokers, g); err != nil {
-		panic(err)
-	} else if err = p.Start(); err != nil {
-		panic(err)
+				// check if receiver is a spammer
+				if detectSpammer(ctx, c) {
+					ctx.Emit(blocker.Stream, ctx.Key(), new(blocker.BlockEvent))
+				}
+			}),
+			goka.Output(blocker.Stream, new(blocker.BlockEventCodec)),
+			goka.Persist(new(CountersCodec)),
+		)
+		p, err := goka.NewProcessor(brokers, g)
+		if err != nil {
+			return err
+		}
+
+		return p.Run(ctx)
 	}
 }