<commit_before>package main

import (
	"encoding/json"
	"github.com/streadway/amqp"
	"koding/databases/mongo"
	"koding/tools/amqputil"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"log"
	"strings"
	"time"
)

type Meta struct {
	CreatedAt  time.Time `bson:"createdAt"`
	ModifiedAt time.Time `bson:"modifiedAt"`
}

type Message struct {
	From       string
	RoutingKey string `bson:"routingKey"`
	Body       string
	Meta       Meta
}

type ConversationSlice struct {
	From time.Time
	To   time.Time
	Id   bson.ObjectId `bson:"_id"`
}

func main() {
	conn := amqputil.CreateConnection("persistence")

	startPersisting(conn)
}

func startPersisting(conn *amqp.Connection) {

	amqpChannel, err := conn.Channel()
	if err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeDeclare(
		"chat-hose", // exchange name
		"fanout",    // kind
		false,       // durable
		false,       // auto delete
		false,       // internal
		false,       // no wait
		nil,         // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeDeclare(
		"broker", // exchange name
		"topic",  // kind
		false,    // durable
		false,    // auto delete
		false,    // internal
		false,    // no wait
		nil,      // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeBind(
		"broker",    // destination
		"",          // key
		"chat-hose", // source
		false,       // no wait
		nil,         // arguments
	); err != nil {
		panic(err)
	}

	if _, err := amqpChannel.QueueDeclare(
		"persistence", // queue name
		false,         // durable
		true,          // autodelete
		true,          // exclusive
		false,         // no wait
		nil,           // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.QueueBind(
		"persistence", // queue name
		"",            // key
		"chat-hose",   // exchange name
		false,         // no wait
		nil,           // arguments
	); err != nil {
		panic(err)
	}

	deliveries, err := amqpChannel.Consume(
		"persistence", // queue name
		"",            // ctag
		false,         // no-ack
		false,         // exlusive
		false,         // no local
		false,         // no wait
		nil,           // arguments
	)
	if err != nil {
		panic(err)
	}

	errors := make(chan error)

	go persistMessages(
		amqpChannel,
		deliveries,
		mongo.GetCollection("jMessages"),
		mongo.GetCollection("jConversationSlices"),
		errors,
	)

	select {
	case errors <- err:
		log.Printf("Handled an error: %v", err)
	}
}

func persistMessages(
	amqpChannel *amqp.Channel,
	deliveries <-chan amqp.Delivery,
	messages *mgo.Collection,
	conversationSlices *mgo.Collection,
	done chan error,
) {

	for d := range deliveries {

		from := d.RoutingKey[strings.LastIndex(d.RoutingKey, ".")+1:]

		t := d.Timestamp

		message := Message{from, d.RoutingKey, string(d.Body), Meta{t, t}}

		info, err := messages.Upsert(bson.M{"_id": nil}, message)
		if err != nil {
			done <- err
			continue
		}

		slice := new(ConversationSlice)
		if err := conversationSlices.
			Find(bson.M{"routingKey": d.RoutingKey}).
			One(&slice); err != nil {
			done <- err
			continue
		}

		m := bson.M{"event": "NewMessage", "payload": bson.M{
			"source": slice.Id,
			"target": info.UpsertedId,
		}}

		neoMessage, err := json.Marshal(m)
		if err != nil {
			done <- err
			continue
		}

		amqpChannel.Publish(
			"neo4jFeederExchange", // exchange name
			"",                    // key
			false,                 // mandatory
			false,                 // immediate
			amqp.Publishing{
				Body: neoMessage,
			},
		)
	}
}<commit_msg>persistence: fix the error handling and keep looping forever<commit_after>package main

import (
	"encoding/json"
	"github.com/streadway/amqp"
	"koding/databases/mongo"
	"koding/tools/amqputil"
	"labix.org/v2/mgo"
	"labix.org/v2/mgo/bson"
	"log"
	"strings"
	"time"
)

type Meta struct {
	CreatedAt  time.Time `bson:"createdAt"`
	ModifiedAt time.Time `bson:"modifiedAt"`
}

type Message struct {
	From       string
	RoutingKey string `bson:"routingKey"`
	Body       string
	Meta       Meta
}

type ConversationSlice struct {
	From time.Time
	To   time.Time
	Id   bson.ObjectId `bson:"_id"`
}

func main() {
	conn := amqputil.CreateConnection("persistence")

	startPersisting(conn)
	// select {}
}

func startPersisting(conn *amqp.Connection) {

	amqpChannel, err := conn.Channel()
	if err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeDeclare(
		"chat-hose", // exchange name
		"fanout",    // kind
		false,       // durable
		false,       // auto delete
		false,       // internal
		false,       // no wait
		nil,         // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeDeclare(
		"broker", // exchange name
		"topic",  // kind
		false,    // durable
		false,    // auto delete
		false,    // internal
		false,    // no wait
		nil,      // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.ExchangeBind(
		"broker",    // destination
		"",          // key
		"chat-hose", // source
		false,       // no wait
		nil,         // arguments
	); err != nil {
		panic(err)
	}

	if _, err := amqpChannel.QueueDeclare(
		"persistence", // queue name
		false,         // durable
		true,          // autodelete
		true,          // exclusive
		false,         // no wait
		nil,           // arguments
	); err != nil {
		panic(err)
	}

	if err := amqpChannel.QueueBind(
		"persistence", // queue name
		"",            // key
		"chat-hose",   // exchange name
		false,         // no wait
		nil,           // arguments
	); err != nil {
		panic(err)
	}

	deliveries, err := amqpChannel.Consume(
		"persistence", // queue name
		"",            // ctag
		false,         // no-ack
		false,         // exlusive
		false,         // no local
		false,         // no wait
		nil,           // arguments
	)
	if err != nil {
		panic(err)
	}

	errors := make(chan error)

	go persistMessages(
		amqpChannel,
		deliveries,
		mongo.GetCollection("jMessages"),
		mongo.GetCollection("jConversationSlices"),
		errors,
	)
	for {
		select {
		case err = <-errors:
			log.Printf("Handled an error: %v", err)
		}
	}
}

func persistMessages(
	amqpChannel *amqp.Channel,
	deliveries <-chan amqp.Delivery,
	messages *mgo.Collection,
	conversationSlices *mgo.Collection,
	done chan error,
) {

	for d := range deliveries {

		from := d.RoutingKey[strings.LastIndex(d.RoutingKey, ".")+1:]

		t := d.Timestamp

		message := Message{from, d.RoutingKey, string(d.Body), Meta{t, t}}

		log.Println(message)

		info, err := messages.Upsert(bson.M{"_id": nil}, message)
		if err != nil {
			done <- err
			continue
		}

		slice := new(ConversationSlice)
		if err := conversationSlices.
			Find(bson.M{"routingKey": d.RoutingKey}).
			One(&slice); err != nil {
			done <- err
			continue
		}

		m := bson.M{"event": "NewMessage", "payload": bson.M{
			"source": slice.Id,
			"target": info.UpsertedId,
		}}

		neoMessage, err := json.Marshal(m)
		if err != nil {
			println("sfsfsf")
			done <- err
			continue
		}

		amqpChannel.Publish(
			"neo4jFeederExchange", // exchange name
			"",                    // key
			false,                 // mandatory
			false,                 // immediate
			amqp.Publishing{
				Body: neoMessage,
			},
		)

		done <- nil

	}
}
{'fraction_non_alphanumeric': 0.14438280166435508, 'fraction_numerical': 0.0011095700416088765, 'mean_word_length': 2.2105966162065895, 'pattern_counts': {'":': 12, '<': 14, '<?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': '4656094', 'n_tokens_mistral': 2794, 'n_tokens_neox': 2532, 'n_words': 580}