<commit_before>package command

import (
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/config"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/color"
	"github.com/funkygao/golib/gofmt"
	"github.com/funkygao/metrics"
)

var (
	stats *peekStats
)

type peekStats struct {
	MsgPerSecond metrics.Meter
}

func newPeekStats() *peekStats {
	this := &peekStats{
		MsgPerSecond: metrics.NewMeter(),
	}

	metrics.Register("msg.per.second", this.MsgPerSecond)
	return this
}

func (this *peekStats) start() {
	metrics.Log(metrics.DefaultRegistry, time.Second*10,
		log.New(os.Stdout, "metrics: ", log.Lmicroseconds))
}

type Peek struct {
	Ui cli.Ui
}

func (this *Peek) Run(args []string) (exitCode int) {
	var (
		cluster     string
		zone        string
		topic       string
		partitionId int
		neat        bool
	)
	cmdFlags := flag.NewFlagSet("peek", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", "", "")
	cmdFlags.StringVar(&cluster, "c", "", "")
	cmdFlags.StringVar(&topic, "t", "", "")
	cmdFlags.IntVar(&partitionId, "p", 0, "")
	cmdFlags.BoolVar(&neat, "n", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).require("-z").invalid(args) {
		return 2
	}

	stats := newPeekStats()
	go stats.start()

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, config.ZonePath(zone)))
	msgChan := make(chan string, 2000) // msg aggerator channel
	if cluster == "" {
		zkzone.WithinClusters(func(name string, path string) {
			zkcluster := zkzone.NewCluster(name)
			this.consumeCluster(zkcluster, topic, partitionId, msgChan)
		})
	} else {
		zkcluster := zkzone.NewCluster(cluster)
		this.consumeCluster(zkcluster, topic, partitionId, msgChan)
	}

	var msg string
	for {
		select {
		case msg = <-msgChan:
			stats.MsgPerSecond.Mark(1)

			if !neat {
				this.Ui.Output(msg)
			}
		}
	}

	return
}

func (this *Peek) consumeCluster(zkcluster *zk.ZkCluster, topic string,
	partitionId int, msgChan chan string) {
	brokerList := zkcluster.BrokerList()
	kfk, err := sarama.NewClient(brokerList, sarama.NewConfig())
	if err != nil {
		this.Ui.Output(err.Error())
		return
	}
	//defer kfk.Close() // FIXME how to close it

	if topic == "" {
		// peek all topics
		topics, err := kfk.Topics()
		if err != nil {
			this.Ui.Output(err.Error())
			return
		}

		for _, t := range topics {
			go this.consumeTopic(kfk, t, int32(partitionId), msgChan)
		}
	} else {
		go this.consumeTopic(kfk, topic, int32(partitionId), msgChan)
	}
}

func (this *Peek) consumeTopic(kfk sarama.Client, topic string, partitionId int32, msgCh chan string) {
	consumer, err := sarama.NewConsumerFromClient(kfk)
	if err != nil {
		panic(err)
	}
	defer consumer.Close()<commit_msg>return early when no brokers found in a cluster<commit_after>package command

import (
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/config"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/color"
	"github.com/funkygao/golib/gofmt"
	"github.com/funkygao/metrics"
)

var (
	stats *peekStats
)

type peekStats struct {
	MsgPerSecond metrics.Meter
}

func newPeekStats() *peekStats {
	this := &peekStats{
		MsgPerSecond: metrics.NewMeter(),
	}

	metrics.Register("msg.per.second", this.MsgPerSecond)
	return this
}

func (this *peekStats) start() {
	metrics.Log(metrics.DefaultRegistry, time.Second*10,
		log.New(os.Stdout, "metrics: ", log.Lmicroseconds))
}

type Peek struct {
	Ui cli.Ui
}

func (this *Peek) Run(args []string) (exitCode int) {
	var (
		cluster     string
		zone        string
		topic       string
		partitionId int
		neat        bool
	)
	cmdFlags := flag.NewFlagSet("peek", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", "", "")
	cmdFlags.StringVar(&cluster, "c", "", "")
	cmdFlags.StringVar(&topic, "t", "", "")
	cmdFlags.IntVar(&partitionId, "p", 0, "")
	cmdFlags.BoolVar(&neat, "n", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).require("-z").invalid(args) {
		return 2
	}

	stats := newPeekStats()
	go stats.start()

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, config.ZonePath(zone)))
	msgChan := make(chan string, 2000) // msg aggerator channel
	if cluster == "" {
		zkzone.WithinClusters(func(name string, path string) {
			zkcluster := zkzone.NewCluster(name)
			this.consumeCluster(zkcluster, topic, partitionId, msgChan)
		})
	} else {
		zkcluster := zkzone.NewCluster(cluster)
		this.consumeCluster(zkcluster, topic, partitionId, msgChan)
	}

	var msg string
	for {
		select {
		case msg = <-msgChan:
			stats.MsgPerSecond.Mark(1)

			if !neat {
				this.Ui.Output(msg)
			}
		}
	}

	return
}

func (this *Peek) consumeCluster(zkcluster *zk.ZkCluster, topic string,
	partitionId int, msgChan chan string) {
	brokerList := zkcluster.BrokerList()
	if len(brokerList) == 0 {
		return
	}
	kfk, err := sarama.NewClient(brokerList, sarama.NewConfig())
	if err != nil {
		this.Ui.Output(err.Error())
		return
	}
	//defer kfk.Close() // FIXME how to close it

	if topic == "" {
		// peek all topics
		topics, err := kfk.Topics()
		if err != nil {
			this.Ui.Output(err.Error())
			return
		}

		for _, t := range topics {
			go this.consumeTopic(kfk, t, int32(partitionId), msgChan)
		}
	} else {
		go this.consumeTopic(kfk, topic, int32(partitionId), msgChan)
	}
}

func (this *Peek) consumeTopic(kfk sarama.Client, topic string, partitionId int32, msgCh chan string) {
	consumer, err := sarama.NewConsumerFromClient(kfk)
	if err != nil {
		panic(err)
	}
	defer consumer.Close()
{'fraction_non_alphanumeric': 0.15214001039681163, 'fraction_numerical': 0.005718246404435973, 'mean_word_length': 3.883248730964467, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4790821', 'n_tokens_mistral': 2378, 'n_tokens_neox': 2200, 'n_words': 455}