Filter subscribe on --product.

.gitignore

@@ -13,8 +13,8 @@
 coverage.*
 
 # Runtime
-*.db
 *.sqlite
+events.db
 mms_authorized_keys
 
 # Config

Makefile

@@ -39,4 +39,11 @@ release:
 puml:
 	go-plantuml generate -rd . -o go-mms.puml
 
+integration_test: build_mmsd
+	./mmsd -w ./test_data 2>/dev/null & echo "$$!" > ./mmsd.pid
+	go test --tags=integration ./cmd/mms/ || (kill `cat ./mmsd.pid`; unlink ./mmsd.pid; exit 1)
+
+	@kill `cat ./mmsd.pid`
+	@unlink ./mmsd.pid
+
 .PHONY: deps go_mod build_mmsd build_mms test image release puml static

cmd/mms/commands.go

@@ -30,124 +30,113 @@ import (
 	"github.com/urfave/cli/v2"
 )
 
-func listAllEvents() func(*cli.Context) error {
-	return func(ctx *cli.Context) error {
-		events := []*mms.ProductEvent{}
-		if ctx.String("production-hub") == "" {
-			return fmt.Errorf("No production-hub specified")
-		}
-		url := ctx.String("production-hub") + "/api/v1/events"
-		newEvents, err := mms.ListProductEvents(url)
-		if err != nil {
-			return fmt.Errorf("failed to access events: %v", err)
-		}
-		events = append(events, newEvents...)
-
-		for _, event := range events {
-			fmt.Printf("Event: %+v\n", event)
-		}
-		return nil
+func listAllEventsCmd(ctx *cli.Context) error {
+	events := []*mms.ProductEvent{}
+	if ctx.String("production-hub") == "" {
+		return fmt.Errorf("No production-hub specified")
+	}
+	url := ctx.String("production-hub") + "/api/v1/events"
+	newEvents, err := mms.ListProductEvents(url)
+	if err != nil {
+		return fmt.Errorf("failed to access events: %v", err)
 	}
+	events = append(events, newEvents...)
+	for _, event := range events {
+		fmt.Printf("Event: %+v\n", event)
+	}
+	return nil
 }
 
-func subscribeEvents() func(*cli.Context) error {
-	return func(ctx *cli.Context) error {
-		errChannel := make(chan error, 1)
-		go func(ctx *cli.Context) {
-			mmsClient, err := mms.NewNatsConsumerClient(ctx.String("production-hub"))
-			if err != nil {
-				errChannel <- err
-				return
-			}
-			if ctx.String("command") != "None" {
-				callback := createExecutableCallback(ctx.String("command"), ctx.Bool("args"))
-				mmsClient.WatchProductEvents(callback)
-			} else {
-				// Same as Aviso-echo
-				mmsClient.WatchProductEvents(productReceiver)
-			}
-
-		}(ctx)
-		select {
-		case err := <-errChannel:
-			return fmt.Errorf("one hub event subscription failed, ending: %v", err)
-		}
+func subscribeEventsCmd(ctx *cli.Context) error {
+	mmsClient, err := mms.NewNatsConsumerClient(ctx.String("production-hub"))
+	if err != nil {
+		return fmt.Errorf("one hub event subscription failed, ending: %v", err)
 	}
-}
 
-func postEvent() func(*cli.Context) error {
-	return func(ctx *cli.Context) error {
-		var err error
-
-		refTime := time.Now()
-		if ctx.String("reftime") != "now" {
-			refTime, err = time.Parse(time.RFC3339, ctx.String("reftime"))
-			if err != nil {
-				log.Println("Could not parse reftime")
-				log.Println("Please use RFC 3339 format:")
-				log.Println("- '2006-01-02T15:04:05Z' for UTC")
-				log.Println("- '2006-01-02T15:04:05+01:00' for other time zones")
-				log.Fatalf("Parser error: %v", err)
-			}
-		}
+	if ctx.String("command") != "None" {
+		callback := createExecutableCallback(ctx.String("command"), ctx.Bool("args"), ctx.String("product"))
+		mmsClient.WatchProductEvents(callback)
+	} else {
+		// Same as Aviso-echo
+		mmsClient.WatchProductEvents(productReceiver(ctx.String("product")))
+	}
 
-		productEvent := mms.ProductEvent{
-			JobName:         ctx.String("jobname"),
-			Product:         ctx.String("product"),
-			ProductLocation: ctx.String("product-location"),
-			ProductionHub:   ctx.String("production-hub"),
-			Counter:         ctx.Int("counter"),
-			TotalCount:      ctx.Int("ntotal"),
-			RefTime:         refTime,
-			CreatedAt:       time.Now(),
-			NextEventAt:     time.Now().Add(time.Second * time.Duration(ctx.Int("event-interval"))),
-		}
+	return nil
+}
 
-		if ctx.String("production-hub") == "" {
-			return fmt.Errorf("No production-hub specified")
+func postEventCmd(ctx *cli.Context) error {
+	var err error
+	refTime := time.Now()
+	if ctx.String("reftime") != "now" {
+		refTime, err = time.Parse(time.RFC3339, ctx.String("reftime"))
+		if err != nil {
+			log.Println("Could not parse reftime")
+			log.Println("Please use RFC 3339 format:")
+			log.Println("- '2006-01-02T15:04:05Z' for UTC")
+			log.Println("- '2006-01-02T15:04:05+01:00' for other time zones")
+			log.Fatalf("Parser error: %v", err)
 		}
-		url := ctx.String("production-hub") + "/api/v1/events"
-
-		// Create a json-payload from productEvent
-		jsonStr, err := json.Marshal(&productEvent)
-		// Create a http-request to post the payload
-		httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonStr))
-
-		httpReq.Header.Set("Api-Key", ctx.String("api-key"))
-		httpReq.Header.Set("Content-Type", "application/json")
-
-		// Create a http connection to the api.
-		var tr *http.Transport
-		if ctx.Bool("insecure") {
-			tr = &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			}
-		} else {
-			tr = &http.Transport{}
+	}
+	productEvent := mms.ProductEvent{
+		JobName:         ctx.String("jobname"),
+		Product:         ctx.String("product"),
+		ProductLocation: ctx.String("product-location"),
+		ProductionHub:   ctx.String("production-hub"),
+		Counter:         ctx.Int("counter"),
+		TotalCount:      ctx.Int("ntotal"),
+		RefTime:         refTime,
+		CreatedAt:       time.Now(),
+		NextEventAt:     time.Now().Add(time.Second * time.Duration(ctx.Int("event-interval"))),
+	}
+	if ctx.String("production-hub") == "" {
+		return fmt.Errorf("No production-hub specified")
+	}
+	url := ctx.String("production-hub") + "/api/v1/events"
+	// Create a json-payload from productEvent
+	jsonStr, err := json.Marshal(&productEvent)
+	// Create a http-request to post the payload
+	httpReq, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonStr))
+	httpReq.Header.Set("Api-Key", ctx.String("api-key"))
+	httpReq.Header.Set("Content-Type", "application/json")
+	// Create a http connection to the api.
+	var tr *http.Transport
+	if ctx.Bool("insecure") {
+		tr = &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
+	} else {
+		tr = &http.Transport{}
+	}
+	httpClient := &http.Client{Transport: tr}
+	httpResp, err := httpClient.Do(httpReq)
+	if err != nil {
+		log.Fatalf("Failed to create http client: %v", err)
+	}
+	defer httpResp.Body.Close()
+	// If 201 is not returned, panic with http response
+	if httpResp.StatusCode != http.StatusCreated {
+		log.Fatalln(httpResp.Status)
+	}
+	return nil
+}
 
-		httpClient := &http.Client{Transport: tr}
-		httpResp, err := httpClient.Do(httpReq)
-		if err != nil {
-			log.Fatalf("Failed to create http client: %v", err)
+func productReceiver(product string) func(event *mms.ProductEvent) error {
+	return func(event *mms.ProductEvent) error {
+		if product != "" && event.Product != product {
+			return nil
 		}
-		defer httpResp.Body.Close()
 
-		// If 201 is not returned, panic with http response
-		if httpResp.StatusCode != http.StatusCreated {
-			log.Fatalln(httpResp.Status)
+		encoded, err := json.Marshal(event)
+		if err != nil {
+			return fmt.Errorf("failed to encode event as json: %s", err)
 		}
 
+		fmt.Println(string(encoded))
 		return nil
 	}
 }
 
-func productReceiver(event *mms.ProductEvent) error {
-	fmt.Println(event)
-	return nil
-}
-
-func createExecutableCallback(filepath string, args bool) func(event *mms.ProductEvent) error {
+func createExecutableCallback(filepath string, args bool, product string) func(event *mms.ProductEvent) error {
 	_, err := exec.LookPath(filepath)
 
 	if err != nil {
@@ -157,6 +146,10 @@ func createExecutableCallback(filepath string, args bool) func(event *mms.Produc
 	return func(event *mms.ProductEvent) error {
 		var productLocation string
 
+		if product != "" && event.Product != product {
+			return nil
+		}
+
 		if args {
 			productLocation = event.ProductLocation
 		} else {

cmd/mms/integration_test.go

@@ -0,0 +1,94 @@
+// +build integration
+
+package main
+
+import (
+	"bytes"
+	"io"
+	"log"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"encoding/json"
+	"github.com/metno/go-mms/pkg/mms"
+)
+
+func TestHelpOption(t *testing.T) {
+	args := os.Args[0:1]
+	args = append(args, "--help")
+
+	output := captureOutput(args, run)
+	expected := "USAGE"
+	if !strings.Contains(output, expected) {
+		t.Errorf("Expected %s; Got %s", expected, output)
+	}
+}
+
+func TestFilteredSubscribe(t *testing.T) {
+	subscribeArgs := os.Args[0:1]
+	subscribeArgs = append(subscribeArgs, "subscribe", "--production-hub", "nats://localhost:4222", "--product", "good")
+
+	go run(subscribeArgs)
+
+	postArgsGood := os.Args[0:1]
+	postArgsGood = append(postArgsGood, "post", "--production-hub", "http://localhost:8080", "--product", "good", "--api-key", "97fIjjoKsYxFiJd67EpC1VuZuFPTNUqQv9eTuKEyRXQ=")
+	output := captureOutput(postArgsGood, run)
+
+	goodEvent := mms.ProductEvent{}
+	err := json.Unmarshal([]byte(output), &goodEvent)
+	if err != nil {
+		t.Errorf("Expected ok unmarshal; Got error; %s, from output %s", err, output)
+		return
+	}
+	if goodEvent.Product != "good" {
+		t.Errorf("Expected event.Product: good; Got %s", goodEvent.Product)
+		return
+	}
+
+	postArgsBad := os.Args[0:1]
+	postArgsBad = append(postArgsBad, "post", "--production-hub", "http://localhost:8080", "--product", "bad", "--api-key", "97fIjjoKsYxFiJd67EpC1VuZuFPTNUqQv9eTuKEyRXQ=")
+	output = captureOutput(postArgsBad, run)
+
+	var badEvent mms.ProductEvent
+	err = json.Unmarshal([]byte(output), &badEvent)
+	if err == nil {
+		t.Errorf("Expected empty output from stdout; Got valid json instead: %s", output)
+		return
+	}
+}
+
+// captureOutput captures all output to stdout and stderr after call f with args.
+// Waits 100 millseconds and returns a string will all stdout and stderr output.
+func captureOutput(args []string, f func([]string) error) string {
+	reader, writer, err := os.Pipe()
+	if err != nil {
+		panic(err)
+	}
+	stdout := os.Stdout
+	stderr := os.Stderr
+	defer func() {
+		os.Stdout = stdout
+		os.Stderr = stderr
+		log.SetOutput(os.Stderr)
+	}()
+	os.Stdout = writer
+	os.Stderr = writer
+	log.SetOutput(writer)
+	out := make(chan string)
+	wg := new(sync.WaitGroup)
+	wg.Add(1)
+	go func() {
+		var buf bytes.Buffer
+		wg.Done()
+		io.Copy(&buf, reader)
+		out <- buf.String()
+	}()
+	wg.Wait()
+	f(args)
+	time.Sleep(100 * time.Millisecond)
+	writer.Close()
+	return <-out
+}

cmd/mms/main.go

@@ -24,8 +24,7 @@ import (
 	"github.com/urfave/cli/v2/altsrc"
 )
 
-func main() {
-
+func run(args []string) error {
 	// Default file name for config
 	// Could be expanded to check and pick a file from a pre-defined list
 	var confFile string = "mms_config.yml"
@@ -48,6 +47,11 @@ func main() {
 			Value:   "None",
 			Aliases: []string{"cmd"},
 		}),
+		altsrc.NewStringFlag(&cli.StringFlag{
+			Name:    "product",
+			Usage:   "Name of the product.",
+			EnvVars: []string{"MMS_PRODUCT"},
+		}),
 		altsrc.NewBoolFlag(&cli.BoolFlag{
 			Name:  "args",
 			Usage: "Toggles sending of productLocation as arg[1] in executable",
@@ -126,14 +130,14 @@ func main() {
 				Aliases: []string{"ls"},
 				Usage:   "List all the latest available events in the system.",
 				Flags:   listFlags,
-				Action:  listAllEvents(),
+				Action:  listAllEventsCmd,
 			},
 			{
 				Name:    "subscribe",
 				Aliases: []string{"s"},
 				Usage:   "Listen for new incoming events, get them printed continuously.",
 				Flags:   subscriptionFlags,
-				Action:  subscribeEvents(),
+				Action:  subscribeEventsCmd,
 			},
 			{
 				Name:    "post",
@@ -149,12 +153,16 @@ func main() {
 					return altsrc.ApplyInputSourceValues(ctx, inputSource, postFlags)
 				},
 				Flags:  postFlags,
-				Action: postEvent(),
+				Action: postEventCmd,
 			},
 		},
 	}
 
-	err := app.Run(os.Args)
+	return app.Run(args)
+}
+
+func main() {
+	err := run(os.Args)
 	if err != nil {
 		log.Fatal(err)
 	}