git/githistory: introduce subpackage 'log'

Author: ttaylorr

git/githistory/log/log.go

@@ -0,0 +1,193 @@
+package log
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"strings"
+	"sync"
+
+	"github.com/git-lfs/git-lfs/tools"
+	"github.com/olekukonko/ts"
+)
+
+// Logger logs a series of tasks to an io.Writer, processing each task in order
+// until completion .
+type Logger struct {
+	// sink is the writer to write to.
+	sink io.Writer
+
+	// widthFn is a function that returns the width of the terminal that
+	// this logger is running within.
+	widthFn func() int
+
+	// queue is the incoming, unbuffered queue of tasks to enqueue.
+	queue chan Task
+	// tasks is the set of tasks to process.
+	tasks chan Task
+	// wg is a WaitGroup that is incremented when new tasks are enqueued,
+	// and decremented when tasks finish.
+	wg *sync.WaitGroup
+}
+
+// NewLogger retuns a new *Logger instance that logs to "sink" and uses the
+// current terminal width as the width of the line.
+func NewLogger(sink io.Writer) *Logger {
+	if sink == nil {
+		sink = ioutil.Discard
+	}
+
+	l := &Logger{
+		sink: sink,
+		widthFn: func() int {
+			size, err := ts.GetSize()
+			if err != nil {
+				return 80
+			}
+			return size.Col()
+		},
+		queue: make(chan Task),
+		tasks: make(chan Task),
+		wg:    new(sync.WaitGroup),
+	}
+
+	go l.consume()
+
+	return l
+}
+
+// Close closes the queue and does not allow new Tasks to be `enqueue()`'d. It
+// waits until the currently running Task has completed.
+func (l *Logger) Close() {
+	close(l.queue)
+
+	l.wg.Wait()
+}
+
+// Waitier creates and enqueues a new *WaitingTask.
+func (l *Logger) Waiter(msg string) *WaitingTask {
+	t := NewWaitingTask(msg)
+	l.enqueue(t)
+
+	return t
+}
+
+// Percentage creates and enqueues a new *PercentageTask.
+func (l *Logger) Percentage(msg string, total uint64) *PercentageTask {
+	t := NewPercentageTask(msg, total)
+	l.enqueue(t)
+
+	return t
+}
+
+// enqueue enqueues the given Tasks "ts".
+func (l *Logger) enqueue(ts ...Task) {
+	if l == nil {
+		for _, t := range ts {
+			go func() {
+				for range <-t.Updates() {
+					// Discard all updates.
+				}
+			}()
+		}
+		return
+	}
+
+	l.wg.Add(len(ts))
+	for _, t := range ts {
+		l.queue <- t
+	}
+}
+
+// consume creates a pseudo-infinte buffer between the incoming set of tasks and
+// the queue of tasks to work on.
+func (l *Logger) consume() {
+	go func() {
+		// Process the single next task in sequence until completion,
+		// then consume the next task.
+		for task := range l.tasks {
+			l.logTask(task)
+		}
+	}()
+
+	pending := make([]Task, 0)
+
+L:
+	for {
+		// If there is a pending task, "peek" it off of the set of
+		// pending tasks.
+		var next Task
+		if len(pending) > 0 {
+			next = pending[0]
+		}
+
+		if next == nil {
+			// If there was no pending task, wait for either a)
+			// l.queue to close, or b) a new task to be submitted.
+			task, ok := <-l.queue
+			if !ok {
+				// If the queue is closed, no more new tasks may
+				// be added.
+				break L
+			}
+
+			// Otherwise, add a new task to the set of tasks to
+			// process immediately, since there is no current
+			// buffer.
+			l.tasks <- task
+		} else {
+			// If there is a pending task, wait for either a) a
+			// write to process the task to become non-blocking, or
+			// b) a new task to enter the queue.
+			select {
+			case task, ok := <-l.queue:
+				if !ok {
+					// If the queue is closed, no more tasks
+					// may be added.
+					break L
+				}
+				// Otherwise, add the next task to the set of
+				// pending, active tasks.
+				pending = append(pending, task)
+			case l.tasks <- next:
+				// Or "pop" the peeked task off of the pending
+				// set.
+				pending = pending[1:]
+			}
+		}
+	}
+
+	close(l.tasks)
+}
+
+// logTask logs the set of updates from a given task to the sink, then logs a
+// "done" message, and then marks the task as done.
+func (l *Logger) logTask(task Task) {
+	defer l.wg.Done()
+
+	var last string
+	for last = range task.Updates() {
+		l.logLine(last)
+	}
+
+	l.log(fmt.Sprintf("%s, done\n", last))
+}
+
+// logLine writes a complete line and moves the cursor to the beginning of the
+// line.
+//
+// It returns the number of bytes "n" written to the sink and the error "err",
+// if one was encountered.
+func (l *Logger) logLine(str string) (n int, err error) {
+	padding := strings.Repeat(" ", tools.MaxInt(0, l.widthFn()-len(str)))
+
+	return l.log(str + padding + "\r")
+}
+
+// log writes a string verbatim to the sink.
+//
+// It returns the number of bytes "n" written to the sink and the error "err",
+// if one was encountered.
+func (l *Logger) log(str string) (n int, err error) {
+	return fmt.Fprint(l.sink, str)
+}

git/githistory/log/log_test.go

@@ -0,0 +1,62 @@
+package log
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type ChanTask chan string
+
+func (e ChanTask) Updates() <-chan string { return e }
+
+func TestLoggerLogsTasks(t *testing.T) {
+	var buf bytes.Buffer
+
+	task := make(chan string)
+	go func() {
+		task <- "first"
+		task <- "second"
+		close(task)
+	}()
+
+	l := NewLogger(&buf)
+	l.widthFn = func() int { return 0 }
+	l.enqueue(ChanTask(task))
+	l.Close()
+
+	assert.Equal(t, "first\rsecond\rsecond, done\n", buf.String())
+}
+
+func TestLoggerLogsMultipleTasksInOrder(t *testing.T) {
+	var buf bytes.Buffer
+
+	t1 := make(chan string)
+	go func() {
+		t1 <- "first"
+		t1 <- "second"
+		close(t1)
+	}()
+	t2 := make(chan string)
+	go func() {
+		t2 <- "third"
+		t2 <- "fourth"
+		close(t2)
+	}()
+
+	l := NewLogger(&buf)
+	l.widthFn = func() int { return 0 }
+	l.enqueue(ChanTask(t1), ChanTask(t2))
+	l.Close()
+
+	assert.Equal(t, strings.Join([]string{
+		"first\r",
+		"second\r",
+		"second, done\n",
+		"third\r",
+		"fourth\r",
+		"fourth, done\n",
+	}, ""), buf.String())
+}

git/githistory/log/percentage_task.go

@@ -0,0 +1,61 @@
+package log
+
+import (
+	"fmt"
+	"sync/atomic"
+)
+
+// PercentageTask is a task that is performed against a known number of
+// elements.
+type PercentageTask struct {
+	// msg is the task message.
+	msg string
+	// n is the number of elements whose work has been completed. It is
+	// managed sync/atomic.
+	n uint64
+	// total is the total number of elements to execute work upon.
+	total uint64
+	// ch is a channel which is written to when the task state changes and
+	// is closed when the task is completed.
+	ch chan string
+}
+
+func NewPercentageTask(msg string, total uint64) *PercentageTask {
+	p := &PercentageTask{
+		msg:   msg,
+		total: total,
+		ch:    make(chan string, 1),
+	}
+	p.Count(0)
+
+	return p
+}
+
+// Count indicates that work has been completed against "n" number of elements,
+// marking the task as complete if the total "n" given to all invocations of
+// this method is equal to total.
+//
+// Count returns the new total number of (atomically managed) elements that have
+// been completed.
+func (c *PercentageTask) Count(n uint64) (new uint64) {
+	new = atomic.AddUint64(&c.n, n)
+
+	percentage := 100 * float64(new) / float64(c.total)
+	msg := fmt.Sprintf("%s: %3.f%% (%d/%d)",
+		c.msg, percentage, new, c.total)
+
+	c.ch <- msg
+
+	if new >= c.total {
+		close(c.ch)
+	}
+
+	return new
+}
+
+// Updates implements Task.Updates and returns a channel which is written to
+// when the state of this task changes, and closed when the task is completed.
+// has been completed.
+func (c *PercentageTask) Updates() <-chan string {
+	return c.ch
+}

git/githistory/log/percentage_task_test.go

@@ -0,0 +1,39 @@
+package log
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPercentageTaskCalculuatesPercentages(t *testing.T) {
+	task := NewPercentageTask("example", 10)
+
+	assert.Equal(t, "example:   0% (0/10)", <-task.Updates())
+
+	n := task.Count(3)
+	assert.EqualValues(t, 3, n)
+
+	assert.Equal(t, "example:  30% (3/10)", <-task.Updates())
+}
+
+func TestPercentageTaskCallsDoneWhenComplete(t *testing.T) {
+	task := NewPercentageTask("example", 10)
+
+	select {
+	case v, ok := <-task.Updates():
+		if ok {
+			assert.Equal(t, "example:   0% (0/10)", v)
+		} else {
+			t.Fatal("expected channel to be open")
+		}
+	default:
+	}
+
+	assert.EqualValues(t, 10, task.Count(10))
+	assert.Equal(t, "example: 100% (10/10)", <-task.Updates())
+
+	if _, ok := <-task.Updates(); ok {
+		t.Fatalf("expected channel to be closed")
+	}
+}

git/githistory/log/task.go

@@ -0,0 +1,9 @@
+package log
+
+// Task is an interface which encapsulates an activity which can be logged.
+type Task interface {
+	// Updates returns a channel which is written to with the current state
+	// of the Task when an update is present. It is closed when the task is
+	// complete.
+	Updates() <-chan string
+}

git/githistory/log/waiting_task.go

@@ -0,0 +1,29 @@
+package log
+
+import "fmt"
+
+// WaitingTask represents a task for which the total number of items to do work
+// is on is unknown.
+type WaitingTask struct {
+	// ch is used to transmit task updates.
+	ch chan string
+}
+
+// NewWaitingTask returns a new *WaitingTask.
+func NewWaitingTask(msg string) *WaitingTask {
+	ch := make(chan string, 1)
+	ch <- fmt.Sprintf("%s: ...", msg)
+
+	return &WaitingTask{ch: ch}
+}
+
+// Complete marks the task as completed.
+func (w *WaitingTask) Complete() {
+	close(w.ch)
+}
+
+// Done implements Task.Done and returns a channel which is closed when
+// Complete() is called.
+func (w *WaitingTask) Updates() <-chan string {
+	return w.ch
+}