x-pack/filebeat/input/httpjson: clean up scaffolding

x-pack/filebeat/input/httpjson/request.go

@@ -610,8 +610,6 @@ func (p *chainProcessor) eventCount() int {
 	return p.n
 }
 
-func (*chainProcessor) close() {}
-
 // processChainPaginationEvents takes a pagination response as input and runs all the chain blocks for the input
 //
 //nolint:bodyclose // response body is closed through drainBody method
@@ -742,22 +740,6 @@ func newPublisher(trCtx *transformContext, pub inputcursor.Publisher, publish bo
 	}
 }
 
-// processAndPublishEvents process and publish events based on event type
-func (p *publisher) processAndPublishEvents(events stream) {
-	for maybeMsg := range events.ch {
-		p.processAndPublishEvent(maybeMsg)
-	}
-}
-
-// processAndPublishEvent processes and publishes one events based on event type
-func (p *publisher) processAndPublishEvent(evt maybeMsg) {
-	if evt.failed() {
-		p.fail(evt.err)
-		return
-	}
-	p.event(nil, evt.msg)
-}
-
 func (p *publisher) event(_ context.Context, msg mapstr.M) {
 	if p.pub != nil {
 		event, err := makeEvent(msg)
@@ -788,8 +770,6 @@ func (p *publisher) eventCount() int {
 	return p.n
 }
 
-func (p *publisher) close() {}
-
 const (
 	// This is generally updated with chain responses, if present, as they continue to occur
 	// Otherwise this is always the last response of the root request w.r.t pagination

x-pack/filebeat/input/httpjson/response.go

@@ -183,35 +183,12 @@ func newChainResponseProcessor(config chainConfig, httpClient *httpClient, xmlDe
 type sendStream interface {
 	event(context.Context, mapstr.M)
 	fail(error)
-	close()
-}
-
-type stream struct {
-	ch chan maybeMsg
-}
-
-func newStream() stream {
-	return stream{make(chan maybeMsg)}
-}
-
-func (s stream) event(_ context.Context, e mapstr.M) {
-	s.ch <- maybeMsg{msg: e}
-}
-
-func (s stream) fail(err error) {
-	s.ch <- maybeMsg{err: err}
-}
-
-func (s stream) close() {
-	close(s.ch)
 }
 
 func (rp *responseProcessor) startProcessing(stdCtx context.Context, trCtx *transformContext, resps []*http.Response, paginate bool, ch sendStream) {
 	trCtx.clearIntervalData()
 
-	defer ch.close()
 	var npages int64
-
 	for i, httpResp := range resps {
 		iter := rp.pagination.newPageIterator(stdCtx, trCtx, httpResp, rp.xmlDetails)
 		for {

x-pack/filebeat/input/httpjson/split_test.go

@@ -5,6 +5,7 @@
 package httpjson
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -702,9 +703,8 @@ func TestSplit(t *testing.T) {
 	}
 
 	for _, tc := range cases {
-		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
-			events := stream{make(chan maybeMsg, len(tc.expectedMessages))}
+			events := &stream{}
 			split, err := newSplitResponse(tc.config, logp.NewLogger(""))
 			assert.NoError(t, err)
 			err = split.run(tc.ctx, tc.resp, events)
@@ -714,12 +714,25 @@ func TestSplit(t *testing.T) {
 				assert.EqualError(t, err, tc.expectedErr.Error())
 			}
 			events.close()
-			assert.Equal(t, len(tc.expectedMessages), len(events.ch))
-			for _, msg := range tc.expectedMessages {
-				e := <-events.ch
-				assert.NoError(t, e.err)
-				assert.Equal(t, msg.Flatten(), e.msg.Flatten())
+			assert.Equal(t, len(tc.expectedMessages), len(events.collected))
+			for i, msg := range tc.expectedMessages {
+				assert.NoError(t, events.collected[i].err)
+				assert.Equal(t, msg.Flatten(), events.collected[i].msg.Flatten())
 			}
 		})
 	}
 }
+
+type stream struct {
+	collected []maybeMsg
+}
+
+func (s *stream) event(_ context.Context, e mapstr.M) {
+	s.collected = append(s.collected, maybeMsg{msg: e})
+}
+
+func (s *stream) fail(err error) {
+	s.collected = append(s.collected, maybeMsg{err: err})
+}
+
+func (s *stream) close() {}