func initiateFirehoseConnection(tcPort int) (*consumer.Consumer, <-chan *events.Envelope) {
	localIP, _ := localip.LocalIP()
	url := fmt.Sprintf("ws://%s:%d", localIP, tcPort)
	firehoseConnection := consumer.New(url, &tls.Config{InsecureSkipVerify: true}, nil)
	msgChan, _ := firehoseConnection.Firehose("uniqueId", "")
	return firehoseConnection, msgChan
}

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

	messages, err := consumer.RecentLogs(appGuid, authToken)

	if err != nil {
		fmt.Printf("===== Error getting recent messages: %v\n", err)
	} else {
		fmt.Println("===== Recent logs")
		for _, msg := range messages {
			fmt.Println(msg)
		}
	}

	fmt.Println("===== Streaming metrics")
	msgChan, errorChan := consumer.Stream(appGuid, authToken)

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

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

func main() {

	uaa, err := uaago.NewClient(uaaEndpoint)
	if err != nil {
		fmt.Printf("Error from uaaClient %s\n", err)
		os.Exit(1)
	}

	refresher := tokenRefresher{uaaClient: uaa}

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

	fmt.Println("===== Streaming metrics")
	msgChan, errorChan := consumer.Firehose(appGuid, "")

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

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

func (o *OpenTSDBFirehoseNozzle) consumeFirehose(authToken string) {
	o.consumer = consumer.New(
		o.config.TrafficControllerURL,
		&tls.Config{InsecureSkipVerify: o.config.InsecureSSLSkipVerify},
		nil)
	o.consumer.SetIdleTimeout(time.Duration(o.config.IdleTimeoutSeconds) * time.Second)
	o.messages, o.errs = o.consumer.Firehose(o.config.FirehoseSubscriptionID, authToken)
}

func SetUpConsumer() (*consumer.Consumer, *TestDebugPrinter) {
	tlsConfig := tls.Config{InsecureSkipVerify: config.SkipSSLVerify}
	printer := &TestDebugPrinter{}

	connection := consumer.New(config.DopplerEndpoint, &tlsConfig, nil)
	connection.SetDebugPrinter(printer)
	return connection, printer
}

func CreateFirehoseChan(DopplerEndpoint string, Token string, subId string, skipSSLValidation bool, keepAlive time.Duration) <-chan *events.Envelope {
	consumer.KeepAlive = keepAlive
	connection := consumer.New(DopplerEndpoint, &tls.Config{InsecureSkipVerify: skipSSLValidation}, nil)
	connection.SetDebugPrinter(ConsoleDebugPrinter{})
	msgChan, errorChan := connection.Firehose(subId, Token)
	go func() {
		for err := range errorChan {
			log.LogError("Firehose Error!", err.Error())
		}
	}()
	return msgChan
}

func main() {
	cfg, err := config.ParseEnv()
	if err != nil {
		log.Fatal(err)
	}
	uaa := &cflib.UAA{
		URL:          cfg.UaaURL,
		Username:     cfg.Username,
		Password:     cfg.Password,
		ClientID:     cfg.ClientID,
		ClientSecret: cfg.ClientSecret,
	}
	cc := &cflib.CC{
		URL: cfg.ApiURL,
	}
	logcounter := logcounter.New(uaa, cc, cfg)

	go func() {
		if err := logcounter.Start(); err != nil {
			log.Fatal(err)
		}
	}()

	consumer := consumer.New(cfg.DopplerURL, &tls.Config{InsecureSkipVerify: true}, nil)

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

	// notify on ctrl+c
	terminate := make(chan os.Signal, 1)
	signal.Notify(terminate, os.Interrupt)
	go func() {
		for range terminate {
			logcounter.Stop()
		}
	}()

	for {
		authToken, err := uaa.GetAuthToken()
		if err != nil || authToken == "" {
			fmt.Fprintf(os.Stderr, "error getting token %s\n", err)
			continue
		}
		fmt.Println("got new oauth token")
		msgs, errors := consumer.FirehoseWithoutReconnect(cfg.SubscriptionID, authToken)

		go logcounter.HandleMessages(msgs)
		done := logcounter.HandleErrors(errors, terminate, consumer)
		if done {
			return
		}
	}
}

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

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

	for {
		containerMetrics, err := consumer.ContainerMetrics(appId, authToken)

		for _, cm := range containerMetrics {
			fmt.Printf("%v \n", cm)
		}

		if err != nil {
			fmt.Fprintf(os.Stderr, "%v\n", err.Error())
		}

		time.Sleep(3 * time.Second)
	}

}

	BeforeEach(func() {
		internal.Timeout = testTimeout

		fakeHandler = make(nullHandler, 1)
		testServer = httptest.NewServer(fakeHandler)
	})

	AfterEach(func() {
		cnsmr.Close()
	})

	Describe("TailingLogsWithoutReconnect", func() {
		It("times out due to handshake timeout", func() {
			defer close(fakeHandler)
			cnsmr = consumer.New(strings.Replace(testServer.URL, "http", "ws", 1), nil, nil)

			_, errCh := cnsmr.TailingLogsWithoutReconnect(appGuid, authToken)
			var err error
			Eventually(errCh, 2*testTimeout).Should(Receive(&err))
			Expect(err.Error()).To(ContainSubstring("i/o timeout"))
		})
	})

	Describe("Stream", func() {
		It("times out due to handshake timeout", func() {
			defer close(fakeHandler)

			cnsmr = consumer.New(strings.Replace(testServer.URL, "http", "ws", 1), nil, nil)

			_, errCh := cnsmr.Stream(appGuid, authToken)

		dropsondeEndpoint = fmt.Sprintf("ws://%s:%d", localIPAddress, TRAFFIC_CONTROLLER_DROPSONDE_PORT)
	})

	AfterEach(func() {
		fakeDoppler.Stop()
	})

	Context("Streaming", func() {
		var (
			client   *consumer.Consumer
			messages <-chan *events.Envelope
			errors   <-chan error
		)

		JustBeforeEach(func() {
			client = consumer.New(dropsondeEndpoint, &tls.Config{}, nil)
			messages, errors = client.StreamWithoutReconnect(APP_ID, AUTH_TOKEN)
		})

		It("passes messages through", func() {
			var request *http.Request
			Eventually(fakeDoppler.TrafficControllerConnected, 10).Should(Receive(&request))
			Expect(request.URL.Path).To(Equal("/apps/1234/stream"))

			currentTime := time.Now().UnixNano()
			dropsondeMessage := makeDropsondeMessage("Hello through NOAA", APP_ID, currentTime)
			fakeDoppler.SendLogMessage(dropsondeMessage)

			var receivedEnvelope *events.Envelope
			Eventually(messages).Should(Receive(&receivedEnvelope))
			Consistently(errors).ShouldNot(Receive())

func NewRepositoryLocator(config coreconfig.ReadWriter, gatewaysByName map[string]net.Gateway, logger trace.Printer, envDialTimeout string) (loc RepositoryLocator) {
	strategy := strategy.NewEndpointStrategy(config.APIVersion())

	cloudControllerGateway := gatewaysByName["cloud-controller"]
	routingAPIGateway := gatewaysByName["routing-api"]
	uaaGateway := gatewaysByName["uaa"]
	loc.authRepo = authentication.NewUAARepository(uaaGateway, config, net.NewRequestDumper(logger))

	// ensure gateway refreshers are set before passing them by value to repositories
	cloudControllerGateway.SetTokenRefresher(loc.authRepo)
	uaaGateway.SetTokenRefresher(loc.authRepo)

	loc.appBitsRepo = applicationbits.NewCloudControllerApplicationBitsRepository(config, cloudControllerGateway)
	loc.appEventsRepo = appevents.NewCloudControllerAppEventsRepository(config, cloudControllerGateway, strategy)
	loc.appFilesRepo = api_appfiles.NewCloudControllerAppFilesRepository(config, cloudControllerGateway)
	loc.appRepo = applications.NewCloudControllerRepository(config, cloudControllerGateway)
	loc.appSummaryRepo = NewCloudControllerAppSummaryRepository(config, cloudControllerGateway)
	loc.appInstancesRepo = appinstances.NewCloudControllerAppInstancesRepository(config, cloudControllerGateway)
	loc.authTokenRepo = NewCloudControllerServiceAuthTokenRepository(config, cloudControllerGateway)
	loc.curlRepo = NewCloudControllerCurlRepository(config, cloudControllerGateway)
	loc.domainRepo = NewCloudControllerDomainRepository(config, cloudControllerGateway, strategy)
	loc.endpointRepo = NewEndpointRepository(cloudControllerGateway)

	tlsConfig := net.NewTLSConfig([]tls.Certificate{}, config.IsSSLDisabled())

	apiVersion, _ := semver.Make(config.APIVersion())

	var noaaRetryTimeout time.Duration
	convertedTime, err := strconv.Atoi(envDialTimeout)
	if err != nil {
		noaaRetryTimeout = noaaRetryDefaultTimeout
	} else {
		noaaRetryTimeout = time.Duration(convertedTime) * 3 * time.Second
	}

	if apiVersion.GTE(cf.NoaaMinimumAPIVersion) {
		consumer := consumer.New(config.DopplerEndpoint(), tlsConfig, http.ProxyFromEnvironment)
		consumer.SetDebugPrinter(terminal.DebugPrinter{Logger: logger})
		loc.logsRepo = logs.NewNoaaLogsRepository(config, consumer, loc.authRepo, noaaRetryTimeout)
	} else {
		consumer := loggregator_consumer.New(config.LoggregatorEndpoint(), tlsConfig, http.ProxyFromEnvironment)
		consumer.SetDebugPrinter(terminal.DebugPrinter{Logger: logger})
		loc.logsRepo = logs.NewLoggregatorLogsRepository(config, consumer, loc.authRepo)
	}

	loc.organizationRepo = organizations.NewCloudControllerOrganizationRepository(config, cloudControllerGateway)
	loc.passwordRepo = password.NewCloudControllerRepository(config, uaaGateway)
	loc.quotaRepo = quotas.NewCloudControllerQuotaRepository(config, cloudControllerGateway)
	loc.routeRepo = NewCloudControllerRouteRepository(config, cloudControllerGateway)
	loc.routeServiceBindingRepo = NewCloudControllerRouteServiceBindingRepository(config, cloudControllerGateway)
	loc.routingAPIRepo = NewRoutingAPIRepository(config, routingAPIGateway)
	loc.stackRepo = stacks.NewCloudControllerStackRepository(config, cloudControllerGateway)
	loc.serviceRepo = NewCloudControllerServiceRepository(config, cloudControllerGateway)
	loc.serviceKeyRepo = NewCloudControllerServiceKeyRepository(config, cloudControllerGateway)
	loc.serviceBindingRepo = NewCloudControllerServiceBindingRepository(config, cloudControllerGateway)
	loc.serviceBrokerRepo = NewCloudControllerServiceBrokerRepository(config, cloudControllerGateway)
	loc.servicePlanRepo = NewCloudControllerServicePlanRepository(config, cloudControllerGateway)
	loc.servicePlanVisibilityRepo = NewCloudControllerServicePlanVisibilityRepository(config, cloudControllerGateway)
	loc.serviceSummaryRepo = NewCloudControllerServiceSummaryRepository(config, cloudControllerGateway)
	loc.spaceRepo = spaces.NewCloudControllerSpaceRepository(config, cloudControllerGateway)
	loc.userProvidedServiceInstanceRepo = NewCCUserProvidedServiceInstanceRepository(config, cloudControllerGateway)
	loc.userRepo = NewCloudControllerUserRepository(config, uaaGateway, cloudControllerGateway)
	loc.buildpackRepo = NewCloudControllerBuildpackRepository(config, cloudControllerGateway)
	loc.buildpackBitsRepo = NewCloudControllerBuildpackBitsRepository(config, cloudControllerGateway, appfiles.ApplicationZipper{})
	loc.securityGroupRepo = securitygroups.NewSecurityGroupRepo(config, cloudControllerGateway)
	loc.stagingSecurityGroupRepo = staging.NewSecurityGroupsRepo(config, cloudControllerGateway)
	loc.runningSecurityGroupRepo = running.NewSecurityGroupsRepo(config, cloudControllerGateway)
	loc.securityGroupSpaceBinder = securitygroupspaces.NewSecurityGroupSpaceBinder(config, cloudControllerGateway)
	loc.spaceQuotaRepo = spacequotas.NewCloudControllerSpaceQuotaRepository(config, cloudControllerGateway)
	loc.featureFlagRepo = featureflags.NewCloudControllerFeatureFlagRepository(config, cloudControllerGateway)
	loc.environmentVariableGroupRepo = environmentvariablegroups.NewCloudControllerRepository(config, cloudControllerGateway)
	loc.copyAppSourceRepo = copyapplicationsource.NewCloudControllerCopyApplicationSourceRepository(config, cloudControllerGateway)

	client := v3client.NewClient(config.APIEndpoint(), config.AuthenticationEndpoint(), config.AccessToken(), config.RefreshToken())
	loc.v3Repository = repository.NewRepository(config, client)

	return
}

	BeforeEach(func() {
		messagesToSend = make(chan []byte, 256)
		testServer = httptest.NewServer(handlers.NewWebsocketHandler(messagesToSend, 100*time.Millisecond, loggertesthelper.Logger()))
		endpoint = "ws://" + testServer.Listener.Addr().String()

		goProxyHandler = goproxy.NewProxyHttpServer()
		goProxyHandler.Logger = log.New(bytes.NewBufferString(""), "", 0)
		testProxyServer = httptest.NewServer(goProxyHandler)
		proxy = func(*http.Request) (*url.URL, error) {
			return url.Parse(testProxyServer.URL)
		}
	})

	JustBeforeEach(func() {
		connection = consumer.New(endpoint, nil, proxy)
	})

	AfterEach(func() {
		testProxyServer.Close()
		testServer.Close()
	})

	Describe("StreamWithoutReconnect", func() {
		var (
			incoming <-chan *events.Envelope
			errs     <-chan error
		)

		JustBeforeEach(func() {
			incoming, errs = connection.StreamWithoutReconnect("fakeAppGuid", "authToken")

			contents, err := ioutil.ReadFile(accessLogFile)
			Expect(err).ToNot(HaveOccurred())
			return string(contents)
		}
	})

	AfterEach(func() {
		Expect(os.Remove(accessLogFile)).To(Succeed())
	})

	Context("with modern endpoints", func() {
		var noaaConsumer *consumer.Consumer

		JustBeforeEach(func() {
			tcURL := fmt.Sprintf("ws://%s:%d", localIPAddress, TRAFFIC_CONTROLLER_DROPSONDE_PORT)
			noaaConsumer = consumer.New(tcURL, &tls.Config{}, nil)
		})

		AfterEach(func() {
			noaaConsumer.Close()
		})

		It("logs stream access", func() {
			noaaConsumer.Stream(APP_ID, AUTH_TOKEN)

			expected := fmt.Sprintf("CEF:0|cloud_foundry|loggregator_trafficcontroller|1.0|GET /apps/%s/stream|GET /apps/%[1]s/stream|0|", APP_ID)
			Eventually(testContents).Should(ContainSubstring(expected))
		})

		It("logs recent access", func() {
			noaaConsumer.RecentLogs(APP_ID, AUTH_TOKEN)

		messagesToSend chan []byte
	)

	BeforeEach(func() {
		trafficControllerURL = ""
		testServer = nil
		fakeHandler = nil
		tlsSettings = nil

		appGuid = ""
		authToken = ""
		messagesToSend = make(chan []byte, 256)
	})

	JustBeforeEach(func() {
		cnsmr = consumer.New(trafficControllerURL, tlsSettings, nil)
	})

	AfterEach(func() {
		cnsmr.Close()
		if testServer != nil {
			testServer.Close()
		}
	})

	Describe("SetOnConnectCallback", func() {
		BeforeEach(func() {
			testServer = httptest.NewServer(handlers.NewWebsocketHandler(messagesToSend, 100*time.Millisecond, loggertesthelper.Logger()))
			trafficControllerURL = "ws://" + testServer.Listener.Addr().String()
			close(messagesToSend)
		})

	"github.com/gogo/protobuf/proto"
)

const (
	cfSetupTimeOut     = 10 * time.Second
	cfPushTimeOut      = 2 * time.Minute
	defaultMemoryLimit = "256MB"
)

var _ = Describe("Logs", func() {
	It("gets through recent logs", func() {
		env := createLogEnvelope("I AM A BANANA!", "foo")
		helpers.EmitToMetron(env)

		tlsConfig := &tls.Config{InsecureSkipVerify: true}
		consumer := consumer.New(config.DopplerEndpoint, tlsConfig, nil)

		getRecentLogs := func() []*events.LogMessage {
			envelopes, err := consumer.RecentLogs("foo", "")
			Expect(err).NotTo(HaveOccurred())
			return envelopes
		}

		Eventually(getRecentLogs).Should(ContainElement(env.LogMessage))
	})

	It("sends log messages for a specific app through the stream endpoint", func() {
		msgChan, errorChan := helpers.ConnectToStream("foo")

		helpers.EmitToMetron(createLogEnvelope("Stream message", "bar"))
		helpers.EmitToMetron(createLogEnvelope("Stream message", "foo"))

func Logs(cliConnection plugin.CliConnection, args []string) {
	appName := args[1]
	rawOutput, _ := cliConnection.CliCommandWithoutTerminalOutput("curl", fmt.Sprintf("/v3/apps?names=%s", appName))
	apps := V3AppsModel{}
	output := strings.Join(rawOutput, "")
	json.Unmarshal([]byte(output), &apps)

	if len(apps.Apps) == 0 {
		fmt.Printf("App %s not found\n", appName)
		return
	}
	app := apps.Apps[0]

	messageQueue := logs.NewNoaaMessageQueue()

	bufferTime := 25 * time.Millisecond
	ticker := time.NewTicker(bufferTime)

	logChan := make(chan logs.Loggable)
	errChan := make(chan error)

	dopplerEndpoint, err := cliConnection.DopplerEndpoint()
	FreakOut(err)

	ssl, err := cliConnection.IsSSLDisabled()
	FreakOut(err)
	tlsConfig := net.NewTLSConfig([]tls.Certificate{}, ssl)

	noaaConsumer := consumer.New(dopplerEndpoint, tlsConfig, http.ProxyFromEnvironment)
	defer func() {
		noaaConsumer.Close()
		flushMessages(logChan, messageQueue)
	}()

	onConnect := func() {
		fmt.Printf("Tailing logs for app %s...\r\n\r\n", appName)
	}
	noaaConsumer.SetOnConnectCallback(onConnect)

	accessToken, err := cliConnection.AccessToken()
	FreakOut(err)

	c, e := noaaConsumer.TailingLogs(app.Guid, accessToken)

	go func() {
		for {
			select {
			case msg, ok := <-c:
				if !ok {
					ticker.Stop()
					flushMessages(logChan, messageQueue)
					close(logChan)
					close(errChan)
					return
				}

				messageQueue.PushMessage(msg)
			case err := <-e:
				if err != nil {
					errChan <- err

					ticker.Stop()
					close(logChan)
					close(errChan)
					return
				}
			}
		}
	}()

	go func() {
		for range ticker.C {
			flushMessages(logChan, messageQueue)
		}
	}()

	for {
		select {
		case msg := <-logChan:
			fmt.Printf("%s\r\n", logMessageOutput(msg, time.Local))
		case err, ok := <-errChan:
			if !ok {
				FreakOut(err)
			}
		}
	}
}

func initiateFirehoseConnection() (*consumer.Consumer, <-chan *events.Envelope) {
	localIP, _ := localip.LocalIP()
	firehoseConnection := consumer.New("ws://"+localIP+":49629", &tls.Config{InsecureSkipVerify: true}, nil)
	msgChan, _ := firehoseConnection.Firehose("uniqueId", "")
	return firehoseConnection, msgChan
}

			cnsmr       *consumer.Consumer
			testHandler *errorRespondingHandler
			tcURL       string
			refresher   *mockTokenRefresher
		)

		BeforeEach(func() {
			testHandler = &errorRespondingHandler{
				subHandler:       handlers.NewWebsocketHandler(make(chan []byte), 100*time.Millisecond, loggertesthelper.Logger()),
				responseStatuses: make(chan int, 10),
			}
			server := httptest.NewServer(testHandler)
			tcURL = "ws://" + server.Listener.Addr().String()

			refresher = newMockTokenRefresher()
			cnsmr = consumer.New(tcURL, nil, nil)

			cnsmr.RefreshTokenFrom(refresher)
		})

		Describe("TailingLogs", func() {
			It("loads a token if the provided token is empty", func() {
				cnsmr.TailingLogs("some-fake-app-guid", "")
				Eventually(refresher.RefreshAuthTokenCalled).Should(BeCalled())
			})

			It("loads a token if the provided token fails with 401", func() {
				testHandler.responseStatuses <- http.StatusUnauthorized
				cnsmr.TailingLogs("some-fake-app-guid", "")
				Eventually(refresher.RefreshAuthTokenCalled).Should(BeCalled())
			})

func (c *Client) Start() {
	var err error
	dopplerConnection := consumer.New(c.dopplerEndpoint, &tls.Config{InsecureSkipVerify: true}, nil)
	if c.options.Debug {
		dopplerConnection.SetDebugPrinter(ConsoleDebugPrinter{ui: c.ui})
	}
	filter := ""
	switch {
	case c.options.NoFilter:
		filter = ""
	case c.options.Filter != "":
		envelopeType, ok := events.Envelope_EventType_value[c.options.Filter]
		if !ok {
			c.ui.Warn("Unable to recognize filter %s", c.options.Filter)
			return
		}
		filter = strconv.Itoa(int(envelopeType))

	default:
		c.ui.Say("What type of firehose messages do you want to see?")
		filter, err = c.promptFilterType()
		if err != nil {
			c.ui.Warn(err.Error())
			return
		}
	}

	var errors <-chan error
	var output <-chan *events.Envelope
	if len(c.options.AppGUID) != 0 {
		c.ui.Say("Starting the nozzle for app %s", c.options.AppGUID)
		output, errors = dopplerConnection.StreamWithoutReconnect(c.options.AppGUID, c.authToken)
	} else {
		subscriptionID := c.options.SubscriptionID
		if len(subscriptionID) == 0 {
			subscriptionID = "FirehosePlugin"
		}
		c.ui.Say("Starting the nozzle")
		output, errors = dopplerConnection.FirehoseWithoutReconnect(subscriptionID, c.authToken)
	}

	done := make(chan struct{})
	go func() {
		defer close(done)
		for err := range errors {
			c.ui.Warn(err.Error())
			return
		}
	}()

	defer dopplerConnection.Close()

	c.ui.Say("Hit Ctrl+c to exit")

	for envelope := range output {
		if filter == "" || filter == strconv.Itoa((int)(envelope.GetEventType())) {
			c.ui.Say("%v \n", envelope)
		}
	}
	<-done
}