func InsensitiveDecodeSpec(c gs.Context) {
	tme, err := time.Parse(time.RFC3339, time.RFC3339[:len(time.RFC3339)-5])
	if err != nil {
		panic(err)
	}
	expected := Insensitive{
		TopString: "string",
		TopInt:    1,
		TopFloat:  1.1,
		TopBool:   true,
		TopDate:   tme,
		TopArray:  []string{"array"},
		MatcH:     "i should be in MatcH only",
		Match:     "i should be in Match only",
		Field:     "neat", // encoding/json would store "messy" here
		Once:      "just once",
		OncE:      "just once", // wait, what?
		Nest: InsensitiveNest{
			Ed: InsensitiveEd{NestedString: "another string"},
		},
	}
	var got Insensitive
	_, err = Decode(caseToml, &got)
	c.Assume(err, gs.IsNil)
	c.Assume(reflect.DeepEqual(expected, got), gs.IsTrue)
}

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 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",
		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 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 GeoIpDecoderSpec(c gs.Context) {
	t := &ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	pConfig := NewPipelineConfig(nil)
	pConfig.Globals.ShareDir = "/foo/bar/baz"

	c.Specify("A GeoIpDecoder", func() {
		decoder := new(GeoIpDecoder)
		decoder.SetPipelineConfig(pConfig)
		rec := new(geoip.GeoIPRecord)
		conf := decoder.ConfigStruct().(*GeoIpDecoderConfig)

		c.Expect(conf.DatabaseFile, gs.Equals, "/foo/bar/baz/GeoLiteCity.dat")

		supply := make(chan *PipelinePack, 1)
		pack := NewPipelinePack(supply)

		nf, _ := message.NewField("remote_host", "74.125.142.147", "")
		pack.Message.AddField(nf)

		decoder.SourceIpField = "remote_host"
		conf.SourceIpField = "remote_host"
		decoder.Init(conf)

		rec.CountryCode = "US"
		rec.CountryCode3 = "USA"
		rec.CountryName = "United States"
		rec.Region = "CA"
		rec.City = "Mountain View"
		rec.PostalCode = "94043"
		rec.Latitude = 37.4192
		rec.Longitude = -122.0574
		rec.AreaCode = 650
		rec.CharSet = 1
		rec.ContinentCode = "NA"

		c.Specify("Test GeoIpDecoder Output", func() {
			buf := decoder.GeoBuff(rec)
			nf, _ = message.NewField("geoip", buf.Bytes(), "")
			pack.Message.AddField(nf)

			b, ok := pack.Message.GetFieldValue("geoip")
			c.Expect(ok, gs.IsTrue)

			c.Expect(string(b.([]byte)), gs.Equals, `{"latitude":37.4192008972168,"longitude":-122.0574035644531,"location":[-122.0574035644531,37.4192008972168],"coordinates":["-122.0574035644531","37.4192008972168"],"countrycode":"US","countrycode3":"USA","countryname":"United States","region":"CA","city":"Mountain View","postalcode":"94043","areacode":650,"charset":1,"continentcode":"NA"}`)
		})

	})
}

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 check(c gs.Context, in, out string) (err error) {
	tmpl := fmt.Sprintf("<%d>%%s %%s syslog_test[%%d]: %s\n", syslog.LOG_USER+syslog.LOG_INFO, in)
	if hostname, err := os.Hostname(); err != nil {
		return errors.New("Error retrieving hostname")
	} else {
		var parsedHostname, timestamp string
		var pid int

		// The stdlib tests that hostname matches parsedHostname, we
		// don't bother
		if n, err := fmt.Sscanf(out, tmpl, &timestamp, &parsedHostname, &pid); n != 3 || err != nil || hostname != parsedHostname {
			return errors.New("Error extracting timestamp, parsedHostname, pid")
		}
		computed_in := fmt.Sprintf(tmpl, timestamp, parsedHostname, pid)
		c.Expect(computed_in, gs.Equals, out)
	}

	return nil
}

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

	globals := &GlobalConfigStruct{
		PluginChanSize: 5,
	}
	NewPipelineConfig(globals)

	mockHelper := NewMockPluginHelper(ctrl)

	c.Specify("Runner restarts a plugin on the first time only", func() {
		var pluginGlobals PluginGlobals
		pluginGlobals.Retries = RetryOptions{
			MaxDelay:   "1us",
			Delay:      "1us",
			MaxJitter:  "1us",
			MaxRetries: 1,
		}
		pc := new(PipelineConfig)
		pc.inputWrappers = make(map[string]*PluginWrapper)

		pw := &PluginWrapper{
			Name:          "stopping",
			ConfigCreator: func() interface{} { return nil },
			PluginCreator: func() interface{} { return new(StoppingInput) },
		}
		pc.inputWrappers["stopping"] = pw

		input := new(StoppingInput)
		iRunner := NewInputRunner("stopping", input, &pluginGlobals, false)
		var wg sync.WaitGroup
		cfgCall := mockHelper.EXPECT().PipelineConfig().Times(7)
		cfgCall.Return(pc)
		wg.Add(1)
		iRunner.Start(mockHelper, &wg)
		wg.Wait()
		c.Expect(stopinputTimes, gs.Equals, 2)
	})
}

func CheckTypeSpec(c gs.Context) {
	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 classics struct {
		Ranking []string
		Bands   map[string]Primitive
	}

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

		empty_ignore := map[string]interface{}{}
		err = CheckType(md.mapping, music, empty_ignore)
		c.Assume(err, gs.IsNil)
	})
}

// decodeMessageAndVerifyOutput takes a decoder conf, message payload, and a fn -> the fn is a number of
// assertions to verify that the message after decoding is as expected.
func decodeMessageAndVerifyOutput(c gs.Context, conf *JsonDecoderConfig, payload string, fn packVerifier) {
	t := &pipeline_ts.SimpleT{}
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	// 1. Initialize test decoder
	decoder := new(JsonDecoder)
	err := decoder.Init(conf)
	c.Assume(err, gs.IsNil)
	dRunner := pipelinemock.NewMockDecoderRunner(ctrl)
	decoder.SetDecoderRunner(dRunner)

	// 2. Set payload to be tested, and decode it
	supply := make(chan *PipelinePack, 1)
	pack := NewPipelinePack(supply)
	pack.Message.SetPayload(payload)
	_, err = decoder.Decode(pack)

	// 3. Assert outcome of decoding
	fn(c, pack)
	pack.Zero()
}

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 := filepath.Join(os.TempDir(), "tmpdir")
			err := os.MkdirAll(tmpdir, 0400)
			c.Assume(err, gs.IsNil)
			config.WorkingDirectory = tmpdir
			err = dashboardOutput.Init(config)
			if runtime.GOOS == "windows" {
				c.Assume(err, gs.IsNil)
			} else {
				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 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())

	ith.AddrStr = "localhost:55565"
	ith.ResolvedAddrStr = "127.0.0.1:55565"

	// set up mock helper, decoder set, and packSupply channel
	ith.MockHelper = pipelinemock.NewMockPluginHelper(ctrl)
	ith.MockInputRunner = pipelinemock.NewMockInputRunner(ctrl)
	ith.Decoder = pipelinemock.NewMockDecoderRunner(ctrl)
	ith.PackSupply = make(chan *PipelinePack, 1)
	ith.DecodeChan = make(chan *PipelinePack)

	c.Specify("A UdpInput", func() {
		udpInput := UdpInput{}
		err := udpInput.Init(&UdpInputConfig{Net: "udp", Address: ith.AddrStr,
			Decoder:    "ProtobufDecoder",
			ParserType: "message.proto"})
		c.Assume(err, gs.IsNil)
		realListener := (udpInput.listener).(*net.UDPConn)
		c.Expect(realListener.LocalAddr().String(), gs.Equals, ith.ResolvedAddrStr)

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

		mockDecoderRunner := ith.Decoder.(*pipelinemock.MockDecoderRunner)
		mockDecoderRunner.EXPECT().InChan().Return(ith.DecodeChan)
		ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply)
		ith.MockInputRunner.EXPECT().Name().Return("UdpInput")
		encCall := ith.MockHelper.EXPECT().DecoderRunner("ProtobufDecoder", "UdpInput-ProtobufDecoder")
		encCall.Return(ith.Decoder, true)

		c.Specify("reads a message from the connection and passes it to the decoder", func() {
			hbytes, _ := proto.Marshal(header)
			go func() {
				udpInput.Run(ith.MockInputRunner, ith.MockHelper)
			}()
			conn, err := net.Dial("udp", ith.AddrStr) // a mock connection will not work here since the mock read cannot block
			c.Assume(err, gs.IsNil)
			buf := encodeMessage(hbytes, mbytes)
			_, err = conn.Write(buf)
			c.Assume(err, gs.IsNil)
			ith.PackSupply <- ith.Pack
			packRef := <-ith.DecodeChan
			udpInput.Stop()
			c.Expect(ith.Pack, gs.Equals, packRef)
			c.Expect(string(ith.Pack.MsgBytes), gs.Equals, string(mbytes))
			c.Expect(ith.Pack.Decoded, gs.IsFalse)
		})
	})

	c.Specify("A UdpInput Multiline input", func() {
		ith.AddrStr = "localhost:55566"
		ith.ResolvedAddrStr = "127.0.0.1:55566"
		udpInput := UdpInput{}
		err := udpInput.Init(&UdpInputConfig{Net: "udp", Address: ith.AddrStr,
			Decoder:    "test",
			ParserType: "token"})
		c.Assume(err, gs.IsNil)
		realListener := (udpInput.listener).(*net.UDPConn)
		c.Expect(realListener.LocalAddr().String(), gs.Equals, ith.ResolvedAddrStr)

		mockDecoderRunner := ith.Decoder.(*pipelinemock.MockDecoderRunner)
		mockDecoderRunner.EXPECT().InChan().Return(ith.DecodeChan).Times(2)
		ith.MockInputRunner.EXPECT().InChan().Return(ith.PackSupply).Times(2)
		ith.MockInputRunner.EXPECT().Name().Return("UdpInput").AnyTimes()
		encCall := ith.MockHelper.EXPECT().DecoderRunner("test", "UdpInput-test")
		encCall.Return(ith.Decoder, true)

		c.Specify("reads two messages from a packet and passes them to the decoder", func() {
			go func() {
				udpInput.Run(ith.MockInputRunner, ith.MockHelper)
			}()
			conn, err := net.Dial("udp", ith.AddrStr) // a mock connection will not work here since the mock read cannot block
			c.Assume(err, gs.IsNil)
			_, err = conn.Write([]byte("message1\nmessage2\n"))
			c.Assume(err, gs.IsNil)
			ith.PackSupply <- ith.Pack
			packRef := <-ith.DecodeChan
			c.Expect(string(packRef.Message.GetPayload()), gs.Equals, "message1\n")
			ith.PackSupply <- ith.Pack
			packRef = <-ith.DecodeChan
			c.Expect(string(packRef.Message.GetPayload()), gs.Equals, "message2\n")
			udpInput.Stop()
		})
	})
}

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

	pConfig := pipeline.NewPipelineConfig(nil)

	jsonPost := `{"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 := pipeline.HttpInput{}

		c.Specify("honors time ticker to flush", func() {
			ith := new(pipeline.InputTestHelper)
			ith.MockHelper = NewMockPluginHelper(ctrl)
			ith.MockInputRunner = NewMockInputRunner(ctrl)
			startInput := func() {
				go func() {
					err := httpInput.Run(ith.MockInputRunner, ith.MockHelper)
					c.Expect(err, gs.IsNil)
				}()
			}

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

			ith.Decoders = make([]DecoderRunner, int(message.Header_JSON+1))
			ith.Decoders[message.Header_JSON] = NewMockDecoderRunner(ctrl)
			ith.MockDecoderSet = NewMockDecoderSet(ctrl)

			// Spin up a http server
			server, err := ts.NewOneHttpServer(jsonPost, "localhost", 9876)
			c.Expect(err, gs.IsNil)
			go server.Start("/")
			time.Sleep(10 * time.Millisecond)

			config := httpInput.ConfigStruct().(*HttpInputConfig)
			config.DecoderName = "JsonDecoder"
			config.Url = "http://localhost:9876/"
			tickChan := make(chan time.Time)

			ith.MockInputRunner.EXPECT().LogMessage(gomock.Any()).Times(2)
			ith.MockHelper.EXPECT().DecoderSet().Return(ith.MockDecoderSet)

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

			mockDecoderRunner := ith.Decoders[message.Header_JSON].(*MockDecoderRunner)

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

			dset := ith.MockDecoderSet.EXPECT().ByName("JsonDecoder")
			dset.Return(ith.Decoders[message.Header_JSON], 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("short circuits packs into the router", func() {
			ith := new(InputTestHelper)
			ith.MockHelper = NewMockPluginHelper(ctrl)
			ith.MockInputRunner = NewMockInputRunner(ctrl)
			startInput := func() {
				go func() {
					err := httpInput.Run(ith.MockInputRunner, ith.MockHelper)
					c.Expect(err, gs.IsNil)
				}()
			}
			ith.Pack = NewPipelinePack(pConfig.inputRecycleChan)
			ith.PackSupply = make(chan *PipelinePack, 1)
			ith.PackSupply <- ith.Pack

			ith.Decoders = make([]DecoderRunner, int(message.Header_JSON+1))
			ith.Decoders[message.Header_JSON] = NewMockDecoderRunner(ctrl)
			ith.MockDecoderSet = NewMockDecoderSet(ctrl)
			config := httpInput.ConfigStruct().(*HttpInputConfig)
			config.Url = "http://localhost:9876/"
			tickChan := make(chan time.Time)

			ith.MockInputRunner.EXPECT().LogMessage(gomock.Any()).Times(2)
			ith.MockHelper.EXPECT().DecoderSet().Return(ith.MockDecoderSet)

			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)
//.........这里部分代码省略.........

func LoadFromConfigSpec(c gs.Context) {
	origGlobals := Globals

	origAvailablePlugins := make(map[string]func() interface{})
	for k, v := range AvailablePlugins {
		origAvailablePlugins[k] = v
	}

	pipeConfig := NewPipelineConfig(nil)
	defer func() {
		Globals = origGlobals
		AvailablePlugins = origAvailablePlugins
	}()

	c.Assume(pipeConfig, gs.Not(gs.IsNil))

	c.Specify("Config file loading", func() {
		c.Specify("works w/ good config file", func() {
			err := pipeConfig.LoadFromConfigFile("./testsupport/config_test.toml")
			c.Assume(err, gs.IsNil)

			// We use a set of Expect's rather than c.Specify because the
			// pipeConfig can't be re-loaded per child as gospec will do
			// since each one needs to bind to the same address

			// and the inputs section loads properly with a custom name
			udp, ok := pipeConfig.InputRunners["UdpInput"]
			c.Expect(ok, gs.Equals, true)

			// and the decoders sections load
			_, ok = pipeConfig.DecoderWrappers["JsonDecoder"]
			c.Expect(ok, gs.Equals, false)
			_, ok = pipeConfig.DecoderWrappers["ProtobufDecoder"]
			c.Expect(ok, gs.Equals, true)

			// and the outputs section loads
			_, ok = pipeConfig.OutputRunners["LogOutput"]
			c.Expect(ok, gs.Equals, true)

			// and the filters sections loads
			_, ok = pipeConfig.FilterRunners["sample"]
			c.Expect(ok, gs.Equals, true)

			// Shut down UdpInput to free up the port for future tests.
			udp.Input().Stop()
		})

		c.Specify("works w/ decoder defaults", func() {
			err := pipeConfig.LoadFromConfigFile("./testsupport/config_test_defaults.toml")
			c.Assume(err, gs.Not(gs.IsNil))

			// Only the ProtobufDecoder is loaded
			c.Expect(len(pipeConfig.DecoderWrappers), gs.Equals, 1)
		})

		c.Specify("works w/ MultiDecoder", func() {
			err := pipeConfig.LoadFromConfigFile("./testsupport/config_test_multidecoder.toml")
			c.Assume(err, gs.IsNil)
			hasSyncDecoder := false

			// ProtobufDecoder will always be loaded
			c.Assume(len(pipeConfig.DecoderWrappers), gs.Equals, 2)

			// Check that the MultiDecoder actually loaded
			for k, _ := range pipeConfig.DecoderWrappers {
				if k == "syncdecoder" {
					hasSyncDecoder = true
					break
				}
			}
			c.Assume(hasSyncDecoder, gs.IsTrue)
		})

		c.Specify("explodes w/ bad config file", func() {
			err := pipeConfig.LoadFromConfigFile("./testsupport/config_bad_test.toml")
			c.Assume(err, gs.Not(gs.IsNil))
			c.Expect(err.Error(), ts.StringContains, "2 errors loading plugins")
			c.Expect(pipeConfig.LogMsgs, gs.ContainsAny, gs.Values("No such plugin: CounterOutput"))
		})

		c.Specify("handles missing config file correctly", func() {
			err := pipeConfig.LoadFromConfigFile("no_such_file.toml")
			c.Assume(err, gs.Not(gs.IsNil))
			if runtime.GOOS == "windows" {
				c.Expect(err.Error(), ts.StringContains, "open no_such_file.toml: The system cannot find the file specified.")
			} else {
				c.Expect(err.Error(), ts.StringContains, "open no_such_file.toml: no such file or directory")
			}
		})

		c.Specify("errors correctly w/ bad outputs config", func() {
			err := pipeConfig.LoadFromConfigFile("./testsupport/config_bad_outputs.toml")
			c.Assume(err, gs.Not(gs.IsNil))
			c.Expect(err.Error(), ts.StringContains, "1 errors loading plugins")
			msg := pipeConfig.LogMsgs[0]
			c.Expect(msg, ts.StringContains, "No such plugin")
		})

		c.Specify("for a DefaultsTestOutput", func() {
			RegisterPlugin("DefaultsTestOutput", func() interface{} {
//.........这里部分代码省略.........

func MatcherSpecificationSpec(c gospec.Context) {
	msg := getTestMessage()
	uuidStr := msg.GetUuidString()
	data := []byte("data")
	date := "Mon Jan 02 15:04:05 -0700 2006"
	field1, _ := NewField("bytes", data, "")
	field2, _ := NewField("int", int64(999), "")
	field2.AddValue(int64(1024))
	field3, _ := NewField("double", float64(99.9), "")
	field4, _ := NewField("bool", true, "")
	field5, _ := NewField("foo", "alternate", "")
	field6, _ := NewField("Payload", "name=test;type=web;", "")
	field7, _ := NewField("Timestamp", date, "date-time")
	field8, _ := NewField("zero", int64(0), "")
	field9, _ := NewField("string", "43", "")
	msg.AddField(field1)
	msg.AddField(field2)
	msg.AddField(field3)
	msg.AddField(field4)
	msg.AddField(field5)
	msg.AddField(field6)
	msg.AddField(field7)
	msg.AddField(field8)
	msg.AddField(field9)

	c.Specify("A MatcherSpecification", func() {
		malformed := []string{
			"",
			"bogus",
			"Type = 'test'",                                               // invalid operator
			"Pid == 'test='",                                              // Pid is not a string
			"Type == 'test' && (Severity==7 || Payload == 'Test Payload'", // missing paren
			"Invalid == 'bogus'",                                          // unknown variable name
			"Fields[]",                                                    // empty name key
			"Fields[test][]",                                              // empty field index
			"Fields[test][a]",                                             // non numeric field index
			"Fields[test][0][]",                                           // empty array index
			"Fields[test][0][a]",                                          // non numeric array index
			"Fields[test][0][0][]",                                        // extra index dimension
			"Fields[test][xxxx",                                           // unmatched bracket
			"Pid =~ /6/",                                                  // regex not allowed on numeric
			"Pid !~ /6/",                                                  // regex not allowed on numeric
			"Type =~ /test",                                               // unmatched slash
			"Type == /test/",                                              // incorrect operator
			"Type =~ 'test'",                                              // string instead of regexp
			"Type =~ /\\ytest/",                                           // invalid escape character
			"Type != 'test\"",                                             // mis matched quote types
			"Pid =~ 6",                                                    // number instead of regexp
			"NIL",                                                         // invalid use of constant
			"Type == NIL",                                                 // existence check only works on fields
			"Fields[test] > NIL",                                          // existence check only works with equals and not equals
		}

		negative := []string{
			"FALSE",
			"Type == 'test'&&(Severity==7||Payload=='Test Payload')",
			"EnvVersion == '0.9'",
			"EnvVersion != '0.8'",
			"EnvVersion > '0.9'",
			"EnvVersion >= '0.9'",
			"EnvVersion < '0.8'",
			"EnvVersion <= '0.7'",
			"Severity == 5",
			"Severity != 6",
			"Severity < 6",
			"Severity <= 5",
			"Severity > 6",
			"Severity >= 7",
			"Fields[foo] == 'ba'",
			"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'",
//.........这里部分代码省略.........

func compareCaptures(c gospec.Context, m1, m2 map[string]string) {
	for k, v := range m1 {
		v1, _ := m2[k]
		c.Expect(v, gs.Equals, v1)
	}
}

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

	pConfig := pipeline.NewPipelineConfig(nil)

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

		decoder := new(SandboxDecoder)
		decoder.SetPipelineConfig(pConfig)
		conf := decoder.ConfigStruct().(*sandbox.SandboxConfig)
		supply := make(chan *pipeline.PipelinePack, 1)
		pack := pipeline.NewPipelinePack(supply)
		dRunner := pm.NewMockDecoderRunner(ctrl)

		c.Specify("that uses lpeg and inject_message", func() {
			dRunner.EXPECT().Name().Return("serialize")
			conf.ScriptFilename = "../lua/testsupport/decoder.lua"
			err := decoder.Init(conf)
			c.Assume(err, gs.IsNil)

			c.Specify("decodes simple messages", func() {
				data := "1376389920 debug id=2321 url=example.com item=1"
				decoder.SetDecoderRunner(dRunner)
				pack.Message.SetPayload(data)
				_, err = decoder.Decode(pack)
				c.Assume(err, gs.IsNil)

				c.Expect(pack.Message.GetTimestamp(),
					gs.Equals,
					int64(1376389920000000000))

				c.Expect(pack.Message.GetSeverity(), gs.Equals, int32(7))

				var ok bool
				var value interface{}
				value, ok = pack.Message.GetFieldValue("id")
				c.Expect(ok, gs.Equals, true)
				c.Expect(value, gs.Equals, "2321")

				value, ok = pack.Message.GetFieldValue("url")
				c.Expect(ok, gs.Equals, true)
				c.Expect(value, gs.Equals, "example.com")

				value, ok = pack.Message.GetFieldValue("item")
				c.Expect(ok, gs.Equals, true)
				c.Expect(value, gs.Equals, "1")
				decoder.Shutdown()
			})

			c.Specify("decodes an invalid messages", func() {
				data := "1376389920 bogus id=2321 url=example.com item=1"
				decoder.SetDecoderRunner(dRunner)
				pack.Message.SetPayload(data)
				packs, err := decoder.Decode(pack)
				c.Expect(len(packs), gs.Equals, 0)
				c.Expect(err.Error(), gs.Equals, "Failed parsing: "+data)
				c.Expect(decoder.processMessageFailures, gs.Equals, int64(1))
				decoder.Shutdown()
			})

			c.Specify("Preserves data", func() {
				conf.ScriptFilename = "../lua/testsupport/serialize.lua"
				conf.PreserveData = true
				err := decoder.Init(conf)
				c.Assume(err, gs.IsNil)
				decoder.SetDecoderRunner(dRunner)
				decoder.Shutdown()
				_, err = os.Stat("sandbox_preservation/serialize.data")
				c.Expect(err, gs.IsNil)
				err = os.Remove("sandbox_preservation/serialize.data")
				c.Expect(err, gs.IsNil)
			})
		})

		c.Specify("that only uses write_message", func() {
			conf.ScriptFilename = "../lua/testsupport/write_message_decoder.lua"
			dRunner.EXPECT().Name().Return("write_message")
			err := decoder.Init(conf)
			decoder.SetDecoderRunner(dRunner)
			c.Assume(err, gs.IsNil)

			c.Specify("adds a string field to the message", func() {
				data := "string field scribble"
				pack.Message.SetPayload(data)
				packs, err := decoder.Decode(pack)
				c.Expect(err, gs.IsNil)
				c.Expect(len(packs), gs.Equals, 1)
				c.Expect(packs[0], gs.Equals, pack)
				value, ok := pack.Message.GetFieldValue("scribble")
				c.Expect(ok, gs.IsTrue)
				c.Expect(value.(string), gs.Equals, "foo")
			})

			c.Specify("adds a numeric field to the message", func() {
				data := "num field scribble"
				pack.Message.SetPayload(data)
				packs, err := decoder.Decode(pack)
				c.Expect(err, gs.IsNil)
//.........这里部分代码省略.........