<commit_before>package logs

import (
	"net/http"
	"strconv"
	"time"

	"github.com/gorilla/websocket"
	"github.com/pivotal-golang/lager"

	"github.com/concourse/atc/auth"
	"github.com/concourse/atc/config"
	"github.com/concourse/atc/db"
	"github.com/concourse/atc/logfanout"
)

var upgrader = websocket.Upgrader{
	CheckOrigin: func(*http.Request) bool {
		return true
	},
}

func NewHandler(
	logger lager.Logger,
	validator auth.Validator,
	jobs config.Jobs,
	tracker *logfanout.Tracker,
	db db.DB,
) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		buildIDStr := r.FormValue(":build_id")

		log := logger.Session("logs-out", lager.Data{
			"build_id": buildIDStr,
		})

		buildID, err := strconv.Atoi(buildIDStr)
		if err != nil {
			log.Error("invalid-build-id", err)
			return
		}

		authenticated := validator.IsAuthenticated(r)

		if !authenticated {
			build, err := db.GetBuild(buildID)
			if err != nil {
				log.Error("invalid-build-id", err)
				w.WriteHeader(http.StatusNotFound)
				return
			}

			job, found := jobs.Lookup(build.JobName)
			if !found || !job.Public {
				w.WriteHeader(http.StatusNotFound)
				return
			}
		}

		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			log.Error("upgrade-failed", err)
			return
		}

		defer conn.Close()

		logFanout := tracker.Register(buildID, conn)
		defer tracker.Unregister(buildID, conn)

		var sink logfanout.Sink
		if authenticated {
			sink = logfanout.NewRawSink(conn)
		} else {
			sink = logfanout.NewCensoredSink(conn)
		}

		err = logFanout.Attach(sink)<commit_msg>fan out build logs without blocking<commit_after>package logs

import (
	"net/http"
	"strconv"
	"time"

	"github.com/gorilla/websocket"
	"github.com/pivotal-golang/lager"

	"github.com/concourse/atc/auth"
	"github.com/concourse/atc/config"
	"github.com/concourse/atc/db"
	"github.com/concourse/atc/logfanout"
)

var upgrader = websocket.Upgrader{
	CheckOrigin: func(*http.Request) bool {
		return true
	},
}

func NewHandler(
	logger lager.Logger,
	validator auth.Validator,
	jobs config.Jobs,
	tracker *logfanout.Tracker,
	db db.DB,
) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		buildIDStr := r.FormValue(":build_id")

		log := logger.Session("logs-out", lager.Data{
			"build_id": buildIDStr,
		})

		buildID, err := strconv.Atoi(buildIDStr)
		if err != nil {
			log.Error("invalid-build-id", err)
			return
		}

		authenticated := validator.IsAuthenticated(r)

		if !authenticated {
			build, err := db.GetBuild(buildID)
			if err != nil {
				log.Error("invalid-build-id", err)
				w.WriteHeader(http.StatusNotFound)
				return
			}

			job, found := jobs.Lookup(build.JobName)
			if !found || !job.Public {
				w.WriteHeader(http.StatusNotFound)
				return
			}
		}

		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			log.Error("upgrade-failed", err)
			return
		}

		defer conn.Close()

		logFanout := tracker.Register(buildID, conn)
		defer tracker.Unregister(buildID, conn)

		var sink logfanout.Sink
		if authenticated {
			sink = logfanout.NewRawSink(conn)
		} else {
			sink = logfanout.NewCensoredSink(conn)
		}

		sink = logfanout.NewAsyncSink(sink, 1000)

		err = logFanout.Attach(sink)
{'fraction_non_alphanumeric': 0.13819636811326563, 'fraction_numerical': 0.0012311480455524776, 'mean_word_length': 3.642857142857143, 'pattern_counts': {'":': 4, '<': 3, '<?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': '2797091', 'n_tokens_mistral': 1400, 'n_tokens_neox': 1225, 'n_words': 221}