<commit_before>package main

import (
	"fmt"
	"flag"
	"encoding/json"
	"os"
	"regexp"
	"sync/atomic"
	"log"
	"path/filepath"
	"github.com/APTrust/bagman"
	"github.com/APTrust/bagman/fluctus/client"
	"launchpad.net/goamz/aws"
	"launchpad.net/goamz/s3"
	"github.com/bitly/go-nsq"
)

type Channels struct {
	FetchChannel     chan bagman.ProcessResult
	UnpackChannel    chan bagman.ProcessResult
	CleanUpChannel   chan bagman.ProcessResult
	ResultsChannel   chan bagman.ProcessResult
}


// Global vars.
var channels *Channels
var config bagman.Config
var jsonLog *log.Logger
var messageLog *log.Logger
var volume *bagman.Volume
var succeeded = int64(0)
var failed = int64(0)
var bytesInS3 = int64(0)
var bytesProcessed = int64(0)
var fluctusClient *client.Client

// bag_processor receives messages from nsqd describing
// items in the S3 receiving buckets. Each item/message
// follows this flow:
//
// 1. Fetch channel: fetches the file from S3.
// 2. Unpack channel: untars the bag files, parses and validates
//    the bag, reads tags, generates checksums and generic file
//    UUIDs.
// 3. **** TODO **** Copy files to S3 permanent storage.
// 4. Results channel: tells the queue whether processing
//    succeeded, and if not, whether the item should be requeued.
//    Also logs results to json and message logs.
// 5. Cleanup channel: cleans up the files after processing
//    completes.
//
// If a failure occurs anywhere in the first three steps,
// processing goes directly to the Results Channel, which
// records the error and the disposition (retry/give up).
//
// As long as the message from nsq contains valid JSON,
// steps 4 and 5 ALWAYS run.
//
// The bag processor has so many responsibilities because
// downloading, untarring and running checksums on
// multi-gigabyte files takes a lot of time. We want to
// avoid having separate processes repeatedly download and
// untar the files, so bag_processor performs all operations
// that require local access to the raw contents of the bags.
func main() {

	loadConfig()
	initVolume()
	initChannels()
	initGoRoutines(channels)


	nsqConfig := nsq.NewConfig()
	nsqConfig.Set("max_in_flight", 100)
	nsqConfig.Set("heartbeat_interval", "0m15s")

	// Try setting this to max allowed (5m) because nsqd thinks
	// client is gone when it spends a long time processing
	// a multi-GB file.
	nsqConfig.Set("read_timeout", "5m")

	consumer, err := nsq.NewConsumer(config.BagProcessorTopic, config.BagProcessorChannel, nsqConfig)
	if err != nil {
		messageLog.Fatalf(err.Error())
	}

	handler := &BagProcessor{}
	consumer.SetHandler(handler)
	consumer.ConnectToNSQLookupd(config.NsqLookupd)

	// This reader blocks until we get an interrupt, so our program does not exit.
	<-consumer.StopChan
}<commit_msg>Removed read_timeout setting - not helpful<commit_after>package main

import (
	"fmt"
	"flag"
	"encoding/json"
	"os"
	"regexp"
	"sync/atomic"
	"log"
	"path/filepath"
	"github.com/APTrust/bagman"
	"github.com/APTrust/bagman/fluctus/client"
	"launchpad.net/goamz/aws"
	"launchpad.net/goamz/s3"
	"github.com/bitly/go-nsq"
)

type Channels struct {
	FetchChannel     chan bagman.ProcessResult
	UnpackChannel    chan bagman.ProcessResult
	CleanUpChannel   chan bagman.ProcessResult
	ResultsChannel   chan bagman.ProcessResult
}


// Global vars.
var channels *Channels
var config bagman.Config
var jsonLog *log.Logger
var messageLog *log.Logger
var volume *bagman.Volume
var succeeded = int64(0)
var failed = int64(0)
var bytesInS3 = int64(0)
var bytesProcessed = int64(0)
var fluctusClient *client.Client

// bag_processor receives messages from nsqd describing
// items in the S3 receiving buckets. Each item/message
// follows this flow:
//
// 1. Fetch channel: fetches the file from S3.
// 2. Unpack channel: untars the bag files, parses and validates
//    the bag, reads tags, generates checksums and generic file
//    UUIDs.
// 3. **** TODO **** Copy files to S3 permanent storage.
// 4. Results channel: tells the queue whether processing
//    succeeded, and if not, whether the item should be requeued.
//    Also logs results to json and message logs.
// 5. Cleanup channel: cleans up the files after processing
//    completes.
//
// If a failure occurs anywhere in the first three steps,
// processing goes directly to the Results Channel, which
// records the error and the disposition (retry/give up).
//
// As long as the message from nsq contains valid JSON,
// steps 4 and 5 ALWAYS run.
//
// The bag processor has so many responsibilities because
// downloading, untarring and running checksums on
// multi-gigabyte files takes a lot of time. We want to
// avoid having separate processes repeatedly download and
// untar the files, so bag_processor performs all operations
// that require local access to the raw contents of the bags.
func main() {

	loadConfig()
	initVolume()
	initChannels()
	initGoRoutines(channels)


	nsqConfig := nsq.NewConfig()
	nsqConfig.Set("max_in_flight", 100)
	nsqConfig.Set("heartbeat_interval", "0m15s")
	consumer, err := nsq.NewConsumer(config.BagProcessorTopic, config.BagProcessorChannel, nsqConfig)
	if err != nil {
		messageLog.Fatalf(err.Error())
	}

	handler := &BagProcessor{}
	consumer.SetHandler(handler)
	consumer.ConnectToNSQLookupd(config.NsqLookupd)

	// This reader blocks until we get an interrupt, so our program does not exit.
	<-consumer.StopChan
}
{'fraction_non_alphanumeric': 0.0976380881532453, 'fraction_numerical': 0.01153059326762135, 'mean_word_length': 3.889090909090909, 'pattern_counts': {'":': 0, '<': 5, '<?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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3713492', 'n_tokens_mistral': 1750, 'n_tokens_neox': 1649, 'n_words': 597}