use sink name and command name for consule lock key

command/allocations/app.go

@@ -60,7 +60,7 @@ func NewFirehose() (*Firehose, error) {
 }
 
 func (f *Firehose) Name() string {
-	return "allocations"
+	return "allocations-" + f.sink.Name()
 }
 
 func (f *Firehose) UpdateCh() <-chan interface{} {

command/deployments/app.go

@@ -42,7 +42,7 @@ func NewFirehose() (*Firehose, error) {
 }
 
 func (f *Firehose) Name() string {
-	return "deployments"
+	return "deployments-" + f.sink.Name()
 }
 
 func (f *Firehose) UpdateCh() <-chan interface{} {

command/evaluations/app.go

@@ -42,7 +42,7 @@ func NewFirehose() (*Firehose, error) {
 }
 
 func (f *Firehose) Name() string {
-	return "evaluations"
+	return "evaluations-" + f.sink.Name()
 }
 
 func (f *Firehose) UpdateCh() <-chan interface{} {

command/jobs/job.go

@@ -24,7 +24,7 @@ func NewJobFirehose() (*JobFirehose, error) {
 
 
 func (f *JobFirehose) Name() string {
-	return "jobs"
+	return "jobs-" + f.sink.Name()
 }
 
 // Publish an update from the firehose

command/jobs/jobsliststub.go

@@ -23,7 +23,7 @@ func NewJobListStubFirehose() (*JobListStubFirehose, error) {
 }
 
 func (f *JobListStubFirehose) Name() string {
-	return "jobliststub"
+	return "jobliststub-" + f.sink.Name()
 }
 
 func (f *JobListStubFirehose) Start() {

sink/http.go

@@ -101,3 +101,8 @@ func (s *HttpSink) send(id int) {
 		}
 	}
 }
+
+// Name ..
+func (s *HttpSink) Name() string {
+	return "http"
+}

sink/kafka.go

@@ -131,3 +131,8 @@ func (s *KafkaSink) write() {
 		}
 	}
 }
+
+// Name ..
+func (s *KafkaSink) Name() string {
+	return "kafka"
+}

sink/kinesis.go

@@ -112,3 +112,8 @@ func (s *KinesisSink) write(id int) {
 		}
 	}
 }
+
+// Name ..
+func (s *KinesisSink) Name() string {
+	return "kinesis"
+}

sink/mongodb.go

@@ -137,3 +137,8 @@ func (s *MongodbSink) write(id int) {
 		}
 	}
 }
+
+// Name ..
+func (s *MongodbSink) Name() string {
+	return "mongodb"
+}

sink/nsq.go

@@ -94,3 +94,8 @@ func (s *NSQSink) write(id int) {
 		}
 	}
 }
+
+// Name ..
+func (s *NSQSink) Name() string {
+	return "nsq"
+}

sink/rabbitmq.go

@@ -136,3 +136,8 @@ func (s *RabbitmqSink) write(id int) {
 		}
 	}
 }
+
+// RabbitmqSink ..
+func (s *RabbitmqSink) Name() string {
+	return "rabbitmq"
+}

sink/redis.go

@@ -102,3 +102,8 @@ func (s *RedisSink) write() {
 		}
 	}
 }
+
+// Name ..
+func (s *RedisSink) Name() string {
+	return "redis"
+}

sink/stdout.go

@@ -55,3 +55,9 @@ func (s *StdoutSink) Put(data []byte) error {
 	fmt.Println(string(data))
 	return nil
 }
+
+// Name ..
+func (s *StdoutSink) Name() string {
+	return "stdout"
+}
+

sink/structs.go

@@ -5,4 +5,5 @@ type Sink interface {
 	Start() error
 	Stop()
 	Put(data []byte) error
+	Name() string
 }