func NewProtocol(server, queue string, connOption []amqp.ConnOption, sessionOption []amqp.SessionOption, opts ...Option) (*Protocol, error) {
	client, err := amqp.Dial(server, connOption...)
	if err != nil {
		return nil, err
	}
	session, err := client.NewSession(sessionOption...)
	if err != nil {
		_ = client.Close()
		return nil, err
	}
	p, err := NewProtocolFromClient(client, session, queue, opts...)
	if err != nil {
		return nil, err
	}
	p.ownedClient = true
	return p, nil
}
NewProtocol creates a new amqp transport.
package amqp
import (
	"context"
	"github.com/Azure/go-amqp"
	"github.com/cloudevents/sdk-go/v2/binding"
	"github.com/cloudevents/sdk-go/v2/protocol"
)
type Protocol struct {
	connOpts         []amqp.ConnOption
	sessionOpts      []amqp.SessionOption
	senderLinkOpts   []amqp.LinkOption
	receiverLinkOpts []amqp.LinkOption
	Client      *amqp.Client
	Session     *amqp.Session
	ownedClient bool
	Node        string
	Sender                  *sender
	SenderContextDecorators []func(context.Context) context.Context
	Receiver *receiver
}
func NewProtocolFromClient(client *amqp.Client, session *amqp.Session, queue string, opts ...Option) (*Protocol, error) {
	t := &Protocol{
		Node:             queue,
		senderLinkOpts:   []amqp.LinkOption(nil),
		receiverLinkOpts: []amqp.LinkOption(nil),
		Client:           client,
		Session:          session,
	}
	if err := t.applyOptions(opts...); err != nil {
		return nil, err
	}
	t.senderLinkOpts = append(t.senderLinkOpts, amqp.LinkTargetAddress(queue))
	amqpSender, err := session.NewSender(t.senderLinkOpts...)
	if err != nil {
		_ = client.Close()
		_ = session.Close(context.Background())
		return nil, err
	}
	t.Sender = NewSender(amqpSender).(*sender)
	t.SenderContextDecorators = []func(context.Context) context.Context{}
	t.receiverLinkOpts = append(t.receiverLinkOpts, amqp.LinkSourceAddress(t.Node))
	amqpReceiver, err := t.Session.NewReceiver(t.receiverLinkOpts...)
	if err != nil {
		return nil, err
	}
	t.Receiver = NewReceiver(amqpReceiver).(*receiver)
	return t, nil
}