Method NewDriver

azqueue.go:30–65

(ep *Endpoint, cfg *Config)

	// seqHeaderSize is the big-endian sequence number prepended to every queue
	// message. Azure Storage Queues are at-least-once and not strictly FIFO, so
	// the receiver uses this to dedup and reorder into a clean byte stream.
	seqHeaderSize = 8
	// queueSizeMargin keeps the base64-encoded message safely below the 64 KiB
	// queue ceiling. See MaxRawSize: base64 expands by 4/3, and the encoded form
	// of (seqHeaderSize + sealed chunk) must stay <= MaxQueueTextMessageSize.
	queueSizeMargin = 64
	// maxPendingMessages caps the reassembly buffer's memory footprint. Liveness
	// is not tied to it: the stall clock in ReadRaw fails a missing sequence.
	maxPendingMessages = 256
	// defaultReassemblyStall bounds the wait for a missing sequence when the
	// config carries no usable idle timeout.
	defaultReassemblyStall = 60 * time.Second
)

// ErrReassemblyOverflow is returned when the azqueue reassembly buffer exceeds
// maxPendingMessages, indicating a sequence gap that will not resolve.
var ErrReassemblyOverflow = errors.New("azqueue: reassembly buffer overflow")

// ErrReassemblyStalled is returned when a sequence gap in the azqueue receive
// stream fails to close within the stall bound, i.e. a message was lost and the
// byte stream can never be completed.
var ErrReassemblyStalled = errors.New("azqueue: reassembly stalled on missing sequence")

func init() {
	RegisterFactory(queueDriverName, &queueFactory{})
}

type queueFactory struct{}

func (d *queueFactory) NewDriver(ep *Endpoint, cfg *Config) (Driver, error) {
	client, err := newQueueClient(ep)
	if err != nil {
		return nil, err
	}

	if client != nil {
		for _, name := range []string{cfg.handshakeEndpoint, cfg.tokenEndpoint} {
			if _, err := client.CreateQueue(cfg.ctx, name, nil); err != nil && !queueerror.HasCode(err, queueerror.QueueAlreadyExists) {
				return nil, err