package main

import (
	"crypto/tls"
	"flag"
	"fmt"
	"os"

	"github.com/cloudfoundry/noaa/consumer"
	"github.com/cloudfoundry/sonde-go/events"
)

const firehoseSubscriptionId = "firehose-a"

var (
	dopplerAddress = os.Getenv("DOPPLER_ADDR")
	authToken      = os.Getenv("CF_ACCESS_TOKEN")
)

func main() {
	filterType := flag.String("filter", "all", "filter messages by 'logs' or 'metrics' (default: all)")
	flag.Parse()

	cnsmr := consumer.New(dopplerAddress, &tls.Config{InsecureSkipVerify: true}, nil)
	cnsmr.SetDebugPrinter(ConsoleDebugPrinter{})

	fmt.Println("===== Streaming Firehose (will only succeed if you have admin credentials)")

	var (
		msgChan   <-chan *events.Envelope
		errorChan <-chan error
	)

	switch *filterType {
	case "logs":
		msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.LogMessages)
	case "metrics":
		msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.Metrics)
	default:
		msgChan, errorChan = cnsmr.Firehose(firehoseSubscriptionId, authToken)
	}

	go func() {
		for err := range errorChan {
			fmt.Fprintf(os.Stderr, "%v\n", err.Error())
		}
	}()

	for msg := range msgChan {
		fmt.Printf("%v \n", msg)
	}
}

type ConsoleDebugPrinter struct{}

func (c ConsoleDebugPrinter) Print(title, dump string) {
	println(title)
	println(dump)
}

package main

import (
	"crypto/tls"
	"fmt"
	"os"

	"github.com/cloudfoundry/noaa/consumer"
)

const firehoseSubscriptionId = "firehose-a"

var (
	dopplerAddress = os.Getenv("DOPPLER_ADDR")
	authToken      = os.Getenv("CF_ACCESS_TOKEN")
)

func main() {
	consumer := consumer.New(dopplerAddress, &tls.Config{InsecureSkipVerify: true}, nil)
	consumer.SetDebugPrinter(ConsoleDebugPrinter{})

	fmt.Println("===== Streaming Firehose (will only succeed if you have admin credentials)")

	msgChan, errorChan := consumer.Firehose(firehoseSubscriptionId, authToken)
	go func() {
		for err := range errorChan {
			fmt.Fprintf(os.Stderr, "%v\n", err.Error())
		}
	}()

	for msg := range msgChan {
		fmt.Printf("%v \n", msg)
	}
}

type ConsoleDebugPrinter struct{}

func (c ConsoleDebugPrinter) Print(title, dump string) {
	println(title)
	println(dump)
}

--- 
+++ 
@@ -2,10 +2,12 @@
 
 import (
 	"crypto/tls"
+	"flag"
 	"fmt"
 	"os"
 
 	"github.com/cloudfoundry/noaa/consumer"
+	"github.com/cloudfoundry/sonde-go/events"
 )
 
 const firehoseSubscriptionId = "firehose-a"
@@ -16,12 +18,28 @@
 )
 
 func main() {
-	consumer := consumer.New(dopplerAddress, &tls.Config{InsecureSkipVerify: true}, nil)
-	consumer.SetDebugPrinter(ConsoleDebugPrinter{})
+	filterType := flag.String("filter", "all", "filter messages by 'logs' or 'metrics' (default: all)")
+	flag.Parse()
+
+	cnsmr := consumer.New(dopplerAddress, &tls.Config{InsecureSkipVerify: true}, nil)
+	cnsmr.SetDebugPrinter(ConsoleDebugPrinter{})
 
 	fmt.Println("===== Streaming Firehose (will only succeed if you have admin credentials)")
 
-	msgChan, errorChan := consumer.Firehose(firehoseSubscriptionId, authToken)
+	var (
+		msgChan   <-chan *events.Envelope
+		errorChan <-chan error
+	)
+
+	switch *filterType {
+	case "logs":
+		msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.LogMessages)
+	case "metrics":
+		msgChan, errorChan = cnsmr.FilteredFirehose(firehoseSubscriptionId, authToken, consumer.Metrics)
+	default:
+		msgChan, errorChan = cnsmr.Firehose(firehoseSubscriptionId, authToken)
+	}
+
 	go func() {
 		for err := range errorChan {
 			fmt.Fprintf(os.Stderr, "%v\n", err.Error())
Add example of firehose filtering by envelope type

[#145888141]

Signed-off-by: Mark Kollasch <8a586b113574ae1b67073fe50ce7d0da2ec8bc53@pivotal.io>
