Merge branch 'erewh0n-fix-file-polling-input' into dev

mozilla-services · Feb 24, 2015 · af72598 · af72598
2 parents f91b9b2 + d7cf3c0
commit af72598
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 1 deletion.
diff --git a/plugins/file/file_output_test.go b/plugins/file/file_output_test.go
@@ -299,6 +299,7 @@ func FileOutputSpec(c gs.Context) {
 
 				c.Specify("when interval triggers first", func() {
 					timerChan <- time.Now()
+					runtime.Gosched()
 					select {
 					case <-fileOutput.batchChan:
 					default:

diff --git a/plugins/file/file_polling_input.go b/plugins/file/file_polling_input.go
@@ -54,6 +54,8 @@ func (input *FilePollingInput) Stop() {
 
 func (input *FilePollingInput) packDecorator(pack *pipeline.PipelinePack) {
 	pack.Message.SetType("heka.file.polling")
+	pack.Message.SetHostname(input.hostname)
+
 	field, err := message.NewField("TickerInterval", int(input.TickerInterval), "")
 	if err != nil {
 		input.runner.LogError(
@@ -76,8 +78,10 @@ func (input *FilePollingInput) Run(runner pipeline.InputRunner,
 	input.runner = runner
 	input.hostname = helper.PipelineConfig().Hostname()
 	tickChan := runner.Ticker()
-
 	sRunner := runner.NewSplitterRunner("")
+	if !sRunner.UseMsgBytes() {
+		sRunner.SetPackDecorator(input.packDecorator)
+	}
 
 	for {
 		select {

diff --git a/plugins/file/file_polling_input_test.go b/plugins/file/file_polling_input_test.go
@@ -80,6 +80,8 @@ func FilePollingInputSpec(c gs.Context) {
 			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) {