<commit_before>package broker

import (
	"bytes"
	"encoding/json"
	"fmt"
	"reflect"

	"github.com/jinzhu/gorm"
	"github.com/streadway/amqp"
	"labix.org/v2/mgo"
)

type Handler interface {
	HandleEvent(string, []byte) error
	ErrHandler
}

type ErrHandler interface {
	// bool is whether publishing the message to maintenance qeueue or not
	DefaultErrHandler(amqp.Delivery, error) bool
}

func (c *Consumer) withMetrics(handler *SubscriptionHandler, delivery amqp.Delivery) error {
	// if metrics is not enabled just invoke the function
	if c.Metrics == nil {
		return handler.HandleEvent(c.contextValue, delivery.Body)
	}

	c.Metrics.GetCounter("message").Inc(1)

	var err error

	c.Metrics.GetTimer(delivery.Type).Time(func() {
		err = handler.HandleEvent(c.contextValue, delivery.Body)
	})

	return err
}

func (c *Consumer) withCounter(counterName string, fn func()) {
	if c.Metrics == nil {
		fn()
	}

	c.Metrics.GetCounter(counterName).Inc(1)
	fn()
}

func (c *Consumer) Start() func(delivery amqp.Delivery) {
	c.Log.Info("Broker sarted to consume")
	return func(delivery amqp.Delivery) {<commit_msg>Go: add debug line \<commit_after>package broker

import (
	"bytes"
	"encoding/json"
	"fmt"
	"reflect"

	"github.com/jinzhu/gorm"
	"github.com/streadway/amqp"
	"labix.org/v2/mgo"
)

type Handler interface {
	HandleEvent(string, []byte) error
	ErrHandler
}

type ErrHandler interface {
	// bool is whether publishing the message to maintenance qeueue or not
	DefaultErrHandler(amqp.Delivery, error) bool
}

func (c *Consumer) withMetrics(handler *SubscriptionHandler, delivery amqp.Delivery) error {
	// if metrics is not enabled just invoke the function
	if c.Metrics == nil {
		return handler.HandleEvent(c.contextValue, delivery.Body)
	}

	c.Metrics.GetCounter("message").Inc(1)

	var err error

	c.Metrics.GetTimer(delivery.Type).Time(func() {
		err = handler.HandleEvent(c.contextValue, delivery.Body)
	})

	return err
}

func (c *Consumer) withCounter(counterName string, fn func()) {
	if c.Metrics == nil {
		fn()
	}

	c.Metrics.GetCounter(counterName).Inc(1)
	fn()
}

func (c *Consumer) Start() func(delivery amqp.Delivery) {
	c.Log.Info("Broker sarted to consume")
	return func(delivery amqp.Delivery) {
		c.Log.Debug("Handling event of %s", delivery.Type)

{'fraction_non_alphanumeric': 0.12825033054208904, 'fraction_numerical': 0.002644336712208021, 'mean_word_length': 4.550122249388753, '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': '5514871', 'n_tokens_mistral': 819, 'n_tokens_neox': 794, 'n_words': 195}