<commit_before>// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package azservicebus

import (
	"context"
	"fmt"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal"
	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/utils"
	"github.com/Azure/go-amqp"
)

// SessionReceiver is a Receiver that handles sessions.
type SessionReceiver struct {
	inner       *Receiver
	sessionID   *string
	lockedUntil time.Time
}

// SessionReceiverOptions contains options for the `Client.AcceptSessionForQueue/Subscription` or `Client.AcceptNextSessionForQueue/Subscription`
// functions.
type SessionReceiverOptions struct {
	// ReceiveMode controls when a message is deleted from Service Bus.
	//
	// `azservicebus.PeekLock` is the default. The message is locked, preventing multiple
	// receivers from processing the message at once. You control the lock state of the message
	// using one of the message settlement functions like SessionReceiver.CompleteMessage(), which removes
	// it from Service Bus, or SessionReceiver..AbandonMessage(), which makes it available again.
	//
	// `azservicebus.ReceiveAndDelete` causes Service Bus to remove the message as soon
	// as it's received.
	//
	// More information about receive modes:
	// https://docs.microsoft.com/azure/service-bus-messaging/message-transfers-locks-settlement#settling-receive-operations
	ReceiveMode ReceiveMode
}

func toReceiverOptions(sropts *SessionReceiverOptions) *ReceiverOptions {
	if sropts == nil {
		return nil
	}

	return &ReceiverOptions{
		ReceiveMode: sropts.ReceiveMode,
	}
}

func newSessionReceiver(ctx context.Context, sessionID *string, ns internal.NamespaceWithNewAMQPLinks, entity entity, cleanupOnClose func(), options *ReceiverOptions) (*SessionReceiver, error) {
	sessionReceiver := &SessionReceiver{
		sessionID:   sessionID,
		lockedUntil: time.Time{},
	}

	r, err := newReceiver(newReceiverArgs{
		ns:             ns,
		entity:         entity,
		cleanupOnClose: cleanupOnClose,
		newLinkFn:      sessionReceiver.newLink,
	}, options)

	if err != nil {
		return nil, err
	}

	sessionReceiver.inner = r

	// temp workaround until we expose the session expiration time from the receiver in go-amqp
	if err := sessionReceiver.RenewSessionLock(ctx); err != nil {
		_ = sessionReceiver.Close(context.Background())
		return nil, err
	}

	return sessionReceiver, nil
}

func (r *SessionReceiver) newLink(ctx context.Context, session internal.AMQPSession) (internal.AMQPSenderCloser, internal.AMQPReceiverCloser, error) {
	const sessionFilterName = "com.microsoft:session-filter"
	const code = uint64(0x00000137000000C)

	linkOptions := createLinkOptions(r.inner.receiveMode, r.inner.amqpLinks.EntityPath())

	if r.sessionID == nil {
		linkOptions = append(linkOptions, amqp.LinkSourceFilter(sessionFilterName, code, nil))
	} else {
		linkOptions = append(linkOptions, amqp.LinkSourceFilter(sessionFilterName, code, r.sessionID))
	}

	link, err := createReceiverLink(ctx, session, linkOptions)

	if err != nil {
		return nil, nil, err
	}

	// check the session ID that came back - if we asked for a named session ID and didn't get it then
	// we failed to get the lock.
	// if we specified nil then we can _set_ our internally held session ID now that we know the value.
	receivedSessionID := link.LinkSourceFilterValue(sessionFilterName)
	receivedSessionIDStr, ok := receivedSessionID.(string)

	if !ok || (r.sessionID != nil && receivedSessionIDStr != *r.sessionID) {
		return nil, nil, fmt.Errorf("invalid type/value for returned sessionID(type:%T, value:%v)", receivedSessionID, receivedSessionID)
	}

	r.sessionID = &receivedSessionIDStr
	return nil, link, nil
}

// ReceiveMessages receives a fixed number of messages, up to numMessages.
// There are two ways to stop receiving messages:
// 1. Cancelling the `ctx` parameter.<commit_msg>Use correct argument name in function description (#17394)<commit_after>// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package azservicebus

import (
	"context"
	"fmt"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal"
	"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/utils"
	"github.com/Azure/go-amqp"
)

// SessionReceiver is a Receiver that handles sessions.
type SessionReceiver struct {
	inner       *Receiver
	sessionID   *string
	lockedUntil time.Time
}

// SessionReceiverOptions contains options for the `Client.AcceptSessionForQueue/Subscription` or `Client.AcceptNextSessionForQueue/Subscription`
// functions.
type SessionReceiverOptions struct {
	// ReceiveMode controls when a message is deleted from Service Bus.
	//
	// `azservicebus.PeekLock` is the default. The message is locked, preventing multiple
	// receivers from processing the message at once. You control the lock state of the message
	// using one of the message settlement functions like SessionReceiver.CompleteMessage(), which removes
	// it from Service Bus, or SessionReceiver..AbandonMessage(), which makes it available again.
	//
	// `azservicebus.ReceiveAndDelete` causes Service Bus to remove the message as soon
	// as it's received.
	//
	// More information about receive modes:
	// https://docs.microsoft.com/azure/service-bus-messaging/message-transfers-locks-settlement#settling-receive-operations
	ReceiveMode ReceiveMode
}

func toReceiverOptions(sropts *SessionReceiverOptions) *ReceiverOptions {
	if sropts == nil {
		return nil
	}

	return &ReceiverOptions{
		ReceiveMode: sropts.ReceiveMode,
	}
}

func newSessionReceiver(ctx context.Context, sessionID *string, ns internal.NamespaceWithNewAMQPLinks, entity entity, cleanupOnClose func(), options *ReceiverOptions) (*SessionReceiver, error) {
	sessionReceiver := &SessionReceiver{
		sessionID:   sessionID,
		lockedUntil: time.Time{},
	}

	r, err := newReceiver(newReceiverArgs{
		ns:             ns,
		entity:         entity,
		cleanupOnClose: cleanupOnClose,
		newLinkFn:      sessionReceiver.newLink,
	}, options)

	if err != nil {
		return nil, err
	}

	sessionReceiver.inner = r

	// temp workaround until we expose the session expiration time from the receiver in go-amqp
	if err := sessionReceiver.RenewSessionLock(ctx); err != nil {
		_ = sessionReceiver.Close(context.Background())
		return nil, err
	}

	return sessionReceiver, nil
}

func (r *SessionReceiver) newLink(ctx context.Context, session internal.AMQPSession) (internal.AMQPSenderCloser, internal.AMQPReceiverCloser, error) {
	const sessionFilterName = "com.microsoft:session-filter"
	const code = uint64(0x00000137000000C)

	linkOptions := createLinkOptions(r.inner.receiveMode, r.inner.amqpLinks.EntityPath())

	if r.sessionID == nil {
		linkOptions = append(linkOptions, amqp.LinkSourceFilter(sessionFilterName, code, nil))
	} else {
		linkOptions = append(linkOptions, amqp.LinkSourceFilter(sessionFilterName, code, r.sessionID))
	}

	link, err := createReceiverLink(ctx, session, linkOptions)

	if err != nil {
		return nil, nil, err
	}

	// check the session ID that came back - if we asked for a named session ID and didn't get it then
	// we failed to get the lock.
	// if we specified nil then we can _set_ our internally held session ID now that we know the value.
	receivedSessionID := link.LinkSourceFilterValue(sessionFilterName)
	receivedSessionIDStr, ok := receivedSessionID.(string)

	if !ok || (r.sessionID != nil && receivedSessionIDStr != *r.sessionID) {
		return nil, nil, fmt.Errorf("invalid type/value for returned sessionID(type:%T, value:%v)", receivedSessionID, receivedSessionID)
	}

	r.sessionID = &receivedSessionIDStr
	return nil, link, nil
}

// ReceiveMessages receives a fixed number of messages, up to maxMessages.
// There are two ways to stop receiving messages:
// 1. Cancelling the `ctx` parameter.
{'fraction_non_alphanumeric': 0.09670860338035328, 'fraction_numerical': 0.005210318973185919, 'mean_word_length': 4.495810055865922, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6193843', 'n_tokens_mistral': 2390, 'n_tokens_neox': 2252, 'n_words': 704}