<commit_before>package main

import (
	"bufio"
	"fmt"
	"log"
	"os"

	"github.com/andrew-d/go-termutil"
	"github.com/codegangsta/cli"
	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
		panic(fmt.Sprintf("%s: %s", msg, err))
	}
}

func main() {
	app := cli.NewApp()
	app.Name = "pipecat"
	app.Usage = "Connect unix pipes and message queues"
	app.Flags = []cli.Flag{
		cli.StringFlag{
			Name:   "amqpUri",
			Value:  "amqp://guest:guest@localhost:5672/",
			Usage:  "AMQP URI",
			EnvVar: "AMQP_URI",
		},
	}

	app.Action = func(c *cli.Context) {
		list := c.Args().First()
		if list == "" {
			fmt.Println("Please provide name of the queue")
			os.Exit(1)
		}

		conn, err := amqp.Dial(c.String("amqpUri"))
		failOnError(err, "Failed to connect to AMPQ broker")
		defer conn.Close()

		channel, err := conn.Channel()
		failOnError(err, "Failed to open a channel")
		defer channel.Close()

		q, err := channel.QueueDeclare(
			list,  // name
			true,  // durable
			false, // delete when unused
			false, // exclusive
			false, // no-wait
			nil,   // arguments
		)
		failOnError(err, "Failed to declare a queue")

		unackedMessages := make(map[string]amqp.Delivery)
		if termutil.Isatty(os.Stdin.Fd()) {
			msgs, err := channel.Consume(
				q.Name, // queue
				"",     // consumer
				false,  // auto-ack
				false,  // exclusive
				false,  // no-local
				false,  // no-wait
				nil,    // args
			)
			failOnError(err, "Failed to register a consumer")
			for msg := range msgs {
				line := fmt.Sprintf("%s", msg.Body)
				unackedMessages[line] = msg
				fmt.Println(line)
				fmt.Fprintln(os.Stderr, fmt.Sprintf("%d", len(unackedMessages)))
			}<commit_msg>Added mutex<commit_after>package main

import (
	"bufio"
	"fmt"
	"log"
	"os"
	"sync"

	"github.com/andrew-d/go-termutil"
	"github.com/codegangsta/cli"
	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
		panic(fmt.Sprintf("%s: %s", msg, err))
	}
}

func main() {
	app := cli.NewApp()
	app.Name = "pipecat"
	app.Usage = "Connect unix pipes and message queues"
	app.Flags = []cli.Flag{
		cli.StringFlag{
			Name:   "amqpUri",
			Value:  "amqp://guest:guest@localhost:5672/",
			Usage:  "AMQP URI",
			EnvVar: "AMQP_URI",
		},
	}

	app.Action = func(c *cli.Context) {
		list := c.Args().First()
		if list == "" {
			fmt.Println("Please provide name of the queue")
			os.Exit(1)
		}

		conn, err := amqp.Dial(c.String("amqpUri"))
		failOnError(err, "Failed to connect to AMPQ broker")
		defer conn.Close()

		channel, err := conn.Channel()
		failOnError(err, "Failed to open a channel")
		defer channel.Close()

		q, err := channel.QueueDeclare(
			list,  // name
			true,  // durable
			false, // delete when unused
			false, // exclusive
			false, // no-wait
			nil,   // arguments
		)
		failOnError(err, "Failed to declare a queue")

		var mutex sync.Mutex
		unackedMessages := make([]amqp.Delivery, 100)
		if termutil.Isatty(os.Stdin.Fd()) {
			msgs, err := channel.Consume(
				q.Name, // queue
				"",     // consumer
				false,  // auto-ack
				false,  // exclusive
				false,  // no-local
				false,  // no-wait
				nil,    // args
			)
			failOnError(err, "Failed to register a consumer")

			go func() {
				for msg := range msgs {
					mutex.Lock()
					unackedMessages = append(unackedMessages, msg)
					mutex.Unlock()

					line := fmt.Sprintf("%s", msg.Body)
					fmt.Println(line)
					fmt.Fprintln(os.Stderr, fmt.Sprintf("%d", len(unackedMessages)))
				}
			}()
{'fraction_non_alphanumeric': 0.16657366071428573, 'fraction_numerical': 0.003627232142857143, 'mean_word_length': 2.821961620469083, 'pattern_counts': {'":': 0, '<': 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': '1136573', 'n_tokens_mistral': 1570, 'n_tokens_neox': 1353, 'n_words': 311}