run gofumpt on codebase

internal/config/resource_reader_test.go

@@ -66,7 +66,7 @@ processor_resources:
 	tCtx, done := context.WithTimeout(context.Background(), time.Second*30)
 	defer done()
 
-	testProc := func(name string, input, output string) {
+	testProc := func(name, input, output string) {
 		require.NoError(t, testMgr.AccessProcessor(tCtx, name, func(p processor.V1) {
 			res, err := p.ProcessBatch(tCtx, message.Batch{
 				message.NewPart([]byte(input)),
@@ -168,7 +168,7 @@ processor_resources:
 	tCtx, done := context.WithTimeout(context.Background(), time.Second*30)
 	defer done()
 
-	testProc := func(name string, input, output string) {
+	testProc := func(name, input, output string) {
 		require.NoError(t, testMgr.AccessProcessor(tCtx, name, func(p processor.V1) {
 			res, err := p.ProcessBatch(tCtx, message.Batch{
 				message.NewPart([]byte(input)),

internal/httpclient/config_test.go

@@ -121,5 +121,4 @@ oauth2:
 			assert.Equal(t, test.outputConf, conf)
 		})
 	}
-
 }

internal/impl/aws/integration_kinesis_test.go

@@ -151,7 +151,6 @@ input:
 						vars.Var1 += fmt.Sprintf(",%v:%v", streamName, shard)
 					}
 				}
-
 			}),
 			integration.StreamTestOptPort(resource.GetPort("4566/tcp")),
 			integration.StreamTestOptAllowDupes(),

internal/impl/couchbase/client.go

@@ -20,7 +20,6 @@ type couchbaseClient struct {
 }
 
 func getClient(conf *service.ParsedConfig, mgr *service.Resources) (*couchbaseClient, error) {
-
 	// retrieve params
 	url, err := conf.FieldString("url")
 	if err != nil {

internal/impl/crypto/argon2.go

@@ -14,9 +14,7 @@ import (
 	"github.com/benthosdev/benthos/v4/public/bloblang"
 )
 
-var (
-	errInvalidArgon2Hash = errors.New("invalid argon2 hash")
-)
+var errInvalidArgon2Hash = errors.New("invalid argon2 hash")
 
 type argon2Value struct {
 	format  string

internal/impl/crypto/jwt_sign.go

@@ -40,7 +40,6 @@ func registerSignJwtHSMethod(name string, signingMethod *jwt.SigningMethodHMAC,
 		)
 
 	return bloblang.RegisterMethodV2(name, spec, jwtHSSigner(signingMethod))
-
 }
 
 func registerSignJwtHSMethods() error {

internal/impl/crypto/jwt_sign_test.go

@@ -75,5 +75,4 @@ func TestBloblangSignJwtHS(t *testing.T) {
 			require.Equal(t, inClaims, outClaims)
 		})
 	}
-
 }

internal/impl/gcp/input_pubsub.go

@@ -81,7 +81,6 @@ func newGCPPubSubInput(conf input.Config, mgr bundle.NewManagement, log log.Modu
 
 func createSubscription(conf input.GCPPubSubConfig, client *pubsub.Client, log log.Modular) {
 	subsExists, err := client.Subscription(conf.SubscriptionID).Exists(context.Background())
-
 	if err != nil {
 		log.Errorf("Error checking if subscription exists", err)
 		return

internal/impl/kafka/input_kafka_franz.go

@@ -474,7 +474,7 @@ func recordToMessage(record *kgo.Record, multiHeader bool) *service.Message {
 	msg.MetaSet("kafka_timestamp_unix", strconv.FormatInt(record.Timestamp.Unix(), 10))
 	if multiHeader {
 		// in multi header mode we gather headers so we can encode them as lists
-		var headers = map[string][]any{}
+		headers := map[string][]any{}
 
 		for _, hdr := range record.Headers {
 			headers[hdr.Key] = append(headers[hdr.Key], string(hdr.Value))

internal/impl/kafka/input_sarama_kafka.go

@@ -389,10 +389,10 @@ func dataToPart(highestOffset int64, data *sarama.ConsumerMessage, multiHeader b
 
 	if multiHeader {
 		// in multi header mode we gather headers so we can encode them as lists
-		var headers = map[string][]any{}
+		headers := map[string][]any{}
 
 		for _, hdr := range data.Headers {
-			var key = string(hdr.Key)
+			key := string(hdr.Key)
 			headers[key] = append(headers[key], string(hdr.Value))
 		}
 

internal/impl/nats/processor_kv.go

@@ -107,7 +107,6 @@ This processor adds the following metadata fields to each message, depending on
       this.key == "" && this.operation != "keys" => [ "'key' must be set when operation is '" + this.operation + "'" ],
       this.key != "" && this.operation == "keys" => [ "'key' cannot be set when operation is '" + this.operation + "'" ],
     }`)
-
 }
 
 func init() {

internal/impl/pure/processor_unarchive.go

@@ -38,6 +38,7 @@ For the unarchive formats that contain file information (tar, zip), a metadata f
 			`csv:x`:          `Attempt to parse the message as a csv file (header required) and for each row in the file expands its contents into a json object in a new message using a custom delimiter. The custom delimiter must be a single character, e.g. the format "csv:\t" would consume a tab delimited file.`,
 		}).Description("The unarchiving format to apply."))
 }
+
 func init() {
 	err := service.RegisterProcessor(
 		"unarchive", unarchiveProcConfig(),

internal/impl/wasm/processor_wazero.go

@@ -44,7 +44,6 @@ func init() {
 		func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) {
 			return newWazeroAllocProcessorFromConfig(conf, mgr)
 		})
-
 	if err != nil {
 		panic(err)
 	}

public/service/config_interpolated_string.go

@@ -8,9 +8,7 @@ import (
 	"github.com/benthosdev/benthos/v4/internal/docs"
 )
 
-var (
-	errInvalidInterpolation = errors.New("failed to parse interpolated field")
-)
+var errInvalidInterpolation = errors.New("failed to parse interpolated field")
 
 // NewInterpolatedStringField defines a new config field that describes a
 // dynamic string that supports Bloblang interpolation functions. It is then

public/wasm/tinygo/tinygo.go

@@ -14,7 +14,7 @@ import (
 //
 //go:wasm-module benthos_wasm
 //export v0_msg_set_bytes
-func _v0_msg_set_bytes(ptr uint32, size uint32)
+func _v0_msg_set_bytes(ptr, size uint32)
 
 // SetMsgBytes sets the contents of the message currently being processed to the
 // value provided.
@@ -47,7 +47,7 @@ func GetMsgAsBytes() ([]byte, error) {
 
 // ptrToBytes returns a byte slice from WebAssembly compatible numeric types
 // representing its pointer and length.
-func ptrToBytes(ptr uint32, size uint32) []byte {
+func ptrToBytes(ptr, size uint32) []byte {
 	// Get a slice view of the underlying bytes in the stream. We use SliceHeader, not StringHeader
 	// as it allows us to fix the capacity to what was allocated.
 	return *(*[]byte)(unsafe.Pointer(&reflect.SliceHeader{