Move chained processor to model/modelprocessor (#4925)

beater/beater.go

@@ -388,9 +388,9 @@ func (s *serverRunner) run() error {
 		// The server has been configured to discard unsampled
 		// transactions. Make sure this is done just before calling
 		// the publisher to avoid affecting aggregations.
-		batchProcessor = chainBatchProcessors(
+		batchProcessor = modelprocessor.Chained{
 			sampling.NewDiscardUnsampledBatchProcessor(), batchProcessor,
-		)
+		}
 	}
 
 	if err := runServer(s.runServerContext, ServerParams{
@@ -620,22 +620,11 @@ func WrapRunServerWithProcessors(runServer RunServerFunc, processors ...model.Ba
 	}
 	return func(ctx context.Context, args ServerParams) error {
 		processors = append(processors, args.BatchProcessor)
-		args.BatchProcessor = chainBatchProcessors(processors...)
+		args.BatchProcessor = modelprocessor.Chained(processors)
 		return runServer(ctx, args)
 	}
 }
 
-func chainBatchProcessors(processors ...model.BatchProcessor) model.BatchProcessor {
-	return model.ProcessBatchFunc(func(ctx context.Context, batch *model.Batch) error {
-		for _, p := range processors {
-			if err := p.ProcessBatch(ctx, batch); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-}
-
 type disablePublisherTracingKey struct{}
 
 type reporterBatchProcessor struct {

model/modelprocessor/chained.go

@@ -0,0 +1,38 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package modelprocessor
+
+import (
+	"context"
+
+	"github.com/elastic/apm-server/model"
+)
+
+// Chained is a chained model.BatchProcessor, calling each of
+// the processors in the slice in series.
+type Chained []model.BatchProcessor
+
+// ProcessBatch calls each of the processors in c in series.
+func (c Chained) ProcessBatch(ctx context.Context, batch *model.Batch) error {
+	for _, p := range c {
+		if err := p.ProcessBatch(ctx, batch); err != nil {
+			return err
+		}
+	}
+	return nil
+}