<commit_before>// Copyright 2016 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.

package logreader

import (
	"io"

	"github.com/juju/errors"
	"github.com/juju/loggo"
	"github.com/juju/version"
	"launchpad.net/tomb"

	"github.com/juju/juju/api/base"
	"github.com/juju/juju/api/common/stream"
	"github.com/juju/juju/apiserver/params"
	"github.com/juju/juju/logfwd"
)

// JSONReadCloser provides the functionality to send JSON-serialized
// values over a streaming connection.
type JSONReadCloser interface {
	io.Closer

	// ReadJSON decodes the next JSON value from the connection and
	// sets the value at the provided pointer to that newly decoded one.
	ReadJSON(interface{}) error
}

// LogRecordReader is a worker that provides log records it gets over
// a streaming connection. After getting each record, it converts them
// from params.LogRecord to logfwd.Record. These are then available
// through the reader's channel.
type LogRecordReader struct {
	tomb tomb.Tomb

	conn           JSONReadCloser
	out            chan logfwd.Record
	controllerUUID string
}

// OpenLogRecordReader opens a stream to the API's /log endpoint and
// returns a reader that wraps that stream.
//
// Note that the caller is responsible for stopping the reader, e.g. by
// passing it to worker.Stop().
func OpenLogRecordReader(conn base.StreamConnector, cfg params.LogStreamConfig, controllerUUID string) (*LogRecordReader, error) {
	wsStream, err := stream.Open(conn, "/logstream", &cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	lrr := newLogRecordReader(wsStream, controllerUUID)
	return lrr, nil
}

// newLogRecordReader starts a new reader and returns it. The provided
// connection is the one from which the reader will stream log records.
func newLogRecordReader(conn JSONReadCloser, controllerUUID string) *LogRecordReader {
	out := make(chan logfwd.Record)
	lrr := &LogRecordReader{
		conn:           conn,
		out:            out,
		controllerUUID: controllerUUID,
	}
	go func() {
		defer lrr.tomb.Done()
		defer close(lrr.out)
		defer lrr.conn.Close()
		lrr.tomb.Kill(lrr.loop())
	}()
	return lrr
}

// Channel returns a channel that can be used to receive log records.
func (lrr *LogRecordReader) Channel() <-chan logfwd.Record {
	return lrr.out
}

func (lrr *LogRecordReader) loop() error {
	for {
		var apiRecord params.LogStreamRecord
		err := lrr.conn.ReadJSON(&apiRecord)
		if err != nil {
			return err
		}

		record, err := api2record(apiRecord, lrr.controllerUUID)
		if err != nil {
			return errors.Trace(err)
		}

		select {
		case <-lrr.tomb.Dying():
			return tomb.ErrDying
		case lrr.out <- record:
		}
	}
}

// Kill implements Worker.Kill()
func (lrr *LogRecordReader) Kill() {
	lrr.tomb.Kill(nil)
}

// Wait implements Worker.Wait()
func (lrr *LogRecordReader) Wait() error {
	return lrr.tomb.Wait()<commit_msg>Clean up some error handling.<commit_after>// Copyright 2016 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.

package logreader

import (
	"io"

	"github.com/juju/errors"
	"github.com/juju/loggo"
	"github.com/juju/version"
	"launchpad.net/tomb"

	"github.com/juju/juju/api/base"
	"github.com/juju/juju/api/common/stream"
	"github.com/juju/juju/apiserver/params"
	"github.com/juju/juju/logfwd"
)

var logger = loggo.GetLogger("juju.api.logreader")

// JSONReadCloser provides the functionality to send JSON-serialized
// values over a streaming connection.
type JSONReadCloser interface {
	io.Closer

	// ReadJSON decodes the next JSON value from the connection and
	// sets the value at the provided pointer to that newly decoded one.
	ReadJSON(interface{}) error
}

// LogRecordReader is a worker that provides log records it gets over
// a streaming connection. After getting each record, it converts them
// from params.LogRecord to logfwd.Record. These are then available
// through the reader's channel.
type LogRecordReader struct {
	tomb tomb.Tomb

	conn           JSONReadCloser
	out            chan logfwd.Record
	controllerUUID string
}

// OpenLogRecordReader opens a stream to the API's /log endpoint and
// returns a reader that wraps that stream.
//
// Note that the caller is responsible for stopping the reader, e.g. by
// passing it to worker.Stop().
func OpenLogRecordReader(conn base.StreamConnector, cfg params.LogStreamConfig, controllerUUID string) (*LogRecordReader, error) {
	wsStream, err := stream.Open(conn, "/logstream", &cfg)
	if err != nil {
		return nil, errors.Trace(err)
	}
	lrr := newLogRecordReader(wsStream, controllerUUID)
	return lrr, nil
}

// newLogRecordReader starts a new reader and returns it. The provided
// connection is the one from which the reader will stream log records.
func newLogRecordReader(conn JSONReadCloser, controllerUUID string) *LogRecordReader {
	out := make(chan logfwd.Record)
	lrr := &LogRecordReader{
		conn:           conn,
		out:            out,
		controllerUUID: controllerUUID,
	}
	go func() {
		defer lrr.tomb.Done()
		defer close(lrr.out)
		defer func() {
			if err := lrr.conn.Close(); err != nil {
				logger.Errorf("while closing stream: %v", err)
			}
		}()
		lrr.tomb.Kill(lrr.loop())
	}()
	return lrr
}

// Channel returns a channel that can be used to receive log records.
func (lrr *LogRecordReader) Channel() <-chan logfwd.Record {
	return lrr.out
}

func (lrr *LogRecordReader) loop() error {
	for {
		var apiRecord params.LogStreamRecord
		err := lrr.conn.ReadJSON(&apiRecord)
		if err != nil {
			return errors.Trace(err)
		}

		record, err := api2record(apiRecord, lrr.controllerUUID)
		if err != nil {
			return errors.Trace(err)
		}

		select {
		case <-lrr.tomb.Dying():
			return tomb.ErrDying
		case lrr.out <- record:
		}
	}
}

// Kill implements Worker.Kill()
func (lrr *LogRecordReader) Kill() {
	lrr.tomb.Kill(nil)
}

// Wait implements Worker.Wait()
func (lrr *LogRecordReader) Wait() error {
	return lrr.tomb.Wait()
{'fraction_non_alphanumeric': 0.10796610169491526, 'fraction_numerical': 0.002033898305084746, 'mean_word_length': 3.6391509433962264, 'pattern_counts': {'":': 0, '<': 9, '<?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': '5185711', 'n_tokens_mistral': 2030, 'n_tokens_neox': 1916, 'n_words': 598}