Function TestTCPChunks

input/tcp_test.go:101–188

This test checks that, given a reasonable amount of time (500ms) and under normal conditions, all log lines sent by chunk on the TCP socket are received by the output.

(t *testing.T)

// normal conditions, all log lines sent by chunk on the TCP socket are
// received by the output.
func TestTCPChunks(t *testing.T) {
	toml := `
	[fields]
	names = ["f0", "f1", "f2"]

	[input]
	name="TCP"

	[output]
	name="RawRecorder"
	procs=1
	`
	c := baker.Components{
		Inputs:  []baker.InputDesc{TCPDesc},
		Outputs: []baker.OutputDesc{outputtest.RawRecorderDesc},
	}

	cfg, err := baker.NewConfigFromToml(strings.NewReader(toml), c)
	if err != nil {
		t.Error(err)
	}

	topology, err := baker.NewTopologyFromConfig(cfg)
	if err != nil {
		t.Error(err)
	}
	topology.Start()
	// Give baker some time to start the tcp server
	time.Sleep(500 * time.Millisecond)

	const (
		nchunks   = 100 // num lines fed
		chunksize = 37  // chunk size
	)

	errc := make(chan error, 1)
	go func() {
		conn, err := net.Dial("tcp", ":6000")
		if err != nil {
			errc <- err
			return
		}
		defer conn.Close()

		w := gzip.NewWriter(conn)
		defer w.Close()

		buf := &bytes.Buffer{}

		for i := 0; i < nchunks; i++ {
			buf.Reset()
			for j := 0; j < chunksize; j++ {
				l := baker.LogLine{FieldSeparator: baker.DefaultLogLineFieldSeparator}
				l.Set(1, []byte("field"))
				buf.Write(l.ToText(nil))
				buf.WriteByte('\n')
				if _, err := buf.WriteTo(w); err != nil {
					errc <- err