Merge pull request #3 from andriyko/fix-client-conn

Fix client conn #2

conn.go

@@ -71,18 +71,19 @@ func newConn(conf connConfig, muted bool) (*conn, error) {
 
 func (c *conn) metric(prefix, bucket string, n interface{}, typ string, rate float32, tags string) {
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	l := len(c.buf)
 	c.appendBucket(prefix, bucket, tags)
 	c.appendNumber(n)
 	c.appendType(typ)
 	c.appendRate(rate)
 	c.closeMetric(tags)
 	c.flushIfBufferFull(l)
-	c.mu.Unlock()
 }
 
 func (c *conn) gauge(prefix, bucket string, value interface{}, tags string) {
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	l := len(c.buf)
 	// To set a gauge to a negative value we must first set it to 0.
 	// https://github.com/etsy/statsd/blob/master/docs/metric_types.md#gauges
@@ -93,7 +94,6 @@ func (c *conn) gauge(prefix, bucket string, value interface{}, tags string) {
 	c.appendBucket(prefix, bucket, tags)
 	c.appendGauge(value, tags)
 	c.flushIfBufferFull(l)
-	c.mu.Unlock()
 }
 
 func (c *conn) appendGauge(value interface{}, tags string) {
@@ -104,13 +104,13 @@ func (c *conn) appendGauge(value interface{}, tags string) {
 
 func (c *conn) unique(prefix, bucket string, value string, tags string) {
 	c.mu.Lock()
+	defer c.mu.Unlock()
 	l := len(c.buf)
 	c.appendBucket(prefix, bucket, tags)
 	c.appendString(value)
 	c.appendType("s")
 	c.closeMetric(tags)
 	c.flushIfBufferFull(l)
-	c.mu.Unlock()
 }
 
 func (c *conn) appendByte(b byte) {

options.go

@@ -148,10 +148,10 @@ func Tags(tags ...string) Option {
 
 		for _, newTag := range newTags {
 			exists := false
-			for _, oldTag := range c.Client.Tags {
-				if newTag.K == oldTag.K {
+			for j := range c.Client.Tags {
+				if newTag.K == c.Client.Tags[j].K {
 					exists = true
-					oldTag.V = newTag.V
+					c.Client.Tags[j].V = newTag.V
 				}
 			}
 			if !exists {

statsd.go

@@ -105,6 +105,9 @@ func (c *Client) Timing(bucket string, value interface{}) {
 	if c.skip() {
 		return
 	}
+	if v, ok := value.(time.Duration); ok {
+		value = float64(v) / float64(time.Millisecond)
+	}
 	c.conn.metric(c.prefix, bucket, value, "ms", c.rate, c.tags)
 }
 
@@ -114,6 +117,9 @@ func (c *Client) TimingWithTags(bucket string, value interface{}, tags map[strin
 		return
 	}
 	strTags := joinTagsMap(c.conn.tagFormat, tags)
+	if v, ok := value.(time.Duration); ok {
+		value = float64(v) / float64(time.Millisecond)
+	}
 	c.conn.metric(c.prefix, bucket, value, "ms", c.rate, strTags)
 }
 
@@ -174,15 +180,15 @@ func (c *Client) NewTiming() Timing {
 
 // Send sends the time elapsed since the creation of the Timing.
 func (t Timing) Send(bucket string) {
-	t.c.Timing(bucket, int(t.Duration()/time.Millisecond))
+	t.c.Timing(bucket, t.Duration())
 }
 
 // SendWithTags sends the time elapsed since the creation of the Timing
 func (t Timing) SendWithTags(bucket string, tags map[string]string) {
-	t.c.TimingWithTags(bucket, float64(t.Duration())/float64(time.Millisecond), tags)
+	t.c.TimingWithTags(bucket, t.Duration, tags)
 }
 
 // Duration returns the time elapsed since the creation of the Timing.
 func (t Timing) Duration() time.Duration {
-	return time.Since(t.start)
+	return now().Sub(t.start)
 }

statsd_test.go

@@ -324,14 +324,14 @@ func TestCloneRate(t *testing.T) {
 }
 
 func TestCloneInfluxDBTags(t *testing.T) {
-	testOutput(t, "test_key,tag1=value1,tag2=value2:5|c", func(c *Client) {
+	testOutput(t, "test_key,tag1=value3,tag2=value2:5|c", func(c *Client) {
 		clone := c.Clone(Tags("tag1", "value3", "tag2", "value2"))
 		clone.Count(testKey, 5)
 	}, TagsFormat(InfluxDB), Tags("tag1", "value1"))
 }
 
 func TestCloneDatadogTags(t *testing.T) {
-	testOutput(t, "test_key:5|c|#tag1:value1,tag2:value2", func(c *Client) {
+	testOutput(t, "test_key:5|c|#tag1:value3,tag2:value2", func(c *Client) {
 		clone := c.Clone(Tags("tag1", "value3", "tag2", "value2"))
 		clone.Count(testKey, 5)
 	}, TagsFormat(Datadog), Tags("tag1", "value1"))