statsd_test.go

package statsd

import (
	"bytes"
	"errors"
	"io"
	"io/ioutil"
	"math"
	"net"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"
)

const (
	testAddr = "127.0.0.1:0"
	testKey  = "test_key"
)

var testDate = time.Date(2015, 10, 22, 16, 53, 0, 0, time.UTC)

func TestCount(t *testing.T) {
	testOutput(t, "test_key:5|c", func(c *Client) {
		c.Count(testKey, 5)
	})
}

func TestIncrement(t *testing.T) {
	testOutput(t, "test_key:1|c", func(c *Client) {
		c.Increment(testKey)
	})
}

func TestGauge(t *testing.T) {
	testOutput(t, "test_key:5|g\ntest_key:0|g\ntest_key:-10|g\ntest_key:5|g\ntest_key:0|g\ntest_key:-10|g\ntest_key:0|g\ntest_key:0|g", func(c *Client) {
		c.Gauge(testKey, 5)
		c.Gauge(testKey, -10)
		c.Gauge(testKey, 5.0)
		c.Gauge(testKey, -10.0)
		c.Gauge(testKey, 0.0)
		c.Gauge(testKey, math.Copysign(0, -1))
	})
}

func TestClient_GaugeRelative(t *testing.T) {
	for _, tc := range [...]struct {
		Name   string
		Output string
		Input  func(c *Client)
	}{
		{
			Name:   "inc then dec",
			Output: "test_key:+5|g\ntest_key:-10|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, 5)
				c.GaugeRelative(testKey, -10)
			},
		},
		{
			Name:   "neg",
			Output: "test_key:-3.123|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, -3.123)
			},
		},
		{
			Name:   "pos",
			Output: "test_key:+3.123|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, 3.123)
			},
		},
		{
			Name:   "zero",
			Output: "test_key:+0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, 0)
			},
		},
		{
			Name:   "zero float64",
			Output: "test_key:+0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float64(0))
			},
		},
		{
			Name:   "zero float32",
			Output: "test_key:+0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(0))
			},
		},
		{
			Name:   "neg zero float64",
			Output: "test_key:-0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, math.Copysign(0, -1))
			},
		},
		{
			Name:   "neg zero float32",
			Output: "test_key:-0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(math.Copysign(0, -1)))
			},
		},
		{
			Name:   "pos large",
			Output: "test_key:+5932443000000000000000000000|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(59324.4289e23))
			},
		},
		{
			Name:   "pos small",
			Output: "test_key:+0.00000000000000000059324427|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(59324.4289e-23))
			},
		},
		{
			Name:   "uint8",
			Output: "test_key:+252|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, byte(252))
			},
		},
		{
			Name:   "neg int64",
			Output: "test_key:-9942423|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, int64(-9942423))
			},
		},
		{
			Name:   "uint 0",
			Output: "test_key:+0|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, uint(0))
			},
		},
		{
			Name:   "pos max float64",
			Output: "test_key:+179769313486231570000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, math.MaxFloat64)
			},
		},
		{
			Name:   "neg max float64",
			Output: "test_key:-179769313486231570000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, -math.MaxFloat64)
			},
		},
		{
			Name:   "pos inf float64",
			Output: "test_key:+Inf|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, math.Inf(1))
			},
		},
		{
			Name:   "neg inf float64",
			Output: "test_key:-Inf|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, math.Inf(-1))
			},
		},
		{
			Name:   "pos inf float32",
			Output: "test_key:+Inf|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(math.Inf(1)))
			},
		},
		{
			Name:   "neg inf float32",
			Output: "test_key:-Inf|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(math.Inf(-1)))
			},
		},
		{
			Name:   "nan float64",
			Output: "test_key:+NaN|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, math.NaN())
			},
		},
		{
			Name:   "nan float32",
			Output: "test_key:+NaN|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, float32(math.NaN()))
			},
		},
		{
			Name:   "unsupported",
			Output: "test_key:+|g",
			Input: func(c *Client) {
				c.GaugeRelative(testKey, struct{}{})
			},
		},
	} {
		t.Run(tc.Name, func(t *testing.T) {
			testOutput(t, tc.Output, tc.Input)
		})
	}
}

func TestTiming(t *testing.T) {
	testOutput(t, "test_key:6|ms", func(c *Client) {
		c.Timing(testKey, 6)
	})
}

func TestHistogram(t *testing.T) {
	testOutput(t, "test_key:17|h", func(c *Client) {
		c.Histogram(testKey, 17)
	})
}

func TestNumbers(t *testing.T) {
	testOutput(t,
		"test_key:1|g\n"+
			"test_key:1|g\n"+
			"test_key:2|g\n"+
			"test_key:2|g\n"+
			"test_key:15|g\n"+
			"test_key:15|g\n"+
			"test_key:137|g\n"+
			"test_key:137|g\n"+
			"test_key:1|g\n"+
			"test_key:1|g\n"+
			"test_key:17.6|g\n"+
			"test_key:0|g\n"+
			"test_key:-42.5|g\n"+
			"test_key:|g",
		func(c *Client) {
			c.Gauge(testKey, 1)
			c.Gauge(testKey, uint(1))
			c.Gauge(testKey, int64(2))
			c.Gauge(testKey, uint64(2))
			c.Gauge(testKey, int32(15))
			c.Gauge(testKey, uint32(15))
			c.Gauge(testKey, int16(137))
			c.Gauge(testKey, uint16(137))
			c.Gauge(testKey, int8(1))
			c.Gauge(testKey, uint8(1))
			c.Gauge(testKey, float64(17.6))
			c.Gauge(testKey, float32(-42.5))
			c.Gauge(testKey, "invalid")
		})
}

func TestNewTiming(t *testing.T) {
	i := 0
	now = func() time.Time {
		i++
		switch i {
		default:
			return testDate
		case 2:
			return testDate.Add(10 * time.Millisecond)
		case 3:
			return testDate.Add(100 * time.Millisecond)
		case 4:
			return testDate.Add(time.Second)
		}
	}
	defer func() { now = time.Now }()

	testOutput(t, "test_key:10|ms\ntest_key:1000|ms", func(c *Client) {
		timing := c.NewTiming()
		timing.Send(testKey)

		got := timing.Duration().Nanoseconds()
		want := int64(100 * time.Millisecond)
		if got != want {
			t.Errorf("Duration() = %v, want %v", got, want)
		}

		timing.Send(testKey)
	})
}

func TestUnique(t *testing.T) {
	testOutput(t, "test_key:foo|s", func(c *Client) {
		c.Unique(testKey, "foo")
	})
}

func TestMute(t *testing.T) {
	dialTimeout = func(string, string, time.Duration) (net.Conn, error) {
		t.Fatal("net.Dial should not be called")
		return nil, nil
	}
	defer func() { dialTimeout = net.DialTimeout }()

	c, err := New(Mute(true))
	if err != nil {
		t.Errorf("New() = %v", err)
	}
	c.Increment(testKey)
	c.Gauge(testKey, 1)
	c.GaugeRelative(testKey, 1)
	c.Timing(testKey, 1)
	c.Histogram(testKey, 1)
	c.Unique(testKey, "1")
	c.Flush()
	c.Close()
}

func TestSamplingRateOK(t *testing.T) {
	testOutput(t, "test_key:3|c|@0.6\ntest_key:4|ms|@0.6", func(c *Client) {
		randFloat = func() float32 { return 0.5 }
		c.Count(testKey, 3)
		c.Timing(testKey, 4)
	}, SampleRate(0.6))
}

func TestSamplingRateKO(t *testing.T) {
	testOutput(t, "", func(c *Client) {
		randFloat = func() float32 { return 0.5 }
		c.Count(testKey, 1)
		c.Timing(testKey, 2)
	}, SampleRate(0.3))
}

func TestPrefix(t *testing.T) {
	testOutput(t, "foo.test_key:1|c", func(c *Client) {
		c.Increment(testKey)
	}, Prefix("foo"))
}

func TestNilTags(t *testing.T) {
	testOutput(t, "test_key:1|c", func(c *Client) {
		c.Increment(testKey)
	}, TagsFormat(InfluxDB), Tags())
}

func TestInfluxDBTags(t *testing.T) {
	testOutput(t, "test_key,tag1=value1,tag2=value2:1|c", func(c *Client) {
		c.Increment(testKey)
	}, TagsFormat(InfluxDB), Tags("tag1", "value1", "tag2", "value2"))
}

func TestDatadogTags(t *testing.T) {
	testOutput(t, "test_key:1|c|#tag1:value1,tag2:value2", func(c *Client) {
		c.Increment(testKey)
	}, TagsFormat(Datadog), Tags("tag1", "value1", "tag2", "value2"))
}

func TestNoTagFormat(t *testing.T) {
	testOutput(t, "test_key:1|c", func(c *Client) {
		c.Increment(testKey)
	}, Tags("tag1", "value1", "tag2", "value2"))
}

func TestOddTagsArgs(t *testing.T) {
	dialTimeout = mockDial
	defer func() { dialTimeout = net.DialTimeout }()

	defer func() {
		r := recover()
		if r == nil {
			t.Fatal("Tags should panic when only one argument is provided")
		}
	}()

	_, _ = New(TagsFormat(InfluxDB), Tags("tag1"))
	t.Fatal("A panic should occur")
}

func TestErrorHandler(t *testing.T) {
	errorCount := 0
	testClient(t, func(c *Client) {
		getBuffer(c).err = errors.New("test error")
		c.Increment(testKey)
		c.Close()
		if errorCount != 2 {
			t.Errorf("Wrong error count, got %d, want 2", errorCount)
		}
	}, ErrorHandler(func(err error) {
		if err == nil {
			t.Error("Error should not be nil")
		}
		errorCount++
	}))
}

func TestFlush(t *testing.T) {
	testClient(t, func(c *Client) {
		c.Increment(testKey)
		c.Flush()
		got := getOutput(c)
		want := "test_key:1|c"
		if got != want {
			t.Errorf("Invalid output, got %q, want %q", got, want)
		}
		c.Close()
	})
}

func TestFlushPeriod(t *testing.T) {
	testClient(t, func(c *Client) {
		c.Increment(testKey)
		time.Sleep(time.Millisecond * 400)
		c.conn.mu.Lock()
		got := getOutput(c)
		want := "test_key:1|c"
		if got != want {
			t.Errorf("Invalid output, got %q, want %q", got, want)
		}
		c.conn.mu.Unlock()
		c.Close()
	}, FlushPeriod(time.Nanosecond))
}

func TestFlushPeriod_writeCloser(t *testing.T) {
	c, err := New(
		ErrorHandler(expectNoError(t)),
		FlushPeriod(time.Nanosecond),
		WriteCloser(&testBuffer{}),
	)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	c.Increment(testKey)
	time.Sleep(time.Millisecond * 400)
	c.conn.mu.Lock()
	got := getOutput(c)
	want := "test_key:1|c\n"
	if got != want {
		t.Errorf("Invalid output, got %q, want %q", got, want)
	}
	c.conn.mu.Unlock()
	c.Close()
}

func TestMaxPacketSize(t *testing.T) {
	testClient(t, func(c *Client) {
		c.Increment(testKey)
		conn := getBuffer(c)
		got := conn.buf.String()
		if got != "" {
			t.Errorf("Output should be empty, got %q", got)
		}

		c.Increment(testKey)
		got = conn.buf.String()
		want := "test_key:1|c"
		if got != want {
			t.Errorf("Invalid output, got %q, want %q", got, want)
		}
		conn.buf.Reset()
		c.Close()

		got = conn.buf.String()
		if got != want {
			t.Errorf("Invalid output, got %q, want %q", got, want)
		}
	}, MaxPacketSize(15))
}

func TestClone(t *testing.T) {
	testOutput(t, "test_key:5|c", func(c *Client) {
		c.Clone().Count(testKey, 5)
	})
}

func TestCloneInherits(t *testing.T) {
	testOutput(t, "app.test_key:5|c|@0.5|#tag1:value1", func(c *Client) {
		clone := c.Clone()
		randFloat = func() float32 { return 0.3 }
		clone.Count(testKey, 5)
		randFloat = func() float32 { return 0.8 }
		clone.Count(testKey, 5)
	},
		TagsFormat(Datadog),
		Prefix("app"),
		SampleRate(0.5),
		Tags("tag1", "value1"),
	)
}

func TestCloneFromMuted(t *testing.T) {
	testOutput(t, "", func(c *Client) {
		c.Clone().Count(testKey, 5)
	}, Mute(true))
}

func TestMuteClone(t *testing.T) {
	testOutput(t, "", func(c *Client) {
		c.Clone(Mute(true)).Count(testKey, 5)
	})
}

func TestClonePrefix(t *testing.T) {
	testOutput(t, "app.http.test_key:5|c", func(c *Client) {
		c.Clone(Prefix("http")).Count(testKey, 5)
	}, Prefix("app"))
}

func TestCloneRate(t *testing.T) {
	testOutput(t, "", func(c *Client) {
		randFloat = func() float32 { return 0.8 }
		c.Clone(SampleRate(0.5)).Count(testKey, 5)
	})
}

func TestCloneInfluxDBTags(t *testing.T) {
	testOutput(t, "test_key,tag1=value3,tag3=value4,tag4=value9,tag5=value6,tag2=value2:5|c", func(c *Client) {
		clone := c.Clone(Tags("tag2", "value2", "tag1", "value3", "tag4", "value8", "tag4", "value9"))
		clone.Count(testKey, 5)
	}, TagsFormat(InfluxDB), Tags("tag1", "value1", "tag3", "value4", "tag4", "value5", "tag5", "value6", "tag4", "value7"))
}

func TestCloneDatadogTags(t *testing.T) {
	testOutput(t, "test_key:5|c|#tag1:value3,tag3:value4,tag2:value2", func(c *Client) {
		clone := c.Clone(Tags("tag1", "value3", "tag2", "value2"))
		clone.Count(testKey, 5)
	}, TagsFormat(Datadog), Tags("tag1", "value1", "tag3", "value4"))
}

func TestDialError(t *testing.T) {
	dialTimeout = func(string, string, time.Duration) (net.Conn, error) {
		return nil, errors.New("")
	}
	defer func() { dialTimeout = net.DialTimeout }()

	c, err := New()
	if c == nil || !c.muted {
		t.Error("New() did not return a muted client")
	}
	if err == nil {
		t.Error("New() did not return an error")
	}
}

func TestConcurrency(t *testing.T) {
	testOutput(t, "test_key:1|c\ntest_key:1|c\ntest_key:1|c", func(c *Client) {
		var wg sync.WaitGroup
		wg.Add(1)
		c.Increment(testKey)
		go func() {
			c.Increment(testKey)
			wg.Done()
		}()
		c.Increment(testKey)
		wg.Wait()
	})
}

func TestNew_udpNotListening(t *testing.T) {
	for _, tc := range [...]struct {
		Name    string
		Options []Option
		Muted   bool
		Errored bool
	}{
		{
			Name:    `default`,
			Muted:   true,
			Errored: true,
		},
		{
			Name:    `true`,
			Options: []Option{UDPCheck(true)},
			Muted:   true,
			Errored: true,
		},
		{
			Name:    `false`,
			Options: []Option{UDPCheck(false)},
		},
		{
			Name: `tcp true`,
			Options: []Option{
				Network(`tcp`),
				UDPCheck(true),
			},
		},
		{
			Name: `wc default`,
			Options: []Option{
				WriteCloser(&mockClosedUDPConn{}),
			},
		},
		{
			Name: `wc true`,
			Options: []Option{
				WriteCloser(&mockClosedUDPConn{}),
				UDPCheck(true),
			},
			Muted:   true,
			Errored: true,
		},
		{
			Name: `wc false`,
			Options: []Option{
				WriteCloser(&mockClosedUDPConn{}),
				UDPCheck(false),
			},
		},
	} {
		t.Run(tc.Name, func(t *testing.T) {
			dialTimeout = mockUDPClosed
			defer func() { dialTimeout = net.DialTimeout }()
			c, err := New(tc.Options...)
			if c == nil {
				t.Fatal(`client should never be nil`)
			}
			if c.muted != tc.Muted {
				t.Error(c.muted)
			}
			if (err != nil) != tc.Errored {
				t.Error(err)
			}
		})
	}
}

func TestWriteCloser(t *testing.T) {
	count := 0
	writer := &mockWriteCloser{
		close: func() error {
			count++
			return nil
		},
	}
	config := new(config)
	WriteCloser(writer)(config)
	if v := config.Conn.WriteCloser; v != writer {
		t.Fatal(v)
	}
	if count != 0 {
		t.Fatal(writer)
	}
	WriteCloser(nil)(config)
	if v := config.Conn.WriteCloser; v != nil {
		t.Fatal(v)
	}
	if count != 1 {
		t.Fatal(writer)
	}
}

func TestNew_writeCloserClosesOnMute(t *testing.T) {
	count := 0
	writer := &mockWriteCloser{
		close: func() error {
			count++
			return nil
		},
	}
	client, err := New(
		FlushPeriod(0),
		ErrorHandler(expectNoError(t)),
		WriteCloser(writer),
		Mute(true),
	)
	if client == nil || err != nil {
		t.Fatal(client, err)
	}
	if client.conn.w != nil {
		t.Error(client.conn.w)
	}
	if count != 1 {
		t.Error(count)
	}
}

func TestNew_inlineFlush(t *testing.T) {
	defer func() func() {
		startGoroutines := runtime.NumGoroutine()
		return func() {
			endGoroutines := runtime.NumGoroutine()
			if startGoroutines < endGoroutines {
				t.Error(startGoroutines, endGoroutines)
			}
		}
	}()()
	client, err := New(
		FlushPeriod(0),
		ErrorHandler(expectNoError(t)),
		WriteCloser(&mockWriteCloser{}),
		InlineFlush(true),
	)
	if client == nil || err != nil || client.muted {
		t.Fatal(client, err)
	}
	if !client.conn.inlineFlush {
		t.Error(client.conn.inlineFlush)
	}
}

func TestInlineFlush(t *testing.T) {
	config := new(config)
	InlineFlush(true)(config)
	if !config.Conn.InlineFlush {
		t.Error(config.Conn.InlineFlush)
	}
}

func TestConn_flushNecessary_inlineFlush(t *testing.T) {
	if !(&conn{inlineFlush: true}).flushNecessary() {
		t.Error(`expected always true if always flush is enabled`)
	}
}

func TestConn_flushIfNecessary_inlineFlush(t *testing.T) {
	var (
		called  bool
		flushed string
		c       = &conn{
			buf:           []byte("test_key:1|c\n"),
			inlineFlush:   true,
			maxPacketSize: 100,
			w: &mockWriteCloser{
				write: func(b []byte) (int, error) {
					if called {
						t.Error(`called more than once`)
					}
					called = true
					flushed = string(b)
					return len(b), nil
				},
			},
		}
	)
	// will actually flush everything
	c.flushIfNecessary(2)
	if !called {
		t.Error(called)
	}
	if flushed != "test_key:1|c\n" {
		t.Error(flushed)
	}
	if len(c.buf) != 0 {
		t.Error(c.buf)
	}
}

type mockWriteCloser struct {
	write func(p []byte) (n int, err error)
	close func() error
}

func (m *mockWriteCloser) Write(p []byte) (n int, err error) {
	if m.write != nil {
		return m.write(p)
	}
	panic("implement me")
}

func (m *mockWriteCloser) Close() error {
	if m.close != nil {
		return m.close()
	}
	panic("implement me")
}

type mockClosedUDPConn struct {
	i int
	net.Conn
}

func (c *mockClosedUDPConn) Write(p []byte) (int, error) {
	c.i++
	if c.i == 2 {
		return 0, errors.New("test error")
	}
	return 0, nil
}

func (c *mockClosedUDPConn) Close() error {
	return nil
}

func mockUDPClosed(string, string, time.Duration) (net.Conn, error) {
	return &mockClosedUDPConn{}, nil
}

func testClient(t *testing.T, f func(*Client), options ...Option) {
	t.Helper()
	dialTimeout = mockDial
	defer func() { dialTimeout = net.DialTimeout }()

	options = append([]Option{
		FlushPeriod(0),
		ErrorHandler(expectNoError(t)),
	}, options...)
	c, err := New(options...)
	if err != nil {
		t.Fatalf("New: %v", err)
	}

	f(c)
}

func testOutput(t *testing.T, want string, f func(*Client), options ...Option) {
	t.Helper()
	testClient(t, func(c *Client) {
		t.Helper()
		f(c)
		c.Close()

		got := getOutput(c)
		if got != want {
			t.Errorf("Invalid output, got:\n%q\nwant:\n%q", got, want)
		}
	}, options...)
}

func expectNoError(t *testing.T) func(error) {
	return func(err error) {
		t.Errorf("ErrorHandler should not receive an error: %v", err)
	}
}

type testBuffer struct {
	buf bytes.Buffer
	err error
	net.Conn
}

func (c *testBuffer) Write(p []byte) (int, error) {
	if c.err != nil {
		return 0, c.err
	}
	return c.buf.Write(p)
}

func (c *testBuffer) Close() error {
	return c.err
}

func getBuffer(c *Client) *testBuffer {
	if mock, ok := c.conn.w.(*testBuffer); ok {
		return mock
	}
	return nil
}

func getOutput(c *Client) string {
	if c.conn.w == nil {
		return ""
	}
	return getBuffer(c).buf.String()
}

func mockDial(string, string, time.Duration) (net.Conn, error) {
	return &testBuffer{}, nil
}

func TestUDP(t *testing.T) {
	testNetwork(t, "udp", nil)
}

func TestTCP(t *testing.T) {
	testNetwork(t, "tcp", nil)
}

func testNetwork(t *testing.T, network string, handler func(srv *server) []Option) {
	received := make(chan bool)
	server := newServer(t, network, testAddr, func(p []byte) {
		s := string(p)
		if network != "udp" {
			if !strings.HasSuffix(s, "\n") {
				t.Error(s)
			} else {
				s = s[:len(s)-1]
			}
		}
		if s != "test_key:1|c" {
			t.Errorf("invalid output: %q", s)
		}
		received <- true
	})
	defer server.Close()

	options := []Option{
		Address(server.addr),
		Network(network),
		ErrorHandler(expectNoError(t)),
	}

	if handler != nil {
		options = append(options, handler(server)...)
	}

	c, err := New(options...)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	c.Increment(testKey)
	c.Close()
	select {
	case <-time.After(100 * time.Millisecond):
		t.Error("server received nothing after 100ms")
	case <-received:
	}
}

type server struct {
	t      testing.TB
	addr   string
	closer io.Closer
	closed chan bool
}

func newServer(t testing.TB, network, addr string, f func([]byte)) *server {
	s := &server{t: t, closed: make(chan bool)}
	switch network {
	case "udp":
		laddr, err := net.ResolveUDPAddr("udp", addr)
		if err != nil {
			t.Fatal(err)
		}
		conn, err := net.ListenUDP("udp", laddr)
		if err != nil {
			t.Fatal(err)
		}
		s.closer = conn
		s.addr = conn.LocalAddr().String()
		go func() {
			buf := make([]byte, 1024)
			for {
				n, err := conn.Read(buf)
				if err != nil {
					s.closed <- true
					return
				}
				if n > 0 {
					f(buf[:n])
				}
			}
		}()
	case "tcp":
		ln, err := net.Listen("tcp", addr)
		if err != nil {
			t.Fatal(err)
		}
		s.closer = ln
		s.addr = ln.Addr().String()
		go func() {
			for {
				conn, err := ln.Accept()
				if err != nil {
					s.closed <- true
					return
				}
				p, err := ioutil.ReadAll(conn)
				if err != nil {
					t.Fatal(err)
				}
				if err := conn.Close(); err != nil {
					t.Fatal(err)
				}
				f(p)
			}
		}()
	default:
		t.Fatalf("Invalid network: %q", network)
	}

	return s
}

func (s *server) Close() {
	if err := s.closer.Close(); err != nil {
		s.t.Error(err)
	}
	<-s.closed
}

func Benchmark(b *testing.B) {
	serv := newServer(b, "udp", testAddr, func([]byte) {})
	c, err := New(Address(serv.addr), FlushPeriod(0))
	if err != nil {
		b.Fatal(err)
	}
	for i := 0; i < b.N; i++ {
		c.Increment(testKey)
		c.Count(testKey, i)
		c.Gauge(testKey, i)
		c.Timing(testKey, i)
		c.NewTiming().Send(testKey)
	}
	c.Close()
	serv.Close()
}