func DecodersSpec(c gospec.Context) {
	msg := getTestMessage()

	c.Specify("A JsonDecoder", func() {
		var fmtString = `{"type":"%s","timestamp":%s,"logger":"%s","severity":%d,"payload":"%s","fields":%s,"env_version":"%s","metlog_pid":%d,"metlog_hostname":"%s"}`
		timestampJson, err := json.Marshal(msg.Timestamp)
		fieldsJson, err := json.Marshal(msg.Fields)
		c.Assume(err, gs.IsNil)
		jsonString := fmt.Sprintf(fmtString, msg.Type,
			timestampJson, msg.Logger, msg.Severity, msg.Payload,
			fieldsJson, msg.Env_version, msg.Pid, msg.Hostname)

		pipelinePack := getTestPipelinePack()
		pipelinePack.MsgBytes = []byte(jsonString)
		jsonDecoder := new(JsonDecoder)

		c.Specify("can decode a JSON message", func() {
			err := jsonDecoder.Decode(pipelinePack)
			c.Expect(pipelinePack.Message, gs.Equals, msg)
			c.Expect(err, gs.IsNil)
		})

		c.Specify("returns `fields` as a map", func() {
			jsonDecoder.Decode(pipelinePack)
			c.Expect(pipelinePack.Message.Fields["foo"], gs.Equals, "bar")
		})

		c.Specify("returns an error for bogus JSON", func() {
			badJson := fmt.Sprint("{{", jsonString)
			pipelinePack.MsgBytes = []byte(badJson)
			err := jsonDecoder.Decode(pipelinePack)
			c.Expect(err, gs.Not(gs.IsNil))
			c.Expect(pipelinePack.Message.Timestamp.IsZero(), gs.IsTrue)
		})
	})

	c.Specify("A MsgPackDecoder", func() {
		msg := getTestMessage()
		encoded, err := msgpack.Marshal(msg)
		c.Assume(err, gs.IsNil)

		decoder := new(MsgPackDecoder)
		decoder.Init(nil)
		pack := getTestPipelinePack()

		c.Specify("decodes a msgpack message", func() {
			pack.MsgBytes = encoded
			err := decoder.Decode(pack)
			c.Expect(err, gs.IsNil)
			c.Expect(pack.Message, gs.Equals, msg)
		})

		c.Specify("returns an error for bunk encoding", func() {
			bunk := append([]byte{0, 0, 0}, encoded...)
			pack.MsgBytes = bunk
			err := decoder.Decode(pack)
			c.Expect(err, gs.Not(gs.IsNil))
		})
	})
}

func ProtobufDecoderSpec(c gospec.Context) {
	t := &ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	msg := ts.GetTestMessage()
	config := NewPipelineConfig(nil) // Initializes globals.

	c.Specify("A ProtobufDecoder", func() {
		encoded, err := proto.Marshal(msg)
		c.Assume(err, gs.IsNil)
		pack := NewPipelinePack(config.inputRecycleChan)
		decoder := new(ProtobufDecoder)
		decoder.sampleDenominator = 1000 // Since we don't call decoder.Init().

		c.Specify("decodes a protobuf message", func() {
			pack.MsgBytes = encoded
			_, err := decoder.Decode(pack)
			c.Expect(err, gs.IsNil)
			c.Expect(pack.Message, gs.Equals, msg)
			v, ok := pack.Message.GetFieldValue("foo")
			c.Expect(ok, gs.IsTrue)
			c.Expect(v, gs.Equals, "bar")
		})

		c.Specify("returns an error for bunk encoding", func() {
			bunk := append([]byte{0, 0, 0}, encoded...)
			pack.MsgBytes = bunk
			_, err := decoder.Decode(pack)
			c.Expect(err, gs.Not(gs.IsNil))
		})
	})
}

func PrimitiveDecodeStrictSpec(c gs.Context) {
	var md MetaData
	var err error

	var tomlBlob = `
ranking = ["Springsteen", "J Geils"]

[bands.Springsteen]
type = "ignore_this"
started = 1973
albums = ["Greetings", "WIESS", "Born to Run", "Darkness"]
not_albums = ["Greetings", "WIESS", "Born to Run", "Darkness"]

[bands.J Geils]
started = 1970
albums = ["The J. Geils Band", "Full House", "Blow Your Face Out"]
`

	type band struct {
		Started int
		Albums  []string
	}

	type classics struct {
		Ranking []string
		Bands   map[string]Primitive
	}

	// Do the initial decode. Reflection is delayed on Primitive values.
	var music classics
	md, err = Decode(tomlBlob, &music)
	c.Assume(err, gs.IsNil)

	// MetaData still includes information on Primitive values.
	c.Assume(md.IsDefined("bands", "Springsteen"), gs.IsTrue)

	ignore_type := map[string]interface{}{"type": true}
	// Decode primitive data into Go values.
	for _, artist := range music.Ranking {
		// A band is a primitive value, so we need to decode it to get a
		// real `band` value.
		primValue := music.Bands[artist]

		var aBand band

		err = PrimitiveDecodeStrict(primValue, &aBand, ignore_type)
		if artist == "Springsteen" {
			c.Assume(err, gs.Not(gs.IsNil))
			c.Expect(err.Error(), gs.Equals, "Configuration contains key [not_albums] which doesn't exist in struct")
			c.Assume(1973, gs.Equals, aBand.Started)
		} else {
			c.Expect(err, gs.IsNil)
			c.Assume(1970, gs.Equals, aBand.Started)
		}

	}
}

func UdpInputSpecFailure(c gs.Context) {
	udpInput := UdpInput{}
	err := udpInput.Init(&UdpInputConfig{Net: "tcp", Address: "localhost:55565",
		Decoder:    "ProtobufDecoder",
		ParserType: "message.proto"})
	c.Assume(err, gs.Not(gs.IsNil))
	c.Assume(err.Error(), gs.Equals, "ResolveUDPAddr failed: unknown network tcp\n")

}

func TcpInputSpecFailure(c gs.Context) {
	tcpInput := TcpInput{}
	err := tcpInput.Init(&TcpInputConfig{Net: "udp", Address: "localhost:55565",
		Decoder:    "ProtobufDecoder",
		ParserType: "message.proto"})
	c.Assume(err, gs.Not(gs.IsNil))
	c.Assume(err.Error(), gs.Equals, "ListenTCP failed: unknown network udp\n")

}

func UdpInputSpecFailure(c gs.Context) {
	udpInput := UdpInput{}
	err := udpInput.Init(&UdpInputConfig{
		Net:     "tcp",
		Address: "localhost:55565",
	})
	c.Assume(err, gs.Not(gs.IsNil))
	c.Assume(err.Error(), gs.Equals, "ResolveUDPAddr failed: unknown network tcp\n")

}

func TcpInputSpecFailure(c gs.Context) {
	tcpInput := TcpInput{}
	err := tcpInput.Init(&TcpInputConfig{
		Net:     "udp",
		Address: "localhost:55565",
	})
	c.Assume(err, gs.Not(gs.IsNil))
	c.Assume(err.Error(), gs.Equals, "ResolveTCPAddress failed: unknown network udp\n")

}

func MessageEqualsSpec(c gospec.Context) {
	msg0 := getTestMessage()
	msg1Real := *msg0
	msg1 := &msg1Real

	c.Specify("Messages are equal", func() {
		c.Expect(msg0, gs.Equals, msg1)
	})

	c.Specify("Messages w/ diff int values are not equal", func() {
		msg1.Severity--
		c.Expect(msg0, gs.Not(gs.Equals), msg1)
	})

	c.Specify("Messages w/ diff string values are not equal", func() {
		msg1.Payload = "Something completely different"
		c.Expect(msg0, gs.Not(gs.Equals), msg1)
	})

	c.Specify("Messages w/ diff maps are not equal", func() {
		msg1.Fields = map[string]interface{}{"sna": "foo"}
		c.Expect(msg0, gs.Not(gs.Equals), msg1)
	})
}

func DecodeSpec(c gs.Context) {
	var val simple

	md, err := Decode(testSimple, &val)
	c.Assume(err, gs.IsNil)
	c.Assume(md.IsDefined("Annoying", "Cats", "plato"), gs.IsTrue)
	c.Assume(md.IsDefined("Cats", "Stinky"), gs.IsFalse)
	var colors = [][]string{[]string{"red", "green", "blue"},
		[]string{"cyan", "magenta", "yellow", "black"}}
	for ridx, row := range colors {
		for cidx, _ := range row {
			c.Assume(val.Colors[ridx][cidx], gs.Equals, colors[ridx][cidx])
		}
	}
	c.Assume(val, gs.Not(gs.IsNil))
}

func DashboardOutputSpec(c gs.Context) {
	t := new(ts.SimpleT)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	c.Specify("A FileOutput", func() {
		dashboardOutput := new(DashboardOutput)

		config := dashboardOutput.ConfigStruct().(*DashboardOutputConfig)
		c.Specify("Init halts if basedirectory is not writable", func() {
			tmpdir := path.Join(os.TempDir(), "tmpdir")
			err := os.MkdirAll(tmpdir, 0400)
			c.Assume(err, gs.IsNil)
			config.WorkingDirectory = tmpdir
			err = dashboardOutput.Init(config)
			c.Assume(err, gs.Not(gs.IsNil))
		})
	})
}

func DashboardOutputSpec(c gs.Context) {
	t := new(pipeline_ts.SimpleT)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	NewPipelineConfig(nil) // Needed for side effect of setting up Globals :P

	if runtime.GOOS != "windows" {
		c.Specify("A DashboardOutput", func() {
			dashboardOutput := new(DashboardOutput)

			config := dashboardOutput.ConfigStruct().(*DashboardOutputConfig)
			c.Specify("Init halts if basedirectory is not writable", func() {
				tmpdir := filepath.Join(os.TempDir(), "tmpdir")
				err := os.MkdirAll(tmpdir, 0400)
				c.Assume(err, gs.IsNil)
				config.WorkingDirectory = tmpdir
				err = dashboardOutput.Init(config)
				c.Assume(err, gs.Not(gs.IsNil))
			})
		})
	}
}

func JsonPathSpec(c gs.Context) {
	c.Specify("JsonPath can read data", func() {
		var s = `{
	"foo": {
		"bar": [
			{
				"baz": "こんにちわ世界",
				"noo": "aaa"
			},
			{
				"maz": "123",
				"moo": 256,
				"muux": 2.10
			}
		],
		"boo": {
			"bag": true,
			"bug": false
		}
	}
}
`
		var err error
		var json_path *JsonPath
		var result interface{}

		json_path = new(JsonPath)
		err = json_path.SetJsonText(s)
		c.Expect(err, gs.IsNil)

		result, err = json_path.Find("$.foo.bar[0].baz")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "こんにちわ世界")

		result, err = json_path.Find("$.foo.bar[0].noo")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "aaa")

		result, err = json_path.Find("$.foo.bar[1].maz")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "123")

		result, err = json_path.Find("$.foo.bar[1].moo")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "256")

		result, err = json_path.Find("$.foo.bar[1].muux")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "2.10")

		result, err = json_path.Find("$.foo.boo.bag")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "true")

		result, err = json_path.Find("$.foo.boo.bug")
		c.Expect(err, gs.IsNil)
		c.Expect(result, gs.Equals, "false")

		result, err = json_path.Find("$.foo.bar[99].baz")
		c.Expect(err, gs.Not(gs.IsNil))

		result, err = json_path.Find("$.badpath")
		c.Expect(err, gs.Not(gs.IsNil))

		result, err = json_path.Find("badpath")
		c.Expect(err, gs.Not(gs.IsNil))

		result, err = json_path.Find("$.foo.bar.3428")
		c.Expect(err, gs.Not(gs.IsNil))

		expected_data := `[{"baz":"こんにちわ世界","noo":"aaa"},{"maz":"123","moo":256,"muux":2.10}]`
		result_data, err := json_path.Find("$.foo.bar")
		c.Expect(err, gs.IsNil)
		c.Expect(result_data, gs.Equals, expected_data)

	})

	c.Specify("JsonPath doesn't crash on nil data", func() {
		var err error
		var json_path *JsonPath

		json_path = new(JsonPath)
		err = json_path.SetJsonText("")
		c.Expect(err, gs.Not(gs.IsNil))

		// Searches should return an error
		result, err := json_path.Find("$.foo.bar.3428")
		c.Expect(err, gs.Not(gs.IsNil))
		c.Expect(err.Error(), gs.Equals, "JSON data is nil")
		c.Expect(result, gs.Equals, "")
	})

	c.Specify("JsonPath handles arrays at top level", func() {
		var err error
		var json_path *JsonPath

		json_path = new(JsonPath)
		err = json_path.SetJsonText(`["foo"]`)
		c.Expect(err, gs.IsNil)

//.........这里部分代码省略.........

func DashboardOutputSpec(c gs.Context) {
	t := new(pipeline_ts.SimpleT)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	pConfig := pipeline.NewPipelineConfig(nil)

	dashboardOutput := new(DashboardOutput)
	dashboardOutput.pConfig = pConfig

	oth := plugins_ts.NewOutputTestHelper(ctrl)
	oth.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	oth.MockOutputRunner = pipelinemock.NewMockOutputRunner(ctrl)

	errChan := make(chan error, 1)

	startOutput := func() {
		go func() {
			errChan <- dashboardOutput.Run(oth.MockOutputRunner, oth.MockHelper)
		}()
	}

	if runtime.GOOS != "windows" {
		c.Specify("A DashboardOutput", func() {

			tmpdir, err := ioutil.TempDir("", "dashboard_output_test")
			c.Assume(err, gs.IsNil)
			config := dashboardOutput.ConfigStruct().(*DashboardOutputConfig)
			config.WorkingDirectory = tmpdir

			c.Specify("Init halts if basedirectory is not writable", func() {
				err := os.MkdirAll(tmpdir, 0400)
				c.Assume(err, gs.IsNil)
				defer os.RemoveAll(tmpdir)
				err = dashboardOutput.Init(config)
				c.Assume(err, gs.Not(gs.IsNil))
			})

			c.Specify("that is running", func() {
				startedChan := make(chan bool, 1)
				defer close(startedChan)
				ts := httptest.NewUnstartedServer(nil)

				dashboardOutput.starterFunc = func(hli *DashboardOutput) error {
					ts.Start()
					startedChan <- true
					return nil
				}

				ticker := make(chan time.Time)
				inChan := make(chan *pipeline.PipelinePack, 1)
				recycleChan := make(chan *pipeline.PipelinePack, 1)
				pack := pipeline.NewPipelinePack(recycleChan)
				pack.Message = pipeline_ts.GetTestMessage()

				oth.MockOutputRunner.EXPECT().InChan().Return(inChan)
				oth.MockOutputRunner.EXPECT().Ticker().Return(ticker)

				err := os.MkdirAll(tmpdir, 0700)
				c.Assume(err, gs.IsNil)
				defer os.RemoveAll(tmpdir)

				dashboardOutput.handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
					// Noop
				})

				c.Specify("sets custom http headers", func() {
					config.Headers = http.Header{
						"One":  []string{"two", "three"},
						"Four": []string{"five", "six", "seven"},
					}
					err = dashboardOutput.Init(config)
					c.Assume(err, gs.IsNil)
					ts.Config = dashboardOutput.server

					startOutput()

					inChan <- pack
					<-startedChan
					resp, err := http.Get(ts.URL)
					c.Assume(err, gs.IsNil)
					resp.Body.Close()
					c.Assume(resp.StatusCode, gs.Equals, 200)

					// Verify headers are there
					eq := reflect.DeepEqual(resp.Header["One"], config.Headers["One"])
					c.Expect(eq, gs.IsTrue)
					eq = reflect.DeepEqual(resp.Header["Four"], config.Headers["Four"])
					c.Expect(eq, gs.IsTrue)
				})

				close(inChan)
				c.Expect(<-errChan, gs.IsNil)

				ts.Close()
			})
		})
	}
}

//.........这里部分代码省略.........
	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
					// to be processed.
					runtime.Gosched()
				}
				lsInput.Stop()
				c.Expect(timed, gs.Equals, false)
				c.Expect(<-runOutChan, gs.Equals, nil)
			})
		})

		c.Specify("with a translation map", func() {
			lsiConfig.Translation = make(ls.SubmatchTranslationMap)
			lsiConfig.Translation["Seq"] = make(ls.MatchTranslationMap)

			c.Specify("allows len 1 translation map for 'missing'", func() {
				lsiConfig.Translation["Seq"]["missing"] = 9999
				err := lsInput.Init(lsiConfig)
				c.Expect(err, gs.IsNil)
			})

			c.Specify("doesn't allow len 1 map for other keys", func() {
				lsiConfig.Translation["Seq"]["missin"] = 9999
				err := lsInput.Init(lsiConfig)
				c.Expect(err, gs.Not(gs.IsNil))
				c.Expect(err.Error(), gs.Equals,
					"A translation map with one entry ('Seq') must be specifying a "+
						"'missing' key.")
			})
		})
	})
}

//.........这里部分代码省略.........

			c.Specify("when specified and timestamp ignored", func() {
				fileOutput.prefix_ts = true
				err := fileOutput.handleMessage(pack, &outData)
				c.Expect(err, gs.IsNil)
				b := []byte{30, 2, 8, uint8(proto.Size(pack.Message)), 31, 10, 16} // sanity check the header and the start of the protocol buffer
				c.Expect(bytes.Equal(b, outData[:len(b)]), gs.IsTrue)
			})
		})

		c.Specify("processes incoming messages", func() {
			err := fileOutput.Init(config)
			defer os.Remove(tmpFilePath)
			c.Assume(err, gs.IsNil)
			// Save for comparison.
			payload := fmt.Sprintf("%s\n", pack.Message.GetPayload())

			oth.MockOutputRunner.EXPECT().InChan().Return(inChan)
			wg.Add(1)
			go fileOutput.receiver(oth.MockOutputRunner, &wg)
			inChan <- pack
			close(inChan)
			outBatch := <-fileOutput.batchChan
			wg.Wait()
			c.Expect(string(outBatch), gs.Equals, payload)
		})

		c.Specify("Init halts if basedirectory is not writable", func() {
			tmpdir := filepath.Join(os.TempDir(), "tmpdir")
			err := os.MkdirAll(tmpdir, 0400)
			c.Assume(err, gs.IsNil)
			config.Path = tmpdir
			err = fileOutput.Init(config)
			c.Assume(err, gs.Not(gs.IsNil))
		})

		c.Specify("commits to a file", func() {
			outStr := "Write me out to the log file"
			outBytes := []byte(outStr)

			c.Specify("with default settings", func() {
				err := fileOutput.Init(config)
				defer os.Remove(tmpFilePath)
				c.Assume(err, gs.IsNil)

				// Start committer loop
				wg.Add(1)
				go fileOutput.committer(oth.MockOutputRunner, &wg)

				// Feed and close the batchChan
				go func() {
					fileOutput.batchChan <- outBytes
					_ = <-fileOutput.backChan // clear backChan to prevent blocking
					close(fileOutput.batchChan)
				}()

				wg.Wait()
				// Wait for the file close operation to happen.
				//for ; err == nil; _, err = fileOutput.file.Stat() {
				//}

				tmpFile, err := os.Open(tmpFilePath)
				defer tmpFile.Close()
				c.Assume(err, gs.IsNil)
				contents, err := ioutil.ReadAll(tmpFile)
				c.Assume(err, gs.IsNil)

//.........这里部分代码省略.........

			c.Specify("when specified and timestamp ignored", func() {
				fileOutput.prefix_ts = true
				err := fileOutput.handleMessage(pack, &outData)
				c.Expect(err, gs.IsNil)
				b := []byte{30, 2, 8, uint8(proto.Size(pack.Message)), 31, 10, 16} // sanity check the header and the start of the protocol buffer
				c.Expect(bytes.Equal(b, outData[:len(b)]), gs.IsTrue)
			})
		})

		c.Specify("processes incoming messages", func() {
			err := fileOutput.Init(config)
			defer os.Remove(tmpFilePath)
			c.Assume(err, gs.IsNil)
			// Save for comparison.
			payload := fmt.Sprintf("%s\n", pack.Message.GetPayload())

			oth.MockOutputRunner.EXPECT().InChan().Return(inChan)
			wg.Add(1)
			go fileOutput.receiver(oth.MockOutputRunner, &wg)
			inChan <- pack
			close(inChan)
			outBatch := <-fileOutput.batchChan
			wg.Wait()
			c.Expect(string(outBatch), gs.Equals, payload)
		})

		c.Specify("Init halts if basedirectory is not writable", func() {
			tmpdir := filepath.Join(os.TempDir(), "tmpdir")
			err := os.MkdirAll(tmpdir, 0400)
			c.Assume(err, gs.IsNil)
			config.Path = tmpdir
			err = fileOutput.Init(config)
			c.Assume(err, gs.Not(gs.IsNil))
		})

		c.Specify("commits to a file", func() {
			outStr := "Write me out to the log file"
			outBytes := []byte(outStr)

			c.Specify("with default settings", func() {
				err := fileOutput.Init(config)
				defer os.Remove(tmpFilePath)
				c.Assume(err, gs.IsNil)

				// Start committer loop
				wg.Add(1)
				go fileOutput.committer(oth.MockOutputRunner, &wg)

				// Feed and close the batchChan
				go func() {
					fileOutput.batchChan <- outBytes
					_ = <-fileOutput.backChan // clear backChan to prevent blocking
					close(fileOutput.batchChan)
				}()

				wg.Wait()
				// Wait for the file close operation to happen.
				//for ; err == nil; _, err = fileOutput.file.Stat() {
				//}

				tmpFile, err := os.Open(tmpFilePath)
				defer tmpFile.Close()
				c.Assume(err, gs.IsNil)
				contents, err := ioutil.ReadAll(tmpFile)
				c.Assume(err, gs.IsNil)

//.........这里部分代码省略.........
			"Fields[foo][1] == 'bar'",
			"Fields[foo][0][1] == 'bar'",
			"Fields[bool] == FALSE",
			"Type =~ /Test/",
			"Type !~ /TEST/",
			"Payload =~ /^Payload/",
			"Type == \"te'st\"",
			"Type == 'te\"st'",
			"Fields[int] =~ /999/",
			"Fields[zero] == \"0\"",
			"Fields[string] == 43",
			"Fields[int] == NIL",
			"Fields[int][0][1] == NIL",
			"Fields[missing] != NIL",
			"Type =~ /^te/",
			"Type =~ /st$/",
			"Type !~ /^TE/",
			"Type !~ /ST$/",
			"Logger =~ /./ && Type =~ /^anything/",
		}

		positive := []string{
			"TRUE",
			"(Severity == 7 || Payload == 'Test Payload') && Type == 'TEST'",
			"EnvVersion == \"0.8\"",
			"EnvVersion == '0.8'",
			"EnvVersion != '0.9'",
			"EnvVersion > '0.7'",
			"EnvVersion >= '0.8'",
			"EnvVersion < '0.9'",
			"EnvVersion <= '0.8'",
			"Hostname != ''",
			"Logger == 'GoSpec'",
			"Pid != 0",
			"Severity != 5",
			"Severity < 7",
			"Severity <= 6",
			"Severity == 6",
			"Severity > 5",
			"Severity >= 6",
			"Timestamp > 0",
			"Type != 'test'",
			"Type == 'TEST' && Severity == 6",
			"Type == 'test' && Severity == 7 || Payload == 'Test Payload'",
			"Type == 'TEST'",
			"Type == 'foo' || Type == 'bar' || Type == 'TEST'",
			fmt.Sprintf("Uuid == '%s'", uuidStr),
			"Fields[foo] == 'bar'",
			"Fields[foo][0] == 'bar'",
			"Fields[foo][0][0] == 'bar'",
			"Fields[foo][1] == 'alternate'",
			"Fields[foo][1][0] == 'alternate'",
			"Fields[foo] == 'bar'",
			"Fields[bytes] == 'data'",
			"Fields[int] == 999",
			"Fields[int][0][1] == 1024",
			"Fields[double] == 99.9",
			"Fields[bool] == TRUE",
			"Type =~ /TEST/",
			"Type !~ /bogus/",
			"Type =~ /TEST/ && Payload =~ /Payload/",
			"Fields[foo][1] =~ /alt/",
			"Fields[Payload] =~ /name=\\w+/",
			"Type =~ /(ST)/",
			"Fields[int] != NIL",
			"Fields[int][0][1] != NIL",
			"Fields[int][0][2] == NIL",
			"Fields[missing] == NIL",
			"Type =~ /^TE/",
			"Type =~ /ST$/",
			"Type !~ /^te/",
			"Type !~ /st$/",
		}

		c.Specify("malformed matcher tests", func() {
			for _, v := range malformed {
				_, err := CreateMatcherSpecification(v)
				c.Expect(err, gs.Not(gs.IsNil))
			}
		})

		c.Specify("negative matcher tests", func() {
			for _, v := range negative {
				ms, err := CreateMatcherSpecification(v)
				c.Expect(err, gs.IsNil)
				match := ms.Match(msg)
				c.Expect(match, gs.IsFalse)
			}
		})

		c.Specify("positive matcher tests", func() {
			for _, v := range positive {
				ms, err := CreateMatcherSpecification(v)
				c.Expect(err, gs.IsNil)
				match := ms.Match(msg)
				c.Expect(match, gs.IsTrue)
			}
		})
	})
}

//.........这里部分代码省略.........
	fRunner.matcher.inChan = make(chan *PipelinePack, chanSize)
	leakCount := 10
	fRunner.SetLeakCount(leakCount)

	iName := "stat_accum"
	input := new(StatAccumInput)
	iRunner := NewInputRunner(iName, input, nil, false)

	c.Specify("`PopulateReportMsg`", func() {
		msg := ts.GetTestMessage()

		c.Specify("w/ a filter", func() {
			err := PopulateReportMsg(fRunner, msg)
			c.Assume(err, gs.IsNil)

			c.Specify("invokes `ReportMsg` on the filter", func() {
				checkForFields(c, msg)
			})

			c.Specify("adds the channel data", func() {
				c.Expect(hasChannelData(msg), gs.IsTrue)
			})

			c.Specify("has its leak count set properly", func() {
				leakVal, ok := msg.GetFieldValue("LeakCount")
				c.Assume(ok, gs.IsTrue)
				i, ok := leakVal.(int64)
				c.Assume(ok, gs.IsTrue)
				c.Expect(int(i), gs.Equals, leakCount)
			})
		})

		c.Specify("w/ an input", func() {
			err := PopulateReportMsg(iRunner, msg)
			c.Assume(err, gs.IsNil)

			c.Specify("invokes `ReportMsg` on the input", func() {
				checkForFields(c, msg)
			})

			c.Specify("doesn't add any channel data", func() {
				capVal, ok := msg.GetFieldValue("InChanCapacity")
				c.Expect(capVal, gs.IsNil)
				c.Expect(ok, gs.IsFalse)
				lenVal, ok := msg.GetFieldValue("InChanLength")
				c.Expect(lenVal, gs.IsNil)
				c.Expect(ok, gs.IsFalse)
			})
		})
	})

	c.Specify("PipelineConfig", func() {
		pc := NewPipelineConfig(nil)
		// Initialize all of the PipelinePacks that we'll need
		pc.reportRecycleChan <- NewPipelinePack(pc.reportRecycleChan)

		pc.FilterRunners = map[string]FilterRunner{fName: fRunner}
		pc.InputRunners = map[string]InputRunner{iName: iRunner}

		c.Specify("returns full set of accurate reports", func() {
			reportChan := make(chan *PipelinePack)
			go pc.reports(reportChan)

			reports := make(map[string]*PipelinePack)
			for r := range reportChan {
				iName, ok := r.Message.GetFieldValue("name")
				c.Expect(ok, gs.IsTrue)
				name, ok := iName.(string)
				c.Expect(ok, gs.IsTrue)
				c.Expect(name, gs.Not(gs.Equals), "MISSING")
				reports[name] = r
				pc.reportRecycleChan <- NewPipelinePack(pc.reportRecycleChan)
			}
			fReport := reports[fName]
			c.Expect(fReport, gs.Not(gs.IsNil))
			checkForFields(c, fReport.Message)
			c.Expect(hasChannelData(fReport.Message), gs.IsTrue)

			iReport := reports[iName]
			c.Expect(iReport, gs.Not(gs.IsNil))
			checkForFields(c, iReport.Message)

			recycleReport := reports["inputRecycleChan"]
			c.Expect(recycleReport, gs.Not(gs.IsNil))
			capVal, ok := recycleReport.Message.GetFieldValue("InChanCapacity")
			c.Expect(ok, gs.IsTrue)
			c.Expect(capVal.(int64), gs.Equals, int64(pConfig.Globals.PoolSize))

			injectReport := reports["injectRecycleChan"]
			c.Expect(injectReport, gs.Not(gs.IsNil))
			capVal, ok = injectReport.Message.GetFieldValue("InChanCapacity")
			c.Expect(ok, gs.IsTrue)
			c.Expect(capVal.(int64), gs.Equals, int64(pConfig.Globals.PoolSize))

			routerReport := reports["Router"]
			c.Expect(routerReport, gs.Not(gs.IsNil))
			c.Expect(hasChannelData(routerReport.Message), gs.IsTrue)
		})
	})
}

//.........这里部分代码省略.........
		})
		c.Specify("emits gauge with correct prefixes", func() {
			prepareSendingStats()
			sendGauge("sample.gauge", 1, 2)
			sendGauge("sample2.gauge", 1, 2, 3, 4, 5)
			msg := finalizeSendingStats()
			validateValueAtKey(msg, "sample.gauge", int64(2))
			validateValueAtKey(msg, "sample2.gauge", int64(5))
		})

		c.Specify("emits correct statsd.numStats count", func() {
			prepareSendingStats()
			sendGauge("sample.gauge", 1, 2)
			sendGauge("sample2.gauge", 1, 2)
			sendCounter("sample.cnt", 1, 2, 3, 4, 5)
			sendCounter("sample2.cnt", 159, 951)
			sendTimer("sample.timer", 10, 10, 20, 20)
			sendTimer("sample2.timer", 10, 20)
			msg := finalizeSendingStats()
			validateValueAtKey(msg, "statsd.numStats", int64(6))
		})
	})

	c.Specify("A StatAccumInput using Legacy namespaces", func() {
		statAccumInput := StatAccumInput{}
		config := statAccumInput.ConfigStruct().(*StatAccumInputConfig)
		config.LegacyNamespaces = true

		tickChan := make(chan time.Time)

		c.Specify("must emit data in payload and/or message fields", func() {
			config.EmitInPayload = false
			err := statAccumInput.Init(config)
			c.Expect(err, gs.Not(gs.IsNil))
			expected := "One of either `EmitInPayload` or `EmitInFields` must be set to true."
			c.Expect(err.Error(), gs.Equals, expected)
		})

		c.Specify("that actually emits a message", func() {
			statName := "sample.stat"
			statVal := int64(303)
			testStat := Stat{statName, strconv.Itoa(int(statVal)), "c", float32(1)}

			validateMsgFields := func(msg *message.Message) {
				c.Expect(len(msg.Fields), gs.Equals, 4)

				// timestamp
				_, ok := msg.GetFieldValue("timestamp")
				c.Expect(ok, gs.IsTrue)

				var tmp interface{}
				var intTmp int64

				// stats.sample.stat
				tmp, ok = msg.GetFieldValue("stats." + statName)
				c.Expect(ok, gs.IsTrue)
				intTmp, ok = tmp.(int64)
				c.Expect(ok, gs.IsTrue)
				c.Expect(intTmp, gs.Equals, int64(30))

				// stats_counts.sample.stat
				tmp, ok = msg.GetFieldValue("stats_counts." + statName)
				c.Expect(ok, gs.IsTrue)
				intTmp, ok = tmp.(int64)
				c.Expect(ok, gs.IsTrue)
				c.Expect(intTmp, gs.Equals, statVal)

//.........这里部分代码省略.........
		neterr := pipeline_ts.NewMockError(ctrl)
		neterr.EXPECT().Temporary().Return(false)
		acceptCall := mockListener.EXPECT().Accept().Return(mockConnection, nil)
		acceptCall.Do(func() {
			acceptCall = mockListener.EXPECT().Accept()
			acceptCall.Return(nil, neterr)
		})

		mockDecoderRunner := ith.Decoder.(*pipelinemock.MockDecoderRunner)
		mockDecoderRunner.EXPECT().InChan().Return(ith.DecodeChan)
		ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply)
		ith.MockInputRunner.EXPECT().Name().Return("logger")
		enccall := ith.MockHelper.EXPECT().DecoderRunner("TokenDecoder", "TcpInput-123-TokenDecoder").AnyTimes()
		enccall.Return(ith.Decoder, true)
		ith.MockHelper.EXPECT().StopDecoderRunner(ith.Decoder)

		c.Specify("reads a message from its connection", func() {
			readCall.Return(len(mbytes), nil)
			readCall.Do(getPayloadText(mbytes))
			go tcpInput.Run(ith.MockInputRunner, ith.MockHelper)
			defer func() {
				mockListener.EXPECT().Close()
				tcpInput.Stop()
				tcpInput.wg.Wait()
			}()
			ith.PackSupply <- ith.Pack
			packRef := <-ith.DecodeChan
			c.Expect(ith.Pack, gs.Equals, packRef)
			c.Expect(ith.Pack.Message.GetPayload(), gs.Equals, string(mbytes))
			c.Expect(ith.Pack.Message.GetLogger(), gs.Equals, "logger")
			c.Expect(ith.Pack.Message.GetHostname(), gs.Equals, "123")
		})
	})

	c.Specify("A TcpInput using TLS", func() {
		tcpInput := TcpInput{}
		config := &TcpInputConfig{
			Net:        "tcp",
			Address:    ith.AddrStr,
			ParserType: "token",
			UseTls:     true,
		}

		c.Specify("fails to init w/ missing key or cert file", func() {
			config.Tls = TlsConfig{}
			err := tcpInput.Init(config)
			c.Expect(err, gs.Not(gs.IsNil))
		})

		c.Specify("accepts TLS client connections", func() {
			ith.MockInputRunner.EXPECT().Name().Return("TcpInput")
			config.Tls = TlsConfig{
				CertFile: "./testsupport/cert.pem",
				KeyFile:  "./testsupport/key.pem",
			}
			err := tcpInput.Init(config)
			c.Expect(err, gs.IsNil)
			go tcpInput.Run(ith.MockInputRunner, ith.MockHelper)
			defer func() {
				tcpInput.Stop()
				tcpInput.wg.Wait()
			}()

			clientConfig := new(tls.Config)
			clientConfig.InsecureSkipVerify = true
			conn, err := tls.Dial("tcp", ith.AddrStr, clientConfig)
			c.Expect(err, gs.IsNil)
			defer conn.Close()
			conn.SetWriteDeadline(time.Now().Add(time.Duration(10000)))
			n, err := conn.Write([]byte("This is a test."))
			c.Expect(err, gs.IsNil)
			c.Expect(n, gs.Equals, len("This is a test."))
		})

		c.Specify("doesn't accept connections below specified min TLS version", func() {
			ith.MockInputRunner.EXPECT().Name().Return("TcpInput")
			config.Tls = TlsConfig{
				CertFile:   "./testsupport/cert.pem",
				KeyFile:    "./testsupport/key.pem",
				MinVersion: "TLS12",
			}
			err := tcpInput.Init(config)
			c.Expect(err, gs.IsNil)
			go tcpInput.Run(ith.MockInputRunner, ith.MockHelper)
			defer func() {
				tcpInput.Stop()
				tcpInput.wg.Wait()
				time.Sleep(time.Duration(1000))
			}()

			clientConfig := &tls.Config{
				InsecureSkipVerify: true,
				MaxVersion:         tls.VersionTLS11,
			}
			conn, err := tls.Dial("tcp", ith.AddrStr, clientConfig)
			c.Expect(conn, gs.IsNil)
			c.Expect(err, gs.Not(gs.IsNil))
		})
	})
}