Locker implementations are now public

Author: dimchansky

broker/consulbroker/broker.go

@@ -53,5 +53,5 @@ func New(config *Config) (*Broker, error) {
 // 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
+	return task.New(NewLocker(b.client, key, LockWaitTime), fun), nil
 }

broker/consulbroker/locker.go

@@ -8,25 +8,25 @@ import (
 	"github.com/monetha/go-distributed/locker"
 )
 
-// consulLocker wraps Consul distributed lock by implementing Locker interface.
-type consulLocker struct {
+// Locker wraps Consul distributed lock by implementing Locker interface.
+type Locker 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{
+// NewLocker creates new Locker instance.
+func NewLocker(client *api.Client, key string, lockWaitTime time.Duration) *Locker {
+	return &Locker{
 		client:       client,
 		key:          key,
 		lockWaitTime: lockWaitTime,
 	}
 }
 
 // Key returns the name of locker.
-func (l *consulLocker) Key() string {
+func (l *Locker) Key() string {
 	return l.key
 }
 
@@ -37,7 +37,7 @@ func (l *consulLocker) Key() string {
 // 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) {
+func (l *Locker) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
 	if l.lock != nil {
 		return nil, locker.ErrLockHeld
 	}
@@ -66,7 +66,7 @@ func (l *consulLocker) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
 
 // Unlock released the locker. It is an error to call this
 // if the locker is not currently held.
-func (l *consulLocker) Unlock() error {
+func (l *Locker) Unlock() error {
 	if l.lock == nil {
 		return locker.ErrLockNotHeld
 	}

broker/consulbroker/locker_test.go

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

broker/mutexbroker/broker.go

@@ -23,7 +23,7 @@ func New() *Broker {
 
 // NewTask creates new long-running task in current process.
 func (b *Broker) NewTask(key string, fun task.Func) (*task.Task, error) {
-	return task.New(newMutexLocker(b, key), fun), nil
+	return task.New(NewLocker(b, key), fun), nil
 }
 
 func (b *Broker) lock(name string, stopCh <-chan struct{}) error {
@@ -53,45 +53,3 @@ func (b *Broker) unlock(name string) error {
 	<-mc
 	return nil
 }
-
-func newMutexLocker(broker *Broker, key string) *mutexLocker {
-	return &mutexLocker{
-		broker: broker,
-		key:    key,
-	}
-}
-
-type mutexLocker struct {
-	broker   *Broker
-	key      string
-	leaderCh chan struct{}
-}
-
-func (l *mutexLocker) Key() string {
-	return l.key
-}
-
-func (l *mutexLocker) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
-	if l.leaderCh != nil {
-		return nil, locker.ErrLockHeld
-	}
-
-	err := l.broker.lock(l.key, stopCh)
-	if err != nil {
-		return nil, err
-	}
-
-	l.leaderCh = make(chan struct{})
-	return l.leaderCh, nil
-}
-
-func (l *mutexLocker) Unlock() error {
-	if l.leaderCh == nil {
-		return locker.ErrLockNotHeld
-	}
-
-	close(l.leaderCh)
-	l.leaderCh = nil
-
-	return l.broker.unlock(l.key)
-}

broker/mutexbroker/example_test.go

@@ -23,7 +23,7 @@ func ExampleNew() {
 	}
 	defer t.Close()
 
-	t2, err := b.NewTask("task2/key", task2)
+	t2, err := b.NewTask("task/key", task2)
 	if err != nil {
 		log.Printf("New task: %v", err)
 		return

broker/mutexbroker/locker.go

@@ -0,0 +1,59 @@
+package mutexbroker
+
+import (
+	"github.com/monetha/go-distributed/locker"
+)
+
+// Locker implements "distributed locker" using channel.
+type Locker struct {
+	broker   *Broker
+	key      string
+	leaderCh chan struct{}
+}
+
+// NewLocker creates new Locker instance.
+func NewLocker(broker *Broker, key string) *Locker {
+	return &Locker{
+		broker: broker,
+		key:    key,
+	}
+}
+
+// Key returns the name of locker.
+func (l *Locker) 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 *Locker) Lock(stopCh <-chan struct{}) (<-chan struct{}, error) {
+	if l.leaderCh != nil {
+		return nil, locker.ErrLockHeld
+	}
+
+	err := l.broker.lock(l.key, stopCh)
+	if err != nil {
+		return nil, err
+	}
+
+	l.leaderCh = make(chan struct{})
+	return l.leaderCh, nil
+}
+
+// Unlock released the locker. It is an error to call this
+// if the locker is not currently held.
+func (l *Locker) Unlock() error {
+	if l.leaderCh == nil {
+		return locker.ErrLockNotHeld
+	}
+
+	close(l.leaderCh)
+	l.leaderCh = nil
+
+	return l.broker.unlock(l.key)
+}

broker/mutexbroker/locker_test.go

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