<commit_before>package command

import (
	"flag"
	"fmt"
	"net"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/color"
	"github.com/funkygao/termui"
)

type TopBroker struct {
	Ui  cli.Ui
	Cmd string

	mu sync.Mutex

	zone, cluster, topic string
	drawMode             bool
	interval             time.Duration
	shortIp              bool

	offsets     map[string]int64 // host => offset sum
	lastOffsets map[string]int64
}

func (this *TopBroker) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("topbroker", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.StringVar(&this.topic, "t", "", "")
	cmdFlags.BoolVar(&this.drawMode, "d", false, "")
	cmdFlags.BoolVar(&this.shortIp, "shortip", false, "")
	cmdFlags.DurationVar(&this.interval, "i", time.Second*5, "refresh interval")

	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	this.offsets = make(map[string]int64)
	this.lastOffsets = make(map[string]int64)

	if this.interval.Seconds() < 1 {
		this.interval = time.Second
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	zkzone.ForSortedClusters(func(zkcluster *zk.ZkCluster) {
		if !patternMatched(zkcluster.Name(), this.cluster) {
			return
		}

		go this.clusterTopProducers(zkcluster)
	})

	if this.drawMode {
		this.drawDashboard()
		return
	}

	ticker := time.NewTicker(this.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			refreshScreen()
			this.showAndResetCounters(true)

		}
	}

	return
}

func (this *TopBroker) drawDashboard() {
	termui.Init()
	width := termui.TermWidth()
	height := termui.TermHeight()
	termui.Close()
	maxWidth := width - 23
	var maxQps float64
	for {
		refreshScreen()
		datas := this.showAndResetCounters(false)
		maxQps = .0
		for _, data := range datas {
			if data.qps > maxQps {
				maxQps = data.qps
			}
		}

		if maxQps < 1 {
			// draw empty lines
			for _, data := range datas {
				this.Ui.Output(fmt.Sprintf("%20s", data.host))
			}
			continue
		}

		for idx, data := range datas {
			if idx >= height-1 {
				break
			}

			if data.qps < 0 {
				data.qps = -data.qps // FIXME
			}

			w := int(data.qps*100/maxQps) * maxWidth / 100
			qps := fmt.Sprintf("%.1f", data.qps)
			bar := ""
			barColorLen := 0
			for i := 0; i < w-len(qps); i++ {
				bar += color.Green("|")
				barColorLen += 9 // color.Green will add extra 9 chars
			}
			bar += qps // TODO qps at the end, space in the middle
			for i := len(bar) - barColorLen; i < maxWidth; i++ {
				bar += " "
			}

			this.Ui.Output(fmt.Sprintf("%20s [%-s]", data.host, bar))
		}

		time.Sleep(this.interval)
	}
<commit_msg>qps will be aligned at the right, space in the middle<commit_after>package command

import (
	"flag"
	"fmt"
	"net"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/gocli"
	"github.com/funkygao/golib/color"
	"github.com/funkygao/termui"
)

type TopBroker struct {
	Ui  cli.Ui
	Cmd string

	mu sync.Mutex

	zone, cluster, topic string
	drawMode             bool
	interval             time.Duration
	shortIp              bool

	offsets     map[string]int64 // host => offset sum
	lastOffsets map[string]int64
}

func (this *TopBroker) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("topbroker", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.StringVar(&this.topic, "t", "", "")
	cmdFlags.BoolVar(&this.drawMode, "d", false, "")
	cmdFlags.BoolVar(&this.shortIp, "shortip", false, "")
	cmdFlags.DurationVar(&this.interval, "i", time.Second*5, "refresh interval")

	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	this.offsets = make(map[string]int64)
	this.lastOffsets = make(map[string]int64)

	if this.interval.Seconds() < 1 {
		this.interval = time.Second
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	zkzone.ForSortedClusters(func(zkcluster *zk.ZkCluster) {
		if !patternMatched(zkcluster.Name(), this.cluster) {
			return
		}

		go this.clusterTopProducers(zkcluster)
	})

	if this.drawMode {
		this.drawDashboard()
		return
	}

	ticker := time.NewTicker(this.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			refreshScreen()
			this.showAndResetCounters(true)

		}
	}

	return
}

func (this *TopBroker) drawDashboard() {
	termui.Init()
	width := termui.TermWidth()
	height := termui.TermHeight()
	termui.Close()
	maxWidth := width - 23
	var maxQps float64
	for {
		refreshScreen()
		datas := this.showAndResetCounters(false)
		maxQps = .0
		for _, data := range datas {
			if data.qps > maxQps {
				maxQps = data.qps
			}
		}

		if maxQps < 1 {
			// draw empty lines
			for _, data := range datas {
				this.Ui.Output(fmt.Sprintf("%20s", data.host))
			}

			continue
		}

		for idx, data := range datas {
			if idx >= height-1 {
				break
			}

			if data.qps < 0 {
				data.qps = -data.qps // FIXME
			}

			w := int(data.qps*100/maxQps) * maxWidth / 100
			qps := fmt.Sprintf("%.1f", data.qps)
			bar := ""
			barColorLen := 0
			for i := 0; i < w-len(qps); i++ {
				bar += color.Green("|")
				barColorLen += 9 // color.Green will add extra 9 chars
			}
			for i := len(bar) - barColorLen; i < maxWidth-len(qps); i++ {
				bar += " "
			}
			bar += qps

			this.Ui.Output(fmt.Sprintf("%20s [%s]", data.host, bar))
		}

		time.Sleep(this.interval)
	}

{'fraction_non_alphanumeric': 0.15808506974341313, 'fraction_numerical': 0.01171000516617875, 'mean_word_length': 3.1724137931034484, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 9, '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': '635289', 'n_tokens_mistral': 2502, 'n_tokens_neox': 2231, 'n_words': 449}