<commit_before>package krater

import (
	"fmt"
	"io"
	"io/ioutil"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"

	"github.com/Shopify/sarama"
)

// UnsafeWriter is an io.Writer that writes messages to Kafka, ignoring (or logging; see UnsafeWriter#SetLogger
// and LogOutput) any error responses sent by the brokers.
// Parallel calls to Write / ReadFrom are safe.
//
// It's recommended to initialize the sarama.Producer passed to UnsafeWriter with AckSuccesses = false.
//
// Close() must be called when the writer is no longer needed.
type UnsafeWriter struct {
	kp        Producer
	id        string
	topic     string
	closed    int32         // nonzero if the writer has started closing. Must be accessed atomically
	stopCh    chan struct{} // used to signal event loop close
	log       StdLogger
	pendingWg sync.WaitGroup // WaitGroup for pending messages
	closeMut  sync.Mutex     // mutex for Close and CloseAll
}

var unswIdGen = sequentialIntGen()

func NewUnsafeWriter(topic string, kp Producer) *UnsafeWriter {
	id := "aw-" + strconv.Itoa(unswIdGen())
	log := NewLogger(fmt.Sprintf("UnsafeWr %s -> %s", id, topic), nil)
	stopCh := make(chan struct{})

	uw := &UnsafeWriter{kp: kp, id: id, topic: topic, log: log, stopCh: stopCh}

	go func() {
		log.Println("Starting error listener")
		for {
			select {
			case <-stopCh:
				log.Println("Closing error listener")
				return
			case perr, ok := <-kp.Errors():
				if !ok {
					log.Println("Errors() channel closed?!")
					return
				}
				log.Println("Got error from Kafka:", perr)
			case _, ok := <-kp.Successes():
				if !ok {
					log.Println("Successes() channel closed?!")
					return
				}
			}
		}
	}()

	return uw<commit_msg>use withRecover in UnsafeWriter's event loop<commit_after>package krater

import (
	"fmt"
	"io"
	"io/ioutil"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"

	"github.com/Shopify/sarama"
)

// UnsafeWriter is an io.Writer that writes messages to Kafka, ignoring (or logging; see UnsafeWriter#SetLogger
// and LogOutput) any error responses sent by the brokers.
// Parallel calls to Write / ReadFrom are safe.
//
// It's recommended to initialize the sarama.Producer passed to UnsafeWriter with AckSuccesses = false.
//
// Close() must be called when the writer is no longer needed.
type UnsafeWriter struct {
	kp        Producer
	id        string
	topic     string
	closed    int32         // nonzero if the writer has started closing. Must be accessed atomically
	stopCh    chan struct{} // used to signal event loop close
	log       StdLogger
	pendingWg sync.WaitGroup // WaitGroup for pending messages
	closeMut  sync.Mutex     // mutex for Close and CloseAll
}

var unswIdGen = sequentialIntGen()

func NewUnsafeWriter(topic string, kp Producer) *UnsafeWriter {
	id := "aw-" + strconv.Itoa(unswIdGen())
	log := NewLogger(fmt.Sprintf("UnsafeWr %s -> %s", id, topic), nil)
	stopCh := make(chan struct{})

	uw := &UnsafeWriter{kp: kp, id: id, topic: topic, log: log, stopCh: stopCh}

	evtLoop := func() {
		log.Println("Starting error listener")
		for {
			select {
			case <-stopCh:
				log.Println("Closing error listener")
				return
			case perr, ok := <-kp.Errors():
				if !ok {
					log.Println("Errors() channel closed?!")
					return
				}
				log.Println("Got error from Kafka:", perr)
			case _, ok := <-kp.Successes():
				if !ok {
					log.Println("Successes() channel closed?!")
					return
				}
			}
		}
	}

	go withRecover(evtLoop)

	return uw
{'fraction_non_alphanumeric': 0.1155516941789748, 'fraction_numerical': 0.0011584129742253113, 'mean_word_length': 3.126642771804062, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '5784882', 'n_tokens_mistral': 1268, 'n_tokens_neox': 1127, 'n_words': 370}