You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: input.go
    path of file: ./repos/benthos/internal/impl/amqp1
    
    the code of the file until where you have to start completion: package amqp1

import (
	"context"
	_ "embed"
	"errors"
	"fmt"
	"math/rand"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/Azure/go-amqp"

	"github.com/benthosdev/benthos/v4/internal/component"
	"github.com/benthosdev/benthos/v4/public/service"
)

//go:embed input_description.md
var inputDescription string

func amqp1InputSpec() *service.ConfigSpec {
	return service.NewConfigSpec().
		Stable().
		Categories("Services").
		Summary("Reads messages from an AMQP (1.0) server.").
		Description(inputDescription).
		Fields(
			service.NewURLField(urlField).
				Description("A URL to connect to.").
				Example("amqp://localhost:5672/").
				Example("amqps://guest:
    
func amqp1InputSpec() *service.ConfigSpec {
	return service.NewConfigSpec().
		Stable().
		Categories("Services").
		Summary("Reads messages from an AMQP (1.0) server.").
		Description(inputDescription).
		Fields(
			service.NewURLField(urlField).
				Description("A URL to connect to.").
				Example("amqp://localhost:5672/").
				Example("amqps://guest:guest@localhost:5672/").
				Deprecated().
				Optional(),
			service.NewURLListField(urlsField).
				Description("A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.").
				Example([]string{"amqp://guest:guest@127.0.0.1:5672/"}).
				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
				Optional().
				Version("4.23.0"),
			service.NewStringField(sourceAddrField).
				Description("The source address to consume from.").
				Example("/foo").
				Example("queue:/bar").
				Example("topic:/baz"),
			service.NewBoolField(azureRenewLockField).
				Description("Experimental: Azure service bus specific option to renew lock if processing takes more then configured lock time").
				Version("3.45.0").
				Default(false).
				Advanced(),
			service.NewBoolField(getMessageHeaderField).
				Description("Read additional message header fields into `amqp_*` metadata properties.").
				Version("4.25.0").
				Default(false).Advanced(),
			service.NewIntField(creditField).
				Description("Specifies the maximum number of unacknowledged messages the sender can transmit. Once this limit is reached, no more messages will arrive until messages are acknowledged and settled.").
				LintRule(`root = if this < 1 { [ "`+creditField+` must be at least 1" ] }`).
				Version("4.26.0").
				Default(64).
				Advanced(),
			service.NewTLSToggledField(tlsField),
			saslFieldSpec(),
		).LintRule(`
root = if this.url.or("") == "" && this.urls.or([]).length() == 0 {
  "field 'urls' must be set"
}
`)
}