func StatsdInputSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	pConfig := NewPipelineConfig(nil)
	ith := new(plugins_ts.InputTestHelper)
	ith.Msg = pipeline_ts.GetTestMessage()
	ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())
	ith.PackSupply = make(chan *PipelinePack, 1)

	// Specify localhost, but we're not really going to use the network
	ith.AddrStr = "localhost:55565"
	ith.ResolvedAddrStr = "127.0.0.1:55565"

	// set up mock helper, input runner, and stat accumulator
	ith.MockHelper = NewMockPluginHelper(ctrl)
	ith.MockInputRunner = NewMockInputRunner(ctrl)
	mockStatAccum := NewMockStatAccumulator(ctrl)

	c.Specify("A StatsdInput", func() {
		statsdInput := StatsdInput{}
		config := statsdInput.ConfigStruct().(*StatsdInputConfig)

		config.Address = ith.AddrStr
		err := statsdInput.Init(config)
		c.Assume(err, gs.IsNil)
		realListener := statsdInput.listener
		c.Expect(realListener.LocalAddr().String(), gs.Equals, ith.ResolvedAddrStr)
		realListener.Close()
		mockListener := pipeline_ts.NewMockConn(ctrl)
		statsdInput.listener = mockListener

		ith.MockHelper.EXPECT().StatAccumulator("StatAccumInput").Return(mockStatAccum, nil)
		mockListener.EXPECT().Close()
		mockListener.EXPECT().SetReadDeadline(gomock.Any())

		c.Specify("sends a Stat to the StatAccumulator", func() {
			statName := "sample.count"
			statVal := 303
			msg := fmt.Sprintf("%s:%d|c\n", statName, statVal)
			expected := Stat{statName, strconv.Itoa(statVal), "c", float32(1)}
			mockStatAccum.EXPECT().DropStat(expected).Return(true)
			readCall := mockListener.EXPECT().Read(make([]byte, 512))
			readCall.Return(len(msg), nil)
			readCall.Do(func(msgBytes []byte) {
				copy(msgBytes, []byte(msg))
				statsdInput.Stop()
			})
			var wg sync.WaitGroup
			wg.Add(1)
			go func() {
				err = statsdInput.Run(ith.MockInputRunner, ith.MockHelper)
				c.Expect(err, gs.IsNil)
				wg.Done()
			}()
			wg.Wait()
		})
	})
}

func TestMatchers(t *testing.T) {
	type e interface{}
	type testCase struct {
		matcher gomock.Matcher
		yes, no []e
	}
	tests := []testCase{
		testCase{gomock.Any(), []e{3, nil, "foo"}, nil},
		testCase{gomock.Eq(4), []e{4}, []e{3, "blah", nil, int64(4)}},
		testCase{gomock.Nil(),
			[]e{nil, (error)(nil), (chan bool)(nil), (*int)(nil)},
			[]e{"", 0, make(chan bool), errors.New("err"), new(int)}},
		testCase{gomock.Not(gomock.Eq(4)), []e{3, "blah", nil, int64(4)}, []e{4}},
	}
	for i, test := range tests {
		for _, x := range test.yes {
			if !test.matcher.Matches(x) {
				t.Errorf(`test %d: "%v %s" should be true.`, i, x, test.matcher)
			}
		}
		for _, x := range test.no {
			if test.matcher.Matches(x) {
				t.Errorf(`test %d: "%v %s" should be false.`, i, x, test.matcher)
			}
		}
	}
}

func TestSocketListenConfig(t *testing.T) {
	mockCtrl := gomock.NewController(t)
	defer mockCtrl.Finish()

	mckLogger := NewMockLogger(mockCtrl)
	mckLogger.EXPECT().ShouldLog(gomock.Any()).Return(true).AnyTimes()
	mckLogger.EXPECT().Log(gomock.Any(), gomock.Any(),
		gomock.Any(), gomock.Any()).AnyTimes()
	mckStat := NewMockStatistician(mockCtrl)
	mckListenerConfig := NewMockListenerConfig(mockCtrl)

	app := NewApplication()
	app.hostname = "example.org"
	app.SetLogger(mckLogger)
	app.SetMetrics(mckStat)

	sh := NewSocketHandler()
	sh.setApp(app)

	// Should forward Listen errors.
	listenErr := errors.New("splines not reticulated")
	mckListenerConfig.EXPECT().Listen().Return(nil, listenErr)
	if err := sh.listenWithConfig(mckListenerConfig); err != listenErr {
		t.Errorf("Wrong error: got %#v; want %#v", err, listenErr)
	}

	// Should use the wss:// scheme if UseTLS returns true.
	ml := newMockListener(netAddr{"test", "[::1]:8080"})
	gomock.InOrder(
		mckListenerConfig.EXPECT().Listen().Return(ml, nil),
		mckListenerConfig.EXPECT().UseTLS().Return(true),
		mckListenerConfig.EXPECT().GetMaxConns().Return(1),
	)
	if err := sh.listenWithConfig(mckListenerConfig); err != nil {
		t.Errorf("Error setting listener: %s", err)
	}
	if maxConns := sh.MaxConns(); maxConns != 1 {
		t.Errorf("Mismatched maximum connection count: got %d; want 1",
			maxConns)
	}
	expectedURL := "wss://example.org:8080"
	if url := sh.URL(); url != expectedURL {
		t.Errorf("Mismatched handler URL: got %q; want %q",
			url, expectedURL)
	}
}

func UdpInputSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	config := NewPipelineConfig(nil)
	ith := new(plugins_ts.InputTestHelper)
	ith.Msg = pipeline_ts.GetTestMessage()
	ith.Pack = NewPipelinePack(config.InputRecycleChan())

	// set up mock helper, decoder set, and packSupply channel
	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)
	ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)

	c.Specify("A UdpInput", func() {
		udpInput := UdpInput{}
		config := &UdpInputConfig{}

		mbytes, _ := proto.Marshal(ith.Msg)
		header := &message.Header{}
		header.SetMessageLength(uint32(len(mbytes)))
		hbytes, _ := proto.Marshal(header)
		buf := encodeMessage(hbytes, mbytes)

		bytesChan := make(chan []byte, 1)

		ith.MockInputRunner.EXPECT().Name().Return("mock_name")
		ith.MockInputRunner.EXPECT().NewSplitterRunner("").Return(ith.MockSplitterRunner)
		ith.MockSplitterRunner.EXPECT().GetRemainingData().AnyTimes()
		ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)
		ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())

		splitCall := ith.MockSplitterRunner.EXPECT().SplitStream(gomock.Any(),
			nil).AnyTimes()
		splitCall.Do(func(conn net.Conn, del Deliverer) {
			recd := make([]byte, 65536)
			n, _ := conn.Read(recd)
			recd = recd[:n]
			bytesChan <- recd
		})

		c.Specify("using a udp address", func() {
			ith.AddrStr = "localhost:55565"
			ith.ResolvedAddrStr = "127.0.0.1:55565"
			config.Net = "udp"
			config.Address = ith.AddrStr

			err := udpInput.Init(config)
			c.Assume(err, gs.IsNil)
			realListener := (udpInput.listener).(*net.UDPConn)
			c.Expect(realListener.LocalAddr().String(), gs.Equals, ith.ResolvedAddrStr)

			c.Specify("passes the connection to SplitStream", func() {
				go udpInput.Run(ith.MockInputRunner, ith.MockHelper)

				conn, err := net.Dial("udp", ith.AddrStr)
				c.Assume(err, gs.IsNil)
				_, err = conn.Write(buf)
				c.Assume(err, gs.IsNil)
				conn.Close()

				recd := <-bytesChan
				c.Expect(string(recd), gs.Equals, string(buf))
				udpInput.Stop()
			})
		})

		if runtime.GOOS != "windows" {
			c.Specify("using a unix datagram socket", func() {
				tmpDir, err := ioutil.TempDir("", "heka-socket")
				c.Assume(err, gs.IsNil)
				unixPath := filepath.Join(tmpDir, "unixgram-socket")
				ith.AddrStr = unixPath
				config.Net = "unixgram"
				config.Address = ith.AddrStr

				err = udpInput.Init(config)
				c.Assume(err, gs.IsNil)
				realListener := (udpInput.listener).(*net.UnixConn)
				c.Expect(realListener.LocalAddr().String(), gs.Equals, unixPath)

				c.Specify("passes the socket to SplitStream", func() {
					go udpInput.Run(ith.MockInputRunner, ith.MockHelper)

					unixAddr, err := net.ResolveUnixAddr("unixgram", unixPath)
					c.Assume(err, gs.IsNil)
					conn, err := net.DialUnix("unixgram", nil, unixAddr)
					c.Assume(err, gs.IsNil)
					_, err = conn.Write(buf)
					c.Assume(err, gs.IsNil)
					conn.Close()

					recd := <-bytesChan
					c.Expect(string(recd), gs.Equals, string(buf))
					udpInput.Stop()
				})
			})
		}
	})
//.........这里部分代码省略.........

func TestReceivePayloadMessage(t *testing.T) {
	b1 := sarama.NewMockBroker(t, 1)
	b2 := sarama.NewMockBroker(t, 2)
	ctrl := gomock.NewController(t)
	tmpDir, tmpErr := ioutil.TempDir("", "kafkainput-tests")
	if tmpErr != nil {
		t.Errorf("Unable to create a temporary directory: %s", tmpErr)
	}

	defer func() {
		if err := os.RemoveAll(tmpDir); err != nil {
			t.Errorf("Cleanup failed: %s", err)
		}
		ctrl.Finish()
	}()

	topic := "test"
	mdr := new(sarama.MetadataResponse)
	mdr.AddBroker(b2.Addr(), b2.BrokerID())
	mdr.AddTopicPartition(topic, 0, 2)
	b1.Returns(mdr)

	or := new(sarama.OffsetResponse)
	or.AddTopicPartition(topic, 0, 0)
	b2.Returns(or)

	fr := new(sarama.FetchResponse)
	fr.AddMessage(topic, 0, nil, sarama.ByteEncoder([]byte{0x41, 0x42}), 0)
	b2.Returns(fr)

	pConfig := NewPipelineConfig(nil)
	pConfig.Globals.BaseDir = tmpDir
	ki := new(KafkaInput)
	ki.SetName(topic)
	ki.SetPipelineConfig(pConfig)
	config := ki.ConfigStruct().(*KafkaInputConfig)
	config.Addrs = append(config.Addrs, b1.Addr())
	config.Topic = topic

	ith := new(plugins_ts.InputTestHelper)
	ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())
	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)

	ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)

	err := ki.Init(config)
	if err != nil {
		t.Fatalf("%s", err)
	}

	ith.MockInputRunner.EXPECT().NewSplitterRunner("").Return(ith.MockSplitterRunner)
	ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)

	decChan := make(chan func(*PipelinePack), 1)
	decCall := ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())
	decCall.Do(func(dec func(pack *PipelinePack)) {
		decChan <- dec
	})

	bytesChan := make(chan []byte, 1)
	splitCall := ith.MockSplitterRunner.EXPECT().SplitBytes(gomock.Any(), nil)
	splitCall.Do(func(recd []byte, del Deliverer) {
		bytesChan <- recd
	})

	errChan := make(chan error)
	go func() {
		errChan <- ki.Run(ith.MockInputRunner, ith.MockHelper)
	}()

	recd := <-bytesChan
	if string(recd) != "AB" {
		t.Errorf("Invalid Payload Expected: AB received: %s", string(recd))
	}

	packDec := <-decChan
	packDec(ith.Pack)
	if ith.Pack.Message.GetType() != "heka.kafka" {
		t.Errorf("Invalid Type %s", ith.Pack.Message.GetType())
	}

	// There is a hang on the consumer close with the mock broker
	// closing the brokers before the consumer works around the issue
	// and is good enough for this test.
	b1.Close()
	b2.Close()

	ki.Stop()
	err = <-errChan
	if err != nil {
		t.Fatal(err)
	}

	filename := filepath.Join(tmpDir, "kafka", "test.test.0.offset.bin")
	if o, err := readCheckpoint(filename); err != nil {
		t.Errorf("Could not read the checkpoint file: %s", filename)
	} else {
		if o != 1 {
			t.Errorf("Incorrect offset Expected: 1 Received: %d", o)
//.........这里部分代码省略.........

func FilePollingInputSpec(c gs.Context) {
	t := new(pipeline_ts.SimpleT)
	ctrl := gomock.NewController(t)

	tmpFileName := fmt.Sprintf("filepollinginput-test-%d", time.Now().UnixNano())
	tmpFilePath := filepath.Join(os.TempDir(), tmpFileName)

	defer func() {
		ctrl.Finish()
		os.Remove(tmpFilePath)
	}()

	pConfig := NewPipelineConfig(nil)
	var wg sync.WaitGroup
	errChan := make(chan error, 1)
	bytesChan := make(chan []byte, 1)
	tickChan := make(chan time.Time)

	retPackChan := make(chan *PipelinePack, 2)
	defer close(retPackChan)

	c.Specify("A FilePollingInput", func() {
		input := new(FilePollingInput)

		ith := new(plugins_ts.InputTestHelper)
		ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
		ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)
		ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)

		config := input.ConfigStruct().(*FilePollingInputConfig)
		config.FilePath = tmpFilePath

		startInput := func(msgCount int) {
			wg.Add(1)
			go func() {
				errChan <- input.Run(ith.MockInputRunner, ith.MockHelper)
				wg.Done()
			}()
		}

		ith.MockInputRunner.EXPECT().Ticker().Return(tickChan)
		ith.MockHelper.EXPECT().PipelineConfig().Return(pConfig)

		c.Specify("gets updated information when reading a file", func() {

			err := input.Init(config)
			c.Assume(err, gs.IsNil)

			ith.MockInputRunner.EXPECT().NewSplitterRunner("").Return(ith.MockSplitterRunner)
			ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)
			ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())
			splitCall := ith.MockSplitterRunner.EXPECT().SplitStream(gomock.Any(),
				nil).Return(io.EOF).Times(2)
			splitCall.Do(func(f *os.File, del Deliverer) {
				fBytes, err := ioutil.ReadAll(f)
				if err != nil {
					fBytes = []byte(err.Error())
				}
				bytesChan <- fBytes
			})

			startInput(2)

			f, err := os.Create(tmpFilePath)
			c.Expect(err, gs.IsNil)

			_, err = f.Write([]byte("test1"))
			c.Expect(err, gs.IsNil)
			c.Expect(f.Close(), gs.IsNil)

			tickChan <- time.Now()

			msgBytes := <-bytesChan
			c.Expect(string(msgBytes), gs.Equals, "test1")

			f, err = os.Create(tmpFilePath)
			c.Expect(err, gs.IsNil)

			_, err = f.Write([]byte("test2"))
			c.Expect(err, gs.IsNil)
			c.Expect(f.Close(), gs.IsNil)

			tickChan <- time.Now()
			msgBytes = <-bytesChan
			c.Expect(string(msgBytes), gs.Equals, "test2")

			input.Stop()
			wg.Wait()
			c.Expect(<-errChan, gs.IsNil)
		})

	})

}

func TestReceiveProtobufMessage(t *testing.T) {
	broker := sarama.NewMockBroker(t, 2)
	ctrl := gomock.NewController(t)
	tmpDir, tmpErr := ioutil.TempDir("", "kafkainput-tests")
	if tmpErr != nil {
		t.Errorf("Unable to create a temporary directory: %s", tmpErr)
	}

	defer func() {
		if err := os.RemoveAll(tmpDir); err != nil {
			t.Errorf("Cleanup failed: %s", err)
		}
		ctrl.Finish()
	}()

	topic := "test"
	mockFetchResponse := sarama.NewMockFetchResponse(t, 1)
	mockFetchResponse.SetMessage(topic, 0, 0, sarama.ByteEncoder([]byte{0x41, 0x42}))

	broker.SetHandlerByMap(map[string]sarama.MockResponse{
		"MetadataRequest": sarama.NewMockMetadataResponse(t).
			SetBroker(broker.Addr(), broker.BrokerID()).
			SetLeader(topic, 0, broker.BrokerID()),
		"OffsetRequest": sarama.NewMockOffsetResponse(t).
			SetOffset(topic, 0, sarama.OffsetOldest, 0).
			SetOffset(topic, 0, sarama.OffsetNewest, 2),
		"FetchRequest": mockFetchResponse,
	})

	pConfig := NewPipelineConfig(nil)
	pConfig.Globals.BaseDir = tmpDir
	ki := new(KafkaInput)
	ki.SetName(topic)
	ki.SetPipelineConfig(pConfig)
	config := ki.ConfigStruct().(*KafkaInputConfig)
	config.Addrs = append(config.Addrs, broker.Addr())
	config.Topic = topic

	ith := new(plugins_ts.InputTestHelper)
	ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())
	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)

	ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)

	err := ki.Init(config)
	if err != nil {
		t.Fatalf("%s", err)
	}

	ith.MockInputRunner.EXPECT().NewSplitterRunner("").Return(ith.MockSplitterRunner)
	ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(true)
	ith.MockSplitterRunner.EXPECT().Done()

	bytesChan := make(chan []byte, 1)
	splitCall := ith.MockSplitterRunner.EXPECT().SplitBytes(gomock.Any(), nil)
	splitCall.Do(func(recd []byte, del Deliverer) {
		bytesChan <- recd
	})

	errChan := make(chan error)
	go func() {
		errChan <- ki.Run(ith.MockInputRunner, ith.MockHelper)
	}()

	recd := <-bytesChan
	if string(recd) != "AB" {
		t.Errorf("Invalid MsgBytes Expected: AB received: %s", string(recd))
	}

	// There is a hang on the consumer close with the mock broker
	// closing the brokers before the consumer works around the issue
	// and is good enough for this test.
	broker.Close()

	ki.Stop()
	err = <-errChan
	if err != nil {
		t.Fatal(err)
	}
}

func TcpOutputSpec(c gs.Context) {
	t := new(pipeline_ts.SimpleT)
	ctrl := gomock.NewController(t)

	tmpDir, tmpErr := ioutil.TempDir("", "tcp-tests")
	defer func() {
		ctrl.Finish()
		tmpErr = os.RemoveAll(tmpDir)
		c.Expect(tmpErr, gs.Equals, nil)
	}()
	globals := DefaultGlobals()
	globals.BaseDir = tmpDir

	pConfig := NewPipelineConfig(globals)
	pConfig.RegisterDefault("HekaFramingSplitter")

	c.Specify("TcpOutput", func() {
		tcpOutput := new(TcpOutput)
		tcpOutput.SetName("test")
		config := tcpOutput.ConfigStruct().(*TcpOutputConfig)
		tcpOutput.Init(config)

		tickChan := make(chan time.Time)
		oth := plugins_ts.NewOutputTestHelper(ctrl)
		oth.MockOutputRunner.EXPECT().Ticker().Return(tickChan).AnyTimes()
		encoder := new(ProtobufEncoder)
		encoder.SetPipelineConfig(pConfig)
		encoder.Init(nil)

		inChan := make(chan *PipelinePack, 1)

		msg := pipeline_ts.GetTestMessage()
		pack := NewPipelinePack(pConfig.InputRecycleChan())
		pack.Message = msg
		pack.Decoded = true

		outStr := "Write me out to the network"
		newpack := NewPipelinePack(nil)
		newpack.Message = msg
		newpack.Decoded = true
		newpack.Message.SetPayload(outStr)
		matchBytes, err := proto.Marshal(newpack.Message)
		c.Expect(err, gs.IsNil)

		inChanCall := oth.MockOutputRunner.EXPECT().InChan().AnyTimes()
		inChanCall.Return(inChan)

		errChan := make(chan error)
		startOutput := func() {
			go func() {
				oth.MockHelper.EXPECT().PipelineConfig().Return(pConfig).AnyTimes()
				oth.MockOutputRunner.EXPECT().Name().Return("TcpOutput")
				err := tcpOutput.Run(oth.MockOutputRunner, oth.MockHelper)
				errChan <- err
			}()
		}

		c.Specify("doesn't use framing w/o ProtobufEncoder", func() {
			encoder := new(plugins.PayloadEncoder)
			oth.MockOutputRunner.EXPECT().Encoder().Return(encoder)
			err := tcpOutput.Init(config)
			c.Assume(err, gs.IsNil)
			close(inChan)
			startOutput()
			err = <-errChan
			c.Expect(err, gs.IsNil)
			// We should fail if SetUseFraming is called since we didn't
			// EXPECT it.
		})

		c.Specify("doesn't use framing if config says not to", func() {
			useFraming := false
			config.UseFraming = &useFraming
			err := tcpOutput.Init(config)
			c.Assume(err, gs.IsNil)
			close(inChan)
			startOutput()
			err = <-errChan
			c.Expect(err, gs.IsNil)
			// We should fail if SetUseFraming is called since we didn't
			// EXPECT it.
		})

		c.Specify("writes out to the network", func() {
			collectData := func(ch chan string) {
				ln, err := net.Listen("tcp", "localhost:9125")
				if err != nil {
					ch <- err.Error()
					return
				}
				ch <- "ready"
				conn, err := ln.Accept()
				if err != nil {
					ch <- err.Error()
					return
				}
				b := make([]byte, 1000)
				n, _ := conn.Read(b)
				ch <- string(b[0:n])
				conn.Close()
//.........这里部分代码省略.........

func LogstreamerInputSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	here, _ := os.Getwd()
	dirPath := filepath.Join(here, "../../logstreamer", "testdir", "filehandling/subdir")

	tmpDir, tmpErr := ioutil.TempDir("", "hekad-tests")
	c.Expect(tmpErr, gs.Equals, nil)
	defer func() {
		tmpErr = os.RemoveAll(tmpDir)
		c.Expect(tmpErr, gs.IsNil)
	}()

	globals := DefaultGlobals()
	globals.BaseDir = tmpDir
	pConfig := NewPipelineConfig(globals)
	ith := new(plugins_ts.InputTestHelper)
	ith.Msg = pipeline_ts.GetTestMessage()
	ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())

	// Specify localhost, but we're not really going to use the network.
	ith.AddrStr = "localhost:55565"
	ith.ResolvedAddrStr = "127.0.0.1:55565"

	// Set up mock helper, runner, and pack supply channel.
	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)
	ith.MockDeliverer = pipelinemock.NewMockDeliverer(ctrl)
	ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)
	ith.PackSupply = make(chan *PipelinePack, 1)

	c.Specify("A LogstreamerInput", func() {
		lsInput := &LogstreamerInput{pConfig: pConfig}
		lsiConfig := lsInput.ConfigStruct().(*LogstreamerInputConfig)
		lsiConfig.LogDirectory = dirPath
		lsiConfig.FileMatch = `file.log(\.?)(?P<Seq>\d+)?`
		lsiConfig.Differentiator = []string{"logfile"}
		lsiConfig.Priority = []string{"^Seq"}

		c.Specify("w/ no translation map", func() {
			err := lsInput.Init(lsiConfig)
			c.Expect(err, gs.IsNil)
			c.Expect(len(lsInput.plugins), gs.Equals, 1)

			// Create pool of packs.
			numLines := 5 // # of lines in the log file we're parsing.
			packs := make([]*PipelinePack, numLines)
			ith.PackSupply = make(chan *PipelinePack, numLines)
			for i := 0; i < numLines; i++ {
				packs[i] = NewPipelinePack(ith.PackSupply)
				ith.PackSupply <- packs[i]
			}

			c.Specify("reads a log file", func() {
				// Expect InputRunner calls to get InChan and inject outgoing msgs.
				ith.MockInputRunner.EXPECT().LogError(gomock.Any()).AnyTimes()
				ith.MockInputRunner.EXPECT().LogMessage(gomock.Any()).AnyTimes()
				ith.MockInputRunner.EXPECT().NewDeliverer("1").Return(ith.MockDeliverer)
				ith.MockInputRunner.EXPECT().NewSplitterRunner("1").Return(
					ith.MockSplitterRunner)
				ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)
				ith.MockSplitterRunner.EXPECT().IncompleteFinal().Return(false)
				ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())

				getRecCall := ith.MockSplitterRunner.EXPECT().GetRecordFromStream(
					gomock.Any()).Times(numLines)
				line := "boo hoo foo foo"
				getRecCall.Return(len(line), []byte(line), nil)
				getRecCall = ith.MockSplitterRunner.EXPECT().GetRecordFromStream(gomock.Any())
				getRecCall.Return(0, make([]byte, 0), io.EOF)

				deliverChan := make(chan []byte, 1)
				deliverCall := ith.MockSplitterRunner.EXPECT().DeliverRecord(gomock.Any(),
					ith.MockDeliverer).Times(numLines)
				deliverCall.Do(func(record []byte, del Deliverer) {
					deliverChan <- record
				})

				ith.MockDeliverer.EXPECT().Done()

				runOutChan := make(chan error, 1)
				go func() {
					err = lsInput.Run(ith.MockInputRunner, ith.MockHelper)
					runOutChan <- err
				}()

				dur, _ := time.ParseDuration("5s")
				timeout := time.After(dur)
				timed := false
				for x := 0; x < numLines; x++ {
					select {
					case record := <-deliverChan:
						c.Expect(string(record), gs.Equals, line)
					case <-timeout:
						timed = true
						x += numLines
					}
					// Free up the scheduler while we wait for the log file lines
//.........这里部分代码省略.........

func TestSocketInvalidOrigin(t *testing.T) {
	mockCtrl := gomock.NewController(t)
	defer mockCtrl.Finish()

	mckLogger := NewMockLogger(mockCtrl)
	mckLogger.EXPECT().ShouldLog(gomock.Any()).Return(true).AnyTimes()
	mckLogger.EXPECT().Log(gomock.Any(), gomock.Any(),
		gomock.Any(), gomock.Any()).AnyTimes()
	mckStat := NewMockStatistician(mockCtrl)
	mckStat.EXPECT().Increment("client.socket.connect").AnyTimes()
	mckStat.EXPECT().Timer("client.socket.lifespan", gomock.Any()).AnyTimes()
	mckStat.EXPECT().Increment("client.socket.disconnect").AnyTimes()
	mckStore := NewMockStore(mockCtrl)
	mckRouter := NewMockRouter(mockCtrl)

	testWithOrigin := func(allowedOrigins []string,
		config *websocket.Config) error {

		app := NewApplication()
		app.SetLogger(mckLogger)
		app.SetMetrics(mckStat)
		app.SetStore(mckStore)
		app.SetRouter(mckRouter)

		sh := NewSocketHandler()
		defer sh.Close()
		sh.setApp(app)
		sh.setOrigins(allowedOrigins)
		pipe := newPipeListener()
		if err := sh.listenWithConfig(listenerConfig{listener: pipe}); err != nil {
			return err
		}
		sh.server = newServeWaiter(&http.Server{Handler: sh.ServeMux()})
		app.SetSocketHandler(sh)

		errChan := make(chan error, 1)
		go sh.Start(errChan)

		conn, err := dialSocketListener(pipe, config)
		if err != nil {
			return err
		}
		conn.Close()
		return nil
	}

	location := &url.URL{Scheme: "https", Host: "example.com"}
	badURL := &url.URL{Scheme: "[email protected]#$", Host: "^&*-"}

	tests := []struct {
		name           string
		allowedOrigins []string
		config         *websocket.Config
		err            error
	}{
		{
			"Should allow all origins if none are specified",
			nil,
			&websocket.Config{
				Version:  websocket.ProtocolVersionHybi13,
				Location: location,
				Origin:   location,
			},
			nil,
		},
		{
			"Should match multiple origins",
			[]string{"https://example.com", "https://example.org", "https://example.net"},
			&websocket.Config{
				Version:  websocket.ProtocolVersionHybi13,
				Location: location,
				Origin:   &url.URL{Scheme: "https", Host: "example.net"},
			},
			nil,
		},
		{
			"Should reject mismatched origins",
			[]string{"https://example.com"},
			&websocket.Config{
				Version:  websocket.ProtocolVersionHybi13,
				Location: location,
				Origin:   &url.URL{Scheme: "http", Host: "example.org"},
			},
			websocket.ErrBadStatus, // 403 status code.
		},
		{
			"Should reject malformed origins if some are specified",
			[]string{"https://example.com"},
			&websocket.Config{
				Version:  websocket.ProtocolVersionHybi13,
				Location: location,
				Origin:   badURL, // Malformed URL.
			},
			websocket.ErrBadStatus, // 403 status code.
		},
		{
			"Should ignore malformed origins if none are specified",
			nil,
			&websocket.Config{
				Version:  websocket.ProtocolVersionHybi13,
//.........这里部分代码省略.........

//.........这里部分代码省略.........
			reader2 := makeMockReader(b)

			// Call GetRemainingData before using sr on a new stream
			sr.GetRemainingData()
			count, errCount, bytesRead, foundEOFCount, remainingDataLength, finalRecordLength,
				eofRecordLength := readRecordsFromStream(sr, reader2, true)

			c.Expect(errCount, gs.Equals, 0)
			c.Expect(count, gs.Equals, 50)
			c.Expect(foundEOFCount, gs.Equals, 50)
			c.Expect(remainingDataLength, gs.Equals, 0)
			c.Expect(finalRecordLength, gs.Equals, 215)
			c.Expect(eofRecordLength, gs.Equals, 0)
			// Now we see the correct number of bytes being read.
			c.Expect(bytesRead, gs.Equals, len(b))
		})
	})

	c.Specify("A SplitterRunner w/ TokenSplitter", func() {
		splitter := &TokenSplitter{}
		config := splitter.ConfigStruct().(*TokenSplitterConfig)

		c.Specify("sets readPos to 0 when read returns ErrShortBuffer", func() {
			config.Delimiter = "\t"
			err := splitter.Init(config)
			c.Assume(err, gs.IsNil)

			sr := NewSplitterRunner("TokenSplitter", splitter, srConfig)

			b := make([]byte, message.MAX_RECORD_SIZE+1)
			reader := bytes.NewReader(b)

			var n int
			var record []byte
			for err == nil {
				n, record, err = sr.GetRecordFromStream(reader)
			}
			c.Expect(n, gs.Equals, int(message.MAX_RECORD_SIZE))
			c.Expect(len(record), gs.Equals, 0)
			c.Expect(err, gs.Equals, io.ErrShortBuffer)
			c.Expect(sr.readPos, gs.Equals, 0)
			c.Expect(sr.scanPos, gs.Equals, 0)
		})

		c.Specify("checks if splitter honors 'deliver_incomplete_final' setting", func() {

			config.Count = 4
			numRecs := 10
			err := splitter.Init(config)
			c.Assume(err, gs.IsNil)

			packSupply := make(chan *PipelinePack, 1)
			pack := NewPipelinePack(packSupply)
			packSupply <- pack
			ir := NewMockInputRunner(ctrl)
			// ir.EXPECT().InChan().Return(packSupply).Times(numRecs)
			// ir.EXPECT().Name().Return("foo").Times(numRecs)
			ir.EXPECT().InChan().Return(packSupply).AnyTimes()
			ir.EXPECT().Name().Return("foo").AnyTimes()

			incompleteFinal := true
			srConfig.IncompleteFinal = &incompleteFinal
			sr := NewSplitterRunner("TokenSplitter", splitter, srConfig)
			sr.ir = ir

			rExpected := []byte("test1\ntest12\ntest123\npartial\n")
			buf := bytes.Repeat(rExpected, numRecs)
			buf = buf[:len(buf)-1] // 40 lines separated by 39 newlines

			reader := bytes.NewReader(buf)
			mockDel := NewMockDeliverer(ctrl)
			delCall := mockDel.EXPECT().Deliver(gomock.Any()).AnyTimes()
			i := 0
			delCall.Do(func(pack *PipelinePack) {
				i++
				if i < numRecs {
					c.Expect(pack.Message.GetPayload(), gs.Equals, string(rExpected))
				} else {
					c.Expect(pack.Message.GetPayload(), gs.Equals,
						string(rExpected[:len(rExpected)-1]))
				}
				pack.Recycle()
			})
			c.Specify("via SplitStream", func() {
				for err == nil {
					err = sr.SplitStream(reader, mockDel)
				}
				c.Expect(err, gs.Equals, io.EOF)
				c.Expect(i, gs.Equals, numRecs)
			})
			c.Specify("via SplitBytes", func() {
				seekPos, err := sr.SplitBytes(buf, mockDel)
				c.Assume(err, gs.IsNil)
				c.Expect(seekPos, gs.Equals, len(buf))
				c.Expect(i, gs.Equals, numRecs)
			})
		})

	})
}

func AMQPPluginSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	config := NewPipelineConfig(nil)

	// Our two user/conn waitgroups.
	ug := new(sync.WaitGroup)
	cg := new(sync.WaitGroup)

	// Setup the mock channel.
	mch := NewMockAMQPChannel(ctrl)

	// Setup the mock amqpHub with the mock chan return.
	aqh := NewMockAMQPConnectionHub(ctrl)
	aqh.EXPECT().GetChannel("", AMQPDialer{}).Return(mch, ug, cg, nil)

	errChan := make(chan error, 1)
	bytesChan := make(chan []byte, 1)

	c.Specify("An amqp input", func() {
		// Setup all the mock calls for Init.
		mch.EXPECT().ExchangeDeclare("", "", false, true, false, false,
			gomock.Any()).Return(nil)
		mch.EXPECT().QueueDeclare("", false, true, false, false,
			gomock.Any()).Return(amqp.Queue{}, nil)
		mch.EXPECT().QueueBind("", "test", "", false, gomock.Any()).Return(nil)
		mch.EXPECT().Qos(2, 0, false).Return(nil)

		ith := new(plugins_ts.InputTestHelper)
		ith.Msg = pipeline_ts.GetTestMessage()
		ith.Pack = NewPipelinePack(config.InputRecycleChan())

		// Set up relevant mocks.
		ith.MockHelper = NewMockPluginHelper(ctrl)
		ith.MockInputRunner = NewMockInputRunner(ctrl)
		ith.MockSplitterRunner = NewMockSplitterRunner(ctrl)
		ith.PackSupply = make(chan *PipelinePack, 1)

		ith.MockInputRunner.EXPECT().NewSplitterRunner("").Return(ith.MockSplitterRunner)

		amqpInput := new(AMQPInput)
		amqpInput.amqpHub = aqh
		config := amqpInput.ConfigStruct().(*AMQPInputConfig)
		config.URL = ""
		config.Exchange = ""
		config.ExchangeType = ""
		config.RoutingKey = "test"
		config.QueueTTL = 300000

		err := amqpInput.Init(config)
		c.Assume(err, gs.IsNil)
		c.Expect(amqpInput.ch, gs.Equals, mch)

		c.Specify("consumes a text message", func() {
			// Create a channel to send data to the input. Drop a message on
			// there and close the channel.
			streamChan := make(chan amqp.Delivery, 1)
			ack := plugins_ts.NewMockAcknowledger(ctrl)
			ack.EXPECT().Ack(gomock.Any(), false)
			streamChan <- amqp.Delivery{
				ContentType:  "text/plain",
				Body:         []byte("This is a message"),
				Timestamp:    time.Now(),
				Acknowledger: ack,
			}
			mch.EXPECT().Consume("", "", false, false, false, false,
				gomock.Any()).Return(streamChan, nil)

			// Increase the usage since Run decrements it on close.
			ug.Add(1)

			splitCall := ith.MockSplitterRunner.EXPECT().SplitBytes(gomock.Any(),
				nil)
			splitCall.Do(func(recd []byte, del Deliverer) {
				bytesChan <- recd
			})
			ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)
			ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())
			go func() {
				err := amqpInput.Run(ith.MockInputRunner, ith.MockHelper)
				errChan <- err
			}()

			msgBytes := <-bytesChan
			c.Expect(string(msgBytes), gs.Equals, "This is a message")
			close(streamChan)
			err = <-errChan
		})

		c.Specify("consumes a protobuf encoded message", func() {
			encoder := client.NewProtobufEncoder(nil)
			streamChan := make(chan amqp.Delivery, 1)

			msg := new(message.Message)
			msg.SetUuid(uuid.NewRandom())
			msg.SetTimestamp(time.Now().UnixNano())
			msg.SetType("logfile")
			msg.SetLogger("/a/nice/path")
//.........这里部分代码省略.........

func TestSocketOrigin(t *testing.T) {
	var err error

	mockCtrl := gomock.NewController(t)
	defer mockCtrl.Finish()

	mckLogger := NewMockLogger(mockCtrl)
	mckLogger.EXPECT().ShouldLog(gomock.Any()).Return(true).AnyTimes()
	mckLogger.EXPECT().Log(gomock.Any(), gomock.Any(),
		gomock.Any(), gomock.Any()).AnyTimes()
	mckStat := NewMockStatistician(mockCtrl)
	mckStore := NewMockStore(mockCtrl)
	mckRouter := NewMockRouter(mockCtrl)

	app := NewApplication()
	app.SetLogger(mckLogger)
	app.SetMetrics(mckStat)
	app.SetStore(mckStore)
	app.SetRouter(mckRouter)

	sh := NewSocketHandler()
	defer sh.Close()
	sh.setApp(app)

	pipe := newPipeListener()
	defer pipe.Close()
	if err := sh.listenWithConfig(listenerConfig{listener: pipe}); err != nil {
		t.Fatalf("Error setting listener: %s", err)
	}
	sh.server = newServeWaiter(&http.Server{Handler: sh.ServeMux()})
	app.SetSocketHandler(sh)

	errChan := make(chan error, 1)
	go sh.Start(errChan)

	uaid := "5e1e5984569c4f00bf4bea47754a6403"
	gomock.InOrder(
		mckStat.EXPECT().Increment("client.socket.connect"),
		mckStore.EXPECT().CanStore(0).Return(true),
		mckRouter.EXPECT().Register(uaid),
		mckStat.EXPECT().Increment("updates.client.hello"),
		mckStore.EXPECT().FetchAll(uaid, gomock.Any()).Return(nil, nil, nil),
		mckStat.EXPECT().Timer("client.flush", gomock.Any()),
		mckRouter.EXPECT().Unregister(uaid),
		mckStat.EXPECT().Timer("client.socket.lifespan", gomock.Any()),
		mckStat.EXPECT().Increment("client.socket.disconnect"),
	)

	origin := &url.URL{Scheme: "https", Host: "example.com"}
	conn, err := dialSocketListener(pipe, &websocket.Config{
		Location: origin,
		Origin:   origin,
		Version:  websocket.ProtocolVersionHybi13,
	})
	if err != nil {
		t.Fatalf("Error dialing origin: %s", err)
	}
	defer conn.Close()
	err = websocket.JSON.Send(conn, struct {
		Type       string   `json:"messageType"`
		DeviceID   string   `json:"uaid"`
		ChannelIDs []string `json:"channelIDs"`
	}{"hello", uaid, []string{}})
	if err != nil {
		t.Fatalf("Error writing client handshake: %s", err)
	}
	reply := new(HelloReply)
	if err = websocket.JSON.Receive(conn, reply); err != nil {
		t.Fatalf("Error reading server handshake: %s", err)
	}
	if reply.DeviceID != uaid {
		t.Fatalf("Mismatched device ID: got %q; want %q", reply.DeviceID, uaid)
	}

	worker, workerConnected := app.GetWorker(uaid)
	if !workerConnected {
		t.Fatalf("Missing worker for device ID %q", uaid)
	}
	workerOrigin := worker.Origin()
	if expectedOrigin := origin.String(); workerOrigin != expectedOrigin {
		t.Errorf("Mismatched origins: got %q; want %q",
			workerOrigin, expectedOrigin)
	}
}

func HttpInputSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	pConfig := NewPipelineConfig(nil)

	json_post := `{"uuid": "xxBI3zyeXU+spG8Uiveumw==", "timestamp": 1372966886023588, "hostname": "Victors-MacBook-Air.local", "pid": 40183, "fields": [{"representation": "", "value_type": "STRING", "name": "cef_meta.syslog_priority", "value_string": [""]}, {"representation": "", "value_type": "STRING", "name": "cef_meta.syslog_ident", "value_string": [""]}, {"representation": "", "value_type": "STRING", "name": "cef_meta.syslog_facility", "value_string": [""]}, {"representation": "", "value_type": "STRING", "name": "cef_meta.syslog_options", "value_string": [""]}], "logger": "", "env_version": "0.8", "type": "cef", "payload": "Jul 04 15:41:26 Victors-MacBook-Air.local CEF:0|mozilla|weave|3|xx\\\\|x|xx\\\\|x|5|cs1Label=requestClientApplication cs1=MySuperBrowser requestMethod=GET request=/ src=127.0.0.1 dest=127.0.0.1 suser=none", "severity": 6}'`

	c.Specify("A HttpInput", func() {

		httpInput := HttpInput{}
		ith := new(plugins_ts.InputTestHelper)
		ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
		ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)

		runOutputChan := make(chan string, 1)
		startInput := func() {
			go func() {
				err := httpInput.Run(ith.MockInputRunner, ith.MockHelper)
				var runOutput string
				if err != nil {
					runOutput = err.Error()
				}
				runOutputChan <- runOutput
			}()
		}

		ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())
		ith.PackSupply = make(chan *PipelinePack, 1)
		ith.PackSupply <- ith.Pack

		config := httpInput.ConfigStruct().(*HttpInputConfig)

		c.Specify("short circuits packs into the router", func() {
			config.Url = "http://localhost:9876/"
			tickChan := make(chan time.Time)

			ith.MockInputRunner.EXPECT().LogMessage(gomock.Any()).Times(2)

			ith.MockHelper.EXPECT().PipelineConfig().Return(pConfig)
			ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply)
			ith.MockInputRunner.EXPECT().Ticker().Return(tickChan)

			err := httpInput.Init(config)
			c.Assume(err, gs.IsNil)
			startInput()

			tickChan <- time.Now()

			// We need for the pipeline to finish up
			time.Sleep(50 * time.Millisecond)
		})

		c.Specify("with a decoder", func() {

			decoderName := "TestDecoder"
			config.DecoderName = decoderName

			c.Specify("honors time ticker to flush", func() {
				// Spin up a http server
				server, err := plugins_ts.NewOneHttpServer(json_post, "localhost", 9876)
				c.Expect(err, gs.IsNil)
				go server.Start("/")
				time.Sleep(10 * time.Millisecond)

				config.Url = "http://localhost:9876/"
				tickChan := make(chan time.Time)

				ith.MockInputRunner.EXPECT().LogMessage(gomock.Any()).Times(2)

				ith.MockHelper.EXPECT().PipelineConfig().Return(pConfig)
				ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply)
				ith.MockInputRunner.EXPECT().Ticker().Return(tickChan)

				mockDecoderRunner := pipelinemock.NewMockDecoderRunner(ctrl)

				// Stub out the DecoderRunner input channel so that we can
				// inspect bytes later on
				dRunnerInChan := make(chan *PipelinePack, 1)
				mockDecoderRunner.EXPECT().InChan().Return(dRunnerInChan)

				ith.MockInputRunner.EXPECT().Name().Return("HttpInput")
				ith.MockHelper.EXPECT().DecoderRunner(decoderName, "HttpInput-TestDecoder").Return(mockDecoderRunner, true)

				err = httpInput.Init(config)
				c.Assume(err, gs.IsNil)
				startInput()

				tickChan <- time.Now()

				// We need for the pipeline to finish up
				time.Sleep(50 * time.Millisecond)
			})

			c.Specify("supports configuring HTTP Basic Authentication", func() {
				// Spin up a http server which expects username "user" and password "password"
				server, err := plugins_ts.NewHttpBasicAuthServer("user", "password", "localhost", 9875)
				c.Expect(err, gs.IsNil)
				go server.Start("/BasicAuthTest")
//.........这里部分代码省略.........

func ProcessInputSpec(c gs.Context) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	pConfig := NewPipelineConfig(nil)
	ith := new(plugins_ts.InputTestHelper)
	ith.Msg = pipeline_ts.GetTestMessage()

	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)
	ith.MockDeliverer = pipelinemock.NewMockDeliverer(ctrl)
	ith.MockSplitterRunner = pipelinemock.NewMockSplitterRunner(ctrl)
	ith.Pack = NewPipelinePack(pConfig.InputRecycleChan())

	c.Specify("A ProcessInput", func() {
		pInput := ProcessInput{}

		config := pInput.ConfigStruct().(*ProcessInputConfig)
		config.Command = make(map[string]cmdConfig)

		ith.MockHelper.EXPECT().Hostname().Return(pConfig.Hostname())

		tickChan := make(chan time.Time)
		ith.MockInputRunner.EXPECT().Ticker().Return(tickChan)

		errChan := make(chan error)

		ith.MockSplitterRunner.EXPECT().UseMsgBytes().Return(false)

		decChan := make(chan func(*PipelinePack), 1)
		setDecCall := ith.MockSplitterRunner.EXPECT().SetPackDecorator(gomock.Any())
		setDecCall.Do(func(dec func(*PipelinePack)) {
			decChan <- dec
		})

		bytesChan := make(chan []byte, 1)
		splitCall := ith.MockSplitterRunner.EXPECT().SplitStream(gomock.Any(),
			ith.MockDeliverer).Return(nil)
		splitCall.Do(func(r io.Reader, del Deliverer) {
			bytes, err := ioutil.ReadAll(r)
			c.Assume(err, gs.IsNil)
			bytesChan <- bytes
		})

		ith.MockDeliverer.EXPECT().Done()

		c.Specify("using stdout", func() {
			ith.MockInputRunner.EXPECT().NewDeliverer("stdout").Return(ith.MockDeliverer)
			ith.MockInputRunner.EXPECT().NewSplitterRunner("stdout").Return(
				ith.MockSplitterRunner)

			c.Specify("reads a message from ProcessInput", func() {
				pInput.SetName("SimpleTest")

				// Note that no working directory is explicitly specified.
				config.Command["0"] = cmdConfig{
					Bin:  PROCESSINPUT_TEST1_CMD,
					Args: PROCESSINPUT_TEST1_CMD_ARGS,
				}
				err := pInput.Init(config)
				c.Assume(err, gs.IsNil)

				go func() {
					errChan <- pInput.Run(ith.MockInputRunner, ith.MockHelper)
				}()
				tickChan <- time.Now()

				actual := <-bytesChan
				c.Expect(string(actual), gs.Equals, PROCESSINPUT_TEST1_OUTPUT+"\n")

				dec := <-decChan
				dec(ith.Pack)
				fPInputName := ith.Pack.Message.FindFirstField("ProcessInputName")
				c.Expect(fPInputName.ValueString[0], gs.Equals, "SimpleTest.stdout")

				pInput.Stop()
				err = <-errChan
				c.Expect(err, gs.IsNil)
			})

			c.Specify("can pipe multiple commands together", func() {
				pInput.SetName("PipedCmd")

				// Note that no working directory is explicitly specified.
				config.Command["0"] = cmdConfig{
					Bin:  PROCESSINPUT_PIPE_CMD1,
					Args: PROCESSINPUT_PIPE_CMD1_ARGS,
				}
				config.Command["1"] = cmdConfig{
					Bin:  PROCESSINPUT_PIPE_CMD2,
					Args: PROCESSINPUT_PIPE_CMD2