<commit_before>package app

import (
	"context"
	"fmt"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"runtime"

	"time"

	"cloud.google.com/go/logging"
	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/messages"
	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/version"
	"github.com/cloudfoundry/lager"
)

func Run(ctx context.Context, a *App) {
	ctx, cancel := context.WithCancel(ctx)

	if a.c.DebugNozzle {
		defer handleFatalError(a, cancel)

		go func() {
			a.logger.Info("pprof", lager.Data{
				"http.ListenAndServe": http.ListenAndServe("localhost:6060", nil),
			})
		}()
	}

	producer := a.newProducer()
	consumer, err := a.newConsumer(ctx)
	if err != nil {
		a.logger.Fatal("construction", err)
	}

	errs, fhErrs := consumer.Start(producer)
	go func() {
		select {
		case err := <-errs:
			a.logger.Error("nozzle", err)
		case err := <-fhErrs:
			a.logger.Error("firehose", err)
		}
	}()

	blockTillInterrupt()

	a.logger.Info("app", lager.Data{"cleanup": "exit recieved, attempting to flush buffers"})
	if err := consumer.Stop(); err != nil {
		a.logger.Error("nozzle.stop", err)
	}
	cancel()

	t := time.NewTimer(5 * time.Second)
	for {
		select {
		case <-time.Tick(500 * time.Millisecond):
			if a.bufferEmpty() {
				a.logger.Info("app", lager.Data{"cleanup": "The metrics buffer was successfully flushed before shutdown"})
				return
			}
		case <-t.C:
			a.logger.Info("app", lager.Data{"cleanup": "The metrics buffer could not be flushed before shutdown"})
			return
		}
	}
}

func blockTillInterrupt() {
	c := make(chan os.Signal, 1)<commit_msg>app: don't stop draining errors<commit_after>package app

import (
	"context"
	"fmt"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"runtime"

	"time"

	"cloud.google.com/go/logging"
	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/messages"
	"github.com/cloudfoundry-community/stackdriver-tools/src/stackdriver-nozzle/version"
	"github.com/cloudfoundry/lager"
)

func Run(ctx context.Context, a *App) {
	ctx, cancel := context.WithCancel(ctx)

	if a.c.DebugNozzle {
		defer handleFatalError(a, cancel)

		go func() {
			a.logger.Info("pprof", lager.Data{
				"http.ListenAndServe": http.ListenAndServe("localhost:6060", nil),
			})
		}()
	}

	producer := a.newProducer()
	consumer, err := a.newConsumer(ctx)
	if err != nil {
		a.logger.Fatal("construction", err)
	}

	errs, fhErrs := consumer.Start(producer)
	go func() {
		for {
			select {
			case err := <-errs:
				a.logger.Error("nozzle", err)
			case err := <-fhErrs:
				a.logger.Error("firehose", err)
			}
		}
	}()

	blockTillInterrupt()

	a.logger.Info("app", lager.Data{"cleanup": "exit recieved, attempting to flush buffers"})
	if err := consumer.Stop(); err != nil {
		a.logger.Error("nozzle.stop", err)
	}
	cancel()

	t := time.NewTimer(5 * time.Second)
	for {
		select {
		case <-time.Tick(500 * time.Millisecond):
			if a.bufferEmpty() {
				a.logger.Info("app", lager.Data{"cleanup": "The metrics buffer was successfully flushed before shutdown"})
				return
			}
		case <-t.C:
			a.logger.Info("app", lager.Data{"cleanup": "The metrics buffer could not be flushed before shutdown"})
			return
		}
	}
}

func blockTillInterrupt() {
	c := make(chan os.Signal, 1)
{'fraction_non_alphanumeric': 0.16605839416058393, 'fraction_numerical': 0.005474452554744526, 'mean_word_length': 3.8225806451612905, 'pattern_counts': {'":': 8, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2121316', 'n_tokens_mistral': 1380, 'n_tokens_neox': 1220, 'n_words': 242}