Code added with Makefile and dependencies

Author: dimchansky

.gitignore

@@ -1,12 +1,40 @@
 # Binaries for programs and plugins
 *.exe
-*.exe~
 *.dll
 *.so
 *.dylib
+*.o
+*.a
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.prof
 
 # Test binary, build with `go test -c`
 *.test
 
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
+
+# Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736
+.glide/
+
+# Gogland
+.idea/
+
+# Dependencies
+vendor/

Makefile

@@ -0,0 +1,55 @@
+PKGS ?= $(shell glide novendor)
+BENCH_FLAGS ?= -benchmem
+
+.PHONY: all
+all: lint test
+
+.PHONY: dependencies
+dependencies:
+	@echo "Installing Glide and locked dependencies..."
+	glide --version || go get -u -f github.com/Masterminds/glide
+	glide install
+	@echo "Installing goimports..."
+	go install ./vendor/golang.org/x/tools/cmd/goimports
+	@echo "Installing golint..."
+	go install ./vendor/golang.org/x/lint/golint
+	@echo "Installing gosimple..."
+	go install ./vendor/honnef.co/go/tools/cmd/gosimple
+	@echo "Installing unused..."
+	go install ./vendor/honnef.co/go/tools/cmd/unused
+	@echo "Installing staticcheck..."
+	go install ./vendor/honnef.co/go/tools/cmd/staticcheck
+
+.PHONY: lint
+lint:
+	@echo "Checking formatting..."
+	@gofiles=$$(go list -f {{.Dir}} $(PKGS) | grep -v mock) && [ -z "$$gofiles" ] || unformatted=$$(for d in $$gofiles; do goimports -l $$d/*.go; done) && [ -z "$$unformatted" ] || (echo >&2 "Go files must be formatted with goimports. Following files has problem:\n$$unformatted" && false)
+	@echo "Checking vet..."
+	@go vet $(PKG_FILES)
+	@echo "Checking simple..."
+	@gosimple $(PKG_FILES)
+	@echo "Checking unused..."
+	@unused $(PKG_FILES)
+	@echo "Checking staticcheck..."
+	@staticcheck $(PKG_FILES)
+	@echo "Checking lint..."
+	@$(foreach dir,$(PKGS),golint $(dir);)
+
+.PHONY: test
+test:
+	go test -timeout 20s -race -v $(PKGS)
+
+.PHONY: cover
+cover:
+	./cover.sh $(PKGS)
+
+.PHONY: bench
+BENCH ?= .
+bench:
+	$(foreach pkg,$(PKGS),go test -bench=$(BENCH) -run="^$$" $(BENCH_FLAGS) $(pkg);)
+
+.PHONY: fmt
+fmt:
+	@echo "Formatting files..."
+	@gofiles=$$(go list -f {{.Dir}} $(PKGS) | grep -v mock) && [ -z "$$gofiles" ] || for d in $$gofiles; do goimports -l -w $$d/*.go; done
+

broker/broker.go

@@ -0,0 +1,9 @@
+package broker
+
+import "github.com/monetha/go-distributed/task"
+
+// Broker is an interface for creating distributed tasks.
+type Broker interface {
+	// NewTask creates long-running distributed task.
+	NewTask(key string, fun task.Func) (*task.Task, error)
+}

broker/consulbroker/broker.go

@@ -0,0 +1,57 @@
+package consulbroker
+
+import (
+	"time"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/monetha/go-distributed/task"
+)
+
+const (
+	// LockWaitTime is how long we block for at a time to check if locker
+	// acquisition is possible. This affects the minimum time it takes to cancel
+	// a Lock acquisition.
+	LockWaitTime = 2 * time.Second
+)
+
+// Config is used to configure the creation of a broker
+type Config struct {
+	// Address is the address of the Consul server
+	Address string
+
+	// Scheme is the URI scheme for the Consul server
+	Scheme string
+
+	// Token is used to provide a per-request ACL token
+	Token string
+}
+
+// Broker can start long-running tasks (minutes to permanent) in Consul cluster.
+type Broker struct {
+	client *api.Client
+}
+
+// New creates a broker that can start long-running tasks in Consul cluster.
+func New(config *Config) (*Broker, error) {
+	if config == nil {
+		config = &Config{}
+	}
+
+	client, err := api.NewClient(&api.Config{
+		Address: config.Address,
+		Scheme:  config.Scheme,
+		Token:   config.Token,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return &Broker{client: client}, nil
+}
+
+// NewTask creates new long-running task in Consul cluster.
+// Task makes a best effort to ensure that exactly one instance of a task is executing in a cluster.
+// Task may be re-started when needed until it's been closed.
+func (b *Broker) NewTask(key string, fun task.Func) (*task.Task, error) {
+	return task.New(newConsulLocker(b.client, key, LockWaitTime), fun), nil
+}

broker/consulbroker/broker_test.go

@@ -0,0 +1,7 @@
+package consulbroker
+
+import "github.com/monetha/go-distributed/broker"
+
+var (
+	_ broker.Broker = &Broker{} // ensure Broker implements Broker interface
+)

broker/consulbroker/example_test.go

@@ -0,0 +1,70 @@
+package consulbroker_test
+
+import (
+	"context"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/monetha/go-distributed/broker/consulbroker"
+)
+
+func ExampleNew() {
+	defer log.Println("service stopped.")
+
+	b, err := consulbroker.New(&consulbroker.Config{Address: "127.0.0.1:8500"})
+	if err != nil {
+		log.Printf("New broker: %v", err)
+		return
+	}
+
+	t, err := b.NewTask("some/long/running/task", someLongRunningTask)
+	if err != nil {
+		log.Printf("New task: %v", err)
+		return
+	}
+	defer t.Close()
+
+	t2, err := b.NewTask("other/long/running/task", otherLongRunningTask)
+	if err != nil {
+		log.Printf("New task: %v", err)
+		return
+	}
+	defer t2.Close()
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGQUIT)
+	<-sigChan
+}
+
+func someLongRunningTask(ctx context.Context) error {
+	log.Println("someLongRunningTask: I'm alive!!")
+
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			log.Println("someLongRunningTask: Doing some interesting stuff..")
+		case <-ctx.Done():
+			log.Println("someLongRunningTask: Oh, no! I should stop my work.")
+			return ctx.Err()
+		}
+	}
+}
+
+func otherLongRunningTask(ctx context.Context) error {
+	log.Println("otherLongRunningTask: BANG BANG!!")
+
+	select {
+	case <-time.After(10 * time.Second):
+		log.Println("otherLongRunningTask: BOOO!!")
+		return nil
+	case <-ctx.Done():
+		log.Println("otherLongRunningTask: good bye")
+		return ctx.Err()
+	}
+}

broker/consulbroker/locker.go

@@ -0,0 +1,78 @@
+package consulbroker
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/monetha/go-distributed/locker"
+)
+
+// consulLocker wraps Consul distributed lock by implementing Locker interface.
+type consulLocker struct {
+	client       *api.Client
+	key          string
+	lockWaitTime time.Duration
+	lock         *api.Lock
+}
+
+// NewConsulLocker creates new consulLocker instance.
+func newConsulLocker(client *api.Client, key string, lockWaitTime time.Duration) *consulLocker {
+	return &consulLocker{
+		client:       client,
+		key:          key,
+		lockWaitTime: lockWaitTime,
+	}
+}
+
+// Key returns the name of locker.
+func (l *consulLocker) Key() string {
+	return l.key
+}
+
+// Lock attempts to acquire the locker and blocks while doing so.
+// Providing a non-nil stopCh can be used to abort the locker attempt.
+// Returns a channel that is closed if our locker is lost or an error.
+// This channel could be closed at any time due to session invalidation,
+// communication errors, operator intervention, etc. It is NOT safe to
+// assume that the locker is held until Unlock(), application must be able
+// to handle the locker being lost.
+func (l *consulLocker) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
+	if l.lock != nil {
+		return nil, locker.ErrLockHeld
+	}
+
+	lock, err := l.client.LockOpts(&api.LockOptions{
+		Key:          l.key,
+		LockWaitTime: l.lockWaitTime,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("locker: creating lock opts %s: %v", l.key, err)
+	}
+
+	lockCh, err := lock.Lock(stopCh)
+	if err != nil {
+		return nil, fmt.Errorf("locker: lock %s: %v", l.key, err)
+	}
+
+	if lockCh == nil {
+		return nil, locker.LockCancelled(l.key)
+	}
+
+	l.lock = lock
+
+	return lockCh, nil
+}
+
+// Unlock released the locker. It is an error to call this
+// if the locker is not currently held.
+func (l *consulLocker) Unlock() error {
+	if l.lock == nil {
+		return locker.ErrLockNotHeld
+	}
+	defer func() {
+		l.lock = nil
+	}()
+
+	return l.lock.Unlock()
+}