text:
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package com.azure.messaging.servicebus;

import com.azure.core.amqp.models.AmqpAddress;
import com.azure.core.amqp.models.AmqpAnnotatedMessage;
import com.azure.core.amqp.models.AmqpMessageBody;
import com.azure.core.amqp.models.AmqpMessageBodyType;
import com.azure.core.amqp.models.AmqpMessageHeader;
import com.azure.core.amqp.models.AmqpMessageId;
import com.azure.core.amqp.models.AmqpMessageProperties;
import com.azure.core.util.logging.ClientLogger;
import com.azure.messaging.servicebus.implementation.DispositionStatus;
import com.azure.messaging.servicebus.implementation.MessagingEntityType;
import com.azure.messaging.servicebus.models.AbandonOptions;
import com.azure.messaging.servicebus.models.CompleteOptions;
import com.azure.messaging.servicebus.models.DeadLetterOptions;
import com.azure.messaging.servicebus.models.DeferOptions;
import com.azure.messaging.servicebus.models.SubQueue;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.nio.charset.Charset;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

import static com.azure.messaging.servicebus.TestUtils.MESSAGE_POSITION_ID;
import static com.azure.messaging.servicebus.TestUtils.getServiceBusMessages;
import static com.azure.messaging.servicebus.TestUtils.getSessionSubscriptionBaseName;
import static com.azure.messaging.servicebus.TestUtils.getSubscriptionBaseName;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
 * Integration tests for {@link ServiceBusReceiverAsyncClient} from queues or subscriptions.
 */
@Tag("integration")
class ServiceBusReceiverAsyncClientIntegrationTest extends IntegrationTestBase {
    private static final ClientLogger LOGGER = new ClientLogger(ServiceBusReceiverAsyncClientIntegrationTest.class);
    private final AtomicInteger messagesPending = new AtomicInteger();
    private final boolean isSessionEnabled = false;
    private final ClientCreationOptions defaultClientCreationOptions = new ClientCreationOptions()
        .setMaxAutoLockRenewDuration(Duration.ofMinutes(5));

    private ServiceBusReceiverAsyncClient receiver;
    private ServiceBusSenderAsyncClient sender;
    private ServiceBusSessionReceiverAsyncClient sessionReceiver;

    ServiceBusReceiverAsyncClientIntegrationTest() {
        super(new ClientLogger(ServiceBusReceiverAsyncClientIntegrationTest.class));
    }

    @Override
    protected void beforeTest() {
        sessionId = UUID.randomUUID().toString();
    }

    @Override
    protected void afterTest() {
        sharedBuilder = null;
        try {
            dispose(receiver, sender, sessionReceiver);
        } catch (Exception e) {
            LOGGER.warning("Error occurred when draining queue.", e);
        }
    }

    /**
     * Verifies that we can create multiple transaction using sender and receiver.
     */
    @Test
    void createMultipleTransactionTest() {
        // Arrange
        setSenderAndReceiver(MessagingEntityType.QUEUE, 0, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.createTransaction())
            .assertNext(Assertions::assertNotNull)
            .verifyComplete();

        StepVerifier.create(receiver.createTransaction())
            .assertNext(Assertions::assertNotNull)
            .verifyComplete();
    }

    /**
     * Verifies that we can create transaction and complete.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void createTransactionAndRollbackMessagesTest(MessagingEntityType entityType) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block(OPERATION_TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        AtomicReference<ServiceBusTransactionContext> transaction = new AtomicReference<>();
        StepVerifier.create(receiver.createTransaction())
            .assertNext(txn -> {
                transaction.set(txn);
                assertNotNull(transaction);
            })
            .verifyComplete();

        StepVerifier.create(receiver.receiveMessages()
            .flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> {
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
            }).verifyComplete();

        StepVerifier.create(receiver.rollbackTransaction(transaction.get()))
            .verifyComplete();
    }

    /**
     * Verifies that we can do following using shared connection and on non session entity. 1. create transaction 2.
     * receive and settle with transactionContext. 3. commit Rollback this transaction.
     */
    @ParameterizedTest
    @EnumSource(DispositionStatus.class)
    void transactionSendReceiveAndCommit(DispositionStatus dispositionStatus) {

        // Arrange
        final MessagingEntityType entityType = MessagingEntityType.QUEUE;
        setSenderAndReceiver(entityType, TestUtils.USE_CASE_PEEK_TRANSACTION_SENDRECEIVE_AND_COMPLETE, isSessionEnabled);

        final String messageId1 = UUID.randomUUID().toString();
        final ServiceBusMessage message1 = getMessage(messageId1, isSessionEnabled);
        final String deadLetterReason = "test reason";

        sendMessage(message1).block(TIMEOUT);

        // Assert & Act
        AtomicReference<ServiceBusTransactionContext> transaction = new AtomicReference<>();
        StepVerifier.create(receiver.createTransaction())
            .assertNext(txn -> {
                transaction.set(txn);
                assertNotNull(transaction);
            })
            .verifyComplete();
        assertNotNull(transaction.get());

        // Assert & Act
        final ServiceBusReceivedMessage message = receiver.receiveMessages()
            .flatMap(receivedMessage -> {
                final Mono<Void> operation;
                switch (dispositionStatus) {
                    case COMPLETED:
                        operation = receiver.complete(receivedMessage, new CompleteOptions().setTransactionContext(transaction.get()));
                        messagesPending.decrementAndGet();
                        break;
                    case ABANDONED:
                        operation = receiver.abandon(receivedMessage, new AbandonOptions().setTransactionContext(transaction.get()));
                        break;
                    case SUSPENDED:
                        DeadLetterOptions deadLetterOptions = new DeadLetterOptions().setTransactionContext(transaction.get())
                            .setDeadLetterReason(deadLetterReason);
                        operation = receiver.deadLetter(receivedMessage, deadLetterOptions);
                        messagesPending.decrementAndGet();
                        break;
                    case DEFERRED:
                        operation = receiver.defer(receivedMessage, new DeferOptions().setTransactionContext(transaction.get()));
                        break;
                    case RELEASED:
                        operation = receiver.release(receivedMessage);
                        break;
                    default:
                        throw LOGGER.logExceptionAsError(new IllegalArgumentException(
                            "Disposition status not recognized for this test case: " + dispositionStatus));
                }
                return operation
                    .thenReturn(receivedMessage);
            })
            .next().block(TIMEOUT);
        assertNotNull(message);

        StepVerifier.create(receiver.commitTransaction(transaction.get()))
            .verifyComplete();

    }

    /**
     * Verifies that we can do following on different clients i.e. sender and receiver. 1. create transaction using
     * sender 2. receive and complete with transactionContext. 3. Commit this transaction using sender.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    @Disabled
    void transactionReceiveCompleteCommitMixClient(MessagingEntityType entityType) {
        // Arrange
        final boolean shareConnection = true;
        final boolean useCredentials = false;
        final int entityIndex = 0;
        this.sender = getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection)
            .buildAsyncClient();
        this.receiver = getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
            .buildAsyncClient();

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block(TIMEOUT);

        // Assert & Act
        AtomicReference<ServiceBusTransactionContext> transaction = new AtomicReference<>();
        StepVerifier.create(sender.createTransaction())
            .assertNext(txn -> {
                transaction.set(txn);
                assertNotNull(transaction);
            })
            .verifyComplete();
        assertNotNull(transaction.get());

        // Assert & Act
        final ServiceBusReceivedMessage receivedMessage = receiver.receiveMessages().next().block(TIMEOUT);
        assertNotNull(receivedMessage);

        StepVerifier.create(receiver.complete(receivedMessage, new CompleteOptions().setTransactionContext(transaction.get())))
            .verifyComplete();

        StepVerifier.create(sender.commitTransaction(transaction.get()))
            .verifyComplete();
    }

    /**
     * Verifies that we can send and receive two messages.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveTwoMessagesAutoComplete(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int entityIndex = 0;
        final boolean shareConnection = false;
        final boolean useCredentials = false;
        final Duration shortWait = Duration.ofSeconds(3);

        this.sender = getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection)
            .buildAsyncClient();

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);
        Mono.when(sendMessage(message), sendMessage(message)).block();

        // Now create receiver
        if (isSessionEnabled) {
            assertNotNull(sessionId, "'sessionId' should have been set.");
            this.sessionReceiver = getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
            this.receiver = sessionReceiver.acceptSession(sessionId).block();
        } else {
            this.receiver = getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
        }

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages()
                .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId())))
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .thenAwait(shortWait) // Give time for autoComplete to finish
            .thenCancel()
            .verify();
    }

    /**
     * Verifies that we can send and receive a message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveMessageAutoComplete(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int entityIndex = 0;
        final boolean shareConnection = false;
        final boolean useCredentials = false;
        final Duration shortWait = Duration.ofSeconds(3);

        this.sender = getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection)
            .buildAsyncClient();
        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block();

        // Now create receiver
        if (isSessionEnabled) {
            assertNotNull(sessionId, "'sessionId' should have been set.");
            this.sessionReceiver = getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
            this.receiver = this.sessionReceiver.acceptSession(sessionId).block();
        } else {
            this.receiver = getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
        }

        // Assert
        StepVerifier.create(receiver.receiveMessages())
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .thenAwait(shortWait) // Give time for autoComplete to finish
            .thenCancel()
            .verify();

        StepVerifier.create(receiver.receiveMessages())
            .thenAwait(shortWait)
            .thenCancel()
            .verify();
    }

    /**
     * Verifies that we can send and peek a message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void peekMessage(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block();

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.peekMessage())
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .verifyComplete();

        // cleanup
        StepVerifier.create(receiver.receiveMessages().flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .verifyComplete();
    }

    /**
     * Verifies that an empty entity does not error when peeking.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void peekMessageEmptyEntity(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setReceiver(entityType, TestUtils.USE_CASE_EMPTY_ENTITY, isSessionEnabled);

        final int fromSequenceNumber = 1;

        // Assert & Act
        StepVerifier.create(receiver.peekMessage(fromSequenceNumber))
            .verifyComplete();
    }

    /**
     * Verifies that we can schedule and receive a message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void sendScheduledMessageAndReceive(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);
        final Duration shortDelay = Duration.ofSeconds(4);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);
        final OffsetDateTime scheduledEnqueueTime = OffsetDateTime.now().plusSeconds(2);

        sender.scheduleMessage(message, scheduledEnqueueTime).block();

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(Mono.delay(shortDelay).then(receiver.receiveMessages()
            .flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).next()))
            .assertNext(receivedMessage -> {
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
            }).verifyComplete();
    }

    /**
     * Verifies that we can cancel a scheduled message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void cancelScheduledMessage(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);
        final OffsetDateTime scheduledEnqueueTime = OffsetDateTime.now().plusSeconds(10);
        final Duration delayDuration = Duration.ofSeconds(3);

        final Long sequenceNumber = sender.scheduleMessage(message, scheduledEnqueueTime).block(TIMEOUT);
        LOGGER.info("Scheduled the message, sequence number {}.", sequenceNumber);

        assertNotNull(sequenceNumber);

        Mono.delay(delayDuration)
            .then(sender.cancelScheduledMessage(sequenceNumber))
            .block(TIMEOUT);

        messagesPending.decrementAndGet();
        LOGGER.info("Cancelled the scheduled message, sequence number {}.", sequenceNumber);
        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages().take(1))
            .thenAwait(Duration.ofSeconds(5))
            .thenCancel()
            .verify();
    }

    /**
     * Verifies that we can send and peek a message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void peekFromSequenceNumberMessage(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int entityIndex = 3;

        setSender(entityType, entityIndex, isSessionEnabled);
        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);
        final CountDownLatch countDownLatch = new CountDownLatch(1);

        sendMessage(message).block();

        setReceiver(entityType, entityIndex, isSessionEnabled);

        // Message are not always guaranteed, so try many times
        final ServiceBusReceivedMessage peekMessage = receiver.peekMessage()
            .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId()))
            .map(receivedMessage -> {
                countDownLatch.countDown();
                return receivedMessage;
            })
            .repeat(() -> countDownLatch.getCount() > 0)
            .next()
            .block(OPERATION_TIMEOUT);
        assertNotNull(peekMessage);
        final long sequenceNumber = peekMessage.getSequenceNumber();

        // Assert & Act
        try {
            StepVerifier.create(receiver.peekMessage(sequenceNumber))
                .assertNext(m -> {
                    assertEquals(sequenceNumber, m.getSequenceNumber());
                    assertMessageEquals(m, messageId, isSessionEnabled);
                })
                .verifyComplete();
        } finally {

            // Cleanup
            StepVerifier.create(receiver.receiveMessages()
                .flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).take(1))
                .expectNextCount(1)
                .verifyComplete();
            messagesPending.decrementAndGet();
        }
    }

    /**
     * Verifies that we can send and peek a batch of messages and the sequence number is tracked correctly.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void peekMessages(MessagingEntityType entityType, boolean isSessionEnabled) throws InterruptedException {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_PEEK_BATCH_MESSAGES, isSessionEnabled);

        final BiConsumer<ServiceBusReceivedMessage, Integer> checkCorrectMessage = (message, index) -> {
            final Map<String, Object> properties = message.getApplicationProperties();
            final Object value = properties.get(MESSAGE_POSITION_ID);
            assertTrue(value instanceof Integer, "Did not contain correct position number: " + value);

            final int position = (int) value;
            assertEquals(index, position);
        };
        final String messageId = UUID.randomUUID().toString();
        final List<ServiceBusMessage> messages = getServiceBusMessages(10, messageId, CONTENTS_BYTES);
        final List<Integer> receivedPositions = Collections.synchronizedList(new ArrayList<Integer>());
        final AtomicInteger messageCount = new AtomicInteger();
        final List<ServiceBusReceivedMessage> receivedMessages = Collections.synchronizedList(new ArrayList<ServiceBusReceivedMessage>());

        if (isSessionEnabled) {
            messages.forEach(m -> m.setSessionId(sessionId));
        }

        sender.sendMessages(messages)
            .doOnSuccess(aVoid -> {
                int number = messagesPending.addAndGet(messages.size());
                LOGGER.info("Number of messages sent: {}", number);
            })
            .block();

        setReceiver(entityType, TestUtils.USE_CASE_PEEK_BATCH_MESSAGES, isSessionEnabled);

        // Assert & Act
        try {
            // maxMessages are not always guaranteed, sometime, we get less than asked for, so we will try many times.
            List<Thread> threadList = new ArrayList<Thread>();
            threadList.add(new Thread(() -> {
                final AtomicLong actualCount = new AtomicLong();
                List<ServiceBusReceivedMessage> peekedMessages
                    = receiver.peekMessages(3)
                    .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId())
                        && (int) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID) >= 0
                        && (int) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID) <= 2
                        && receivedPositions.stream().noneMatch(position ->
                            Objects.equals(position, receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID))))
                    .map(receivedMessage -> {
                        receivedPositions.add((Integer) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID));
                        actualCount.incrementAndGet();
                        return receivedMessage;
                    })
                    .repeat(() -> actualCount.get() < 3)
                    .collectList().block();
                if (Objects.nonNull(peekedMessages) && !peekedMessages.isEmpty()) {
                    receivedMessages.addAll(peekedMessages);
                }
            }));

            // maxMessages are not always guaranteed, sometime, we get less than asked for, so we will try many times.
            threadList.add(new Thread(() -> {
                final AtomicLong actualCount = new AtomicLong();
                List<ServiceBusReceivedMessage> peekedMessages
                    = receiver.peekMessages(4)
                    .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId())
                        && (int) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID) >= 3
                        && (int) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID) <= 6
                        && receivedPositions.stream().noneMatch(position ->
                            Objects.equals(position, receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID))))
                    .map(receivedMessage -> {
                        receivedPositions.add((Integer) receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID));
                        actualCount.incrementAndGet();
                        return receivedMessage;
                    })
                    .repeat(() -> actualCount.get() < 4)
                    .collectList().block();
                if (Objects.nonNull(peekedMessages) && !peekedMessages.isEmpty()) {
                    receivedMessages.addAll(peekedMessages);
                }
            }));

            // Reads the next active message is not always guaranteed, so we will try many times.
            threadList.add(new Thread(() -> {
                final AtomicLong actualCount = new AtomicLong();
                List<ServiceBusReceivedMessage> peekedMessages
                    = receiver.peekMessage()
                    .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId())
                        && Objects.equals(7, receivedMessage.getApplicationProperties().get(MESSAGE_POSITION_ID)))
                    .map(receivedMessage -> {
                        actualCount.incrementAndGet();
                        return receivedMessage;
                    })
                    .repeat(() -> actualCount.get() < 1)
                    .collectList().block();
                if (Objects.nonNull(peekedMessages) && !peekedMessages.isEmpty()) {
                    receivedMessages.addAll(peekedMessages);
                }
            }));

            threadList.forEach(Thread::start);
            threadList.forEach(t -> {
                try {
                    t.join(TIMEOUT.toMillis());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });

            receivedMessages.stream()
                .sorted((o1, o2) -> {
                    int position1 = (int) o1.getApplicationProperties().get(MESSAGE_POSITION_ID);
                    int position2 = (int) o2.getApplicationProperties().get(MESSAGE_POSITION_ID);
                    return position1 - position2;
                })
                .forEach(actualMessage -> {
                    LOGGER.info("The position id of received message : {}", actualMessage.getApplicationProperties().get(MESSAGE_POSITION_ID));
                    checkCorrectMessage.accept(actualMessage, messageCount.getAndIncrement());
                });

        } finally {
            Thread finallyThread = new Thread(() -> {
                receiver.receiveMessages()
                    .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId()))
                    .subscribe(serviceBusReceivedMessage ->
                        receiver.complete(serviceBusReceivedMessage)
                            .thenReturn(serviceBusReceivedMessage)
                            .block()
                    );
                messagesPending.addAndGet(-messages.size());
                receivedPositions.clear();
            });
            finallyThread.start();
            finallyThread.join(TIMEOUT.toMillis());
        }
    }

    /**
     * Verifies that we can send and peek a batch of messages.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void peekMessagesFromSequence(MessagingEntityType entityType) throws InterruptedException {
        // Arrange
        setSenderAndReceiver(entityType, TestUtils.USE_CASE_PEEK_MESSAGE_FROM_SEQUENCE, false);

        final AtomicInteger messageId = new AtomicInteger();
        final int maxMessages = 2;
        final AtomicLong fromSequenceNumber = new AtomicLong();
        final CountDownLatch countdownLatch = new CountDownLatch(maxMessages);
        fromSequenceNumber.set(1);

        final byte[] content = "peek-message-from-sequence".getBytes(Charset.defaultCharset());
        List<String> messageIds = Collections.synchronizedList(new ArrayList<String>());
        for (int i = 0; i < maxMessages; ++i) {
            ServiceBusMessage message = getMessage(String.valueOf(i), isSessionEnabled, AmqpMessageBody.fromData(content));
            messageIds.add(String.valueOf(i));
            Mono.when(sendMessage(message)).block();
        }

        // Assert & Act

        // maxMessages are not always guaranteed, sometime, we get less than asked for, just trying two times is not enough, so we will try many times
        // https://github.com/Azure/azure-sdk-for-java/issues/21168
        List<String> receivedMessages = Collections.synchronizedList(new ArrayList<String>());
        receiver.peekMessages(maxMessages, fromSequenceNumber.get())
            .filter(receivedMessage -> messageIds.contains(receivedMessage.getMessageId())
                && receivedMessages.parallelStream().noneMatch(mid ->
                    mid.equals(receivedMessage.getMessageId())))
            .sort(Comparator.comparing(ServiceBusReceivedMessage::getMessageId))
            .flatMap(receivedMessage -> {
                Long previousSequenceNumber = fromSequenceNumber.get();
                fromSequenceNumber.set(receivedMessage.getSequenceNumber() + 1);
                countdownLatch.countDown();
                receivedMessages.add(receivedMessage.getMessageId());
                assertEquals(String.valueOf(messageId.getAndIncrement()), receivedMessage.getMessageId(),
                    String.format("Message id did not match. Message payload: [%s], peek from Sequence Number [%s], "
                            + " received message Sequence Number [%s]", receivedMessage.getBody(),
                        previousSequenceNumber, receivedMessage.getSequenceNumber()));
                return Mono.just(receivedMessage);
            })
            .repeat(() -> countdownLatch.getCount() > 0)
            .subscribe();

        if (!countdownLatch.await(20, TimeUnit.SECONDS)) {
            Assertions.fail("Failed peek messages from sequence.");
        }

        StepVerifier.create(receiver.receiveMessages().take(maxMessages))
            .assertNext(receivedMessage -> receiver.complete(receivedMessage).block(Duration.ofSeconds(15)))
            .assertNext(receivedMessage -> receiver.complete(receivedMessage).block(Duration.ofSeconds(15)))
            .expectComplete()
            .verify(TIMEOUT);
    }

    /**
     * Verifies that an empty entity does not error when peeking.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void peekMessagesFromSequenceEmptyEntity(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setReceiver(entityType, TestUtils.USE_CASE_EMPTY_ENTITY, isSessionEnabled);

        final int maxMessages = 10;
        final int fromSequenceNumber = 1;

        // Assert & Act
        StepVerifier.create(receiver.peekMessages(maxMessages, fromSequenceNumber))
            .verifyComplete();
    }

    /**
     * Verifies that we can dead-letter a message.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void deadLetterMessage(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int entityIndex = 0;
        setSender(entityType, entityIndex, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block();

        setReceiver(entityType, entityIndex, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages()
            .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId()))
            .flatMap(receivedMessage -> receiver.deadLetter(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> {
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
            })
            .thenCancel()
            .verify();

    }

    /**
     * Verifies that we can send and receive a message AMQP Sequence andValue object.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveMessageAmqpTypes(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int entityIndex = TestUtils.USE_CASE_AMQP_TYPES;
        final boolean shareConnection = false;
        final boolean useCredentials = false;
        final Duration shortWait = Duration.ofSeconds(3);
        final Long expectedLongValue = Long.parseLong("6");

        this.sender = getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection)
            .buildAsyncClient();

        // Send  value Object
        String messageId = UUID.randomUUID().toString();
        ServiceBusMessage message = getMessage(messageId, isSessionEnabled, AmqpMessageBody.fromValue(expectedLongValue));
        sendMessage(message).block(TIMEOUT);

        // send SEQUENCE
        messageId = UUID.randomUUID().toString();

        List<Object> sequenceData = new ArrayList<>();
        sequenceData.add("A1");
        sequenceData.add(1L);
        sequenceData.add(2);

        message = getMessage(messageId, isSessionEnabled, AmqpMessageBody.fromSequence(sequenceData));
        sendMessage(message).block(TIMEOUT);

        // Now create receiver
        if (isSessionEnabled) {
            assertNotNull(sessionId, "'sessionId' should have been set.");
            this.sessionReceiver = getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
            this.receiver = this.sessionReceiver.acceptSession(sessionId).block();
        } else {
            this.receiver = getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .buildAsyncClient();
        }

        // Assert
        StepVerifier.create(receiver.receiveMessages())
            .assertNext(receivedMessage -> {
                AmqpAnnotatedMessage amqpAnnotatedMessage = receivedMessage.getRawAmqpMessage();
                AmqpMessageBodyType type = amqpAnnotatedMessage.getBody().getBodyType();
                assertEquals(AmqpMessageBodyType.VALUE, type);
                Object value = amqpAnnotatedMessage.getBody().getValue();
                assertTrue(value instanceof Long);
                assertEquals(expectedLongValue.longValue(), ((Long) value).longValue());
            })
            .assertNext(receivedMessage -> {
                AmqpAnnotatedMessage amqpAnnotatedMessage = receivedMessage.getRawAmqpMessage();
                AmqpMessageBodyType type = amqpAnnotatedMessage.getBody().getBodyType();
                assertEquals(AmqpMessageBodyType.SEQUENCE, type);
                assertArrayEquals(sequenceData.toArray(), amqpAnnotatedMessage.getBody().getSequence().toArray());
            })
            .thenAwait(shortWait) // Give time for autoComplete to finish
            .thenCancel()
            .verify();

        if (!isSessionEnabled) {
            StepVerifier.create(receiver.receiveMessages())
                .thenAwait(shortWait)
                .thenCancel()
                .verify();
        }
    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveAndComplete(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages()
            .flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> {
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
            }).verifyComplete();

        messagesPending.decrementAndGet();
    }

    /**
     * Verifies that we can renew message lock on a non-session receiver.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void receiveAndRenewLock(MessagingEntityType entityType) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, false);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, false);

        // Blocking here because it is not part of the scenario we want to test.
        sendMessage(message).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, false);

        final ServiceBusReceivedMessage receivedMessage = receiver.receiveMessages().next().block(TIMEOUT);
        assertNotNull(receivedMessage);
        assertNotNull(receivedMessage.getLockedUntil());

        final OffsetDateTime initialLock = receivedMessage.getLockedUntil();
        LOGGER.info("Received message. Seq: {}. lockedUntil: {}", receivedMessage.getSequenceNumber(), initialLock);

        // Assert & Act
        try {
            StepVerifier.create(Mono.delay(Duration.ofSeconds(7))
                .then(Mono.defer(() -> receiver.renewMessageLock(receivedMessage))))
                .assertNext(lockedUntil -> assertTrue(lockedUntil.isAfter(initialLock),
                    String.format("Updated lock is not after the initial Lock. updated: [%s]. initial:[%s]",
                        lockedUntil, initialLock)))
                .verifyComplete();
        } finally {
            LOGGER.info("Completing message. Seq: {}.", receivedMessage.getSequenceNumber());

            receiver.complete(receivedMessage)
                .doOnSuccess(aVoid -> messagesPending.decrementAndGet())
                .block(TIMEOUT);
        }
    }

    /**
     * Receiver should receive the messages even if user is not "settling the messages" in PEEK LOCK mode and
     * autoComplete is disabled.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveMessagesNoMessageSettlement(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int totalMessages = 5;
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Send messages
        final String messageId = UUID.randomUUID().toString();
        final List<ServiceBusMessage> messages = TestUtils.getServiceBusMessages(totalMessages, messageId, CONTENTS_BYTES);
        if (isSessionEnabled) {
            messages.forEach(m -> m.setSessionId(sessionId));
        }
        sender.sendMessages(messages).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages().take(totalMessages))
            .expectNextCount(totalMessages)
            .verifyComplete();

        messagesPending.addAndGet(-totalMessages);
    }

    /**
     * Receiver should receive the messages if  processing time larger than message lock duration and
     * maxAutoLockRenewDuration is set to a large enough duration so user can complete in end.
     * This test takes longer time.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveMessagesLargeProcessingTime(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final int totalMessages = 2;
        // The lock timeout property set on the queue.
        final Duration lockRenewTimeout = Duration.ofSeconds(15);
        final ClientCreationOptions clientCreationOptions = new ClientCreationOptions().setMaxAutoLockRenewDuration(Duration.ofMinutes(1));
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Send messages
        final String messageId = UUID.randomUUID().toString();
        final List<ServiceBusMessage> messages = TestUtils.getServiceBusMessages(totalMessages, messageId, CONTENTS_BYTES);
        if (isSessionEnabled) {
            messages.forEach(m -> m.setSessionId(sessionId));
        }
        sender.sendMessages(messages).block();

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled, clientCreationOptions);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages()
                .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId()))
                .map(receivedMessage -> Mono.delay(lockRenewTimeout.plusSeconds(2))
            .then(receiver.complete(receivedMessage)).thenReturn(receivedMessage).block()).take(totalMessages))
            .expectNextCount(totalMessages)
            .verifyComplete();

        messagesPending.addAndGet(-totalMessages);
    }

    /**
     * Verifies that the lock can be automatically renewed.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void autoRenewLockOnReceiveMessage(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final AtomicInteger lockRenewCount = new AtomicInteger();

        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        // Send the message to verify.
        sendMessage(message).block();

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        // Act & Assert
        StepVerifier.create(receiver.receiveMessages().flatMap(received -> {
            LOGGER.info("{}: lockToken[{}]. lockedUntil[{}]. now[{}]", received.getSequenceNumber(),
                received.getLockToken(), received.getLockedUntil(), OffsetDateTime.now());

            // Simulate some sort of long processing.
            while (lockRenewCount.get() < 4) {
                lockRenewCount.incrementAndGet();
                LOGGER.info("Iteration {}: Curren time {}.", lockRenewCount.get(), OffsetDateTime.now());
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException error) {
                    LOGGER.error("Error occurred while sleeping: " + error);
                }
            }
            return receiver.complete(received).thenReturn(received);
        }))
            .assertNext(received -> {
                assertTrue(lockRenewCount.get() > 0);
                messagesPending.decrementAndGet();
            })
            .thenCancel()
            .verify();
    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveAndAbandon(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);
        StepVerifier.create(receiver.receiveMessages()
            .flatMap(receivedMessage -> receiver.abandon(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> assertMessageEquals(receivedMessage, messageId, isSessionEnabled))
            .expectComplete();

    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveAndDefer(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_PEEK_RECEIVE_AND_DEFER, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_PEEK_RECEIVE_AND_DEFER, isSessionEnabled);
        AtomicReference<ServiceBusReceivedMessage> received = new AtomicReference<ServiceBusReceivedMessage>();

        // Act & Assert
        StepVerifier.create(receiver.receiveMessages()
            .flatMap(receivedMessage -> receiver.defer(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(m -> {
                received.set(m);
                assertMessageEquals(m, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
            }).verifyComplete();

        // TODO(Hemant): Identify if this is valid scenario (https://github.com/Azure/azure-sdk-for-java/issues/19673)
        /*receiver.receiveDeferredMessage(received.get().getSequenceNumber())
            .flatMap(m -> receiver.complete(m))
            .block(TIMEOUT);
        messagesPending.decrementAndGet();
         */
    }

    /**
     * Test we can receive a deferred message via sequence number and then perform abandon, suspend, or complete on it.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#receiveDeferredMessageBySequenceNumber")
    @ParameterizedTest
    void receiveDeferredMessageBySequenceNumber(MessagingEntityType entityType, DispositionStatus dispositionStatus) {
        // Arrange
        setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFERRED_MESSAGE_BY_SEQUENCE_NUMBER, false);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, false);
        sendMessage(message).block(TIMEOUT);

        final ServiceBusReceivedMessage receivedMessage = receiver.receiveMessages()
            .flatMap(m -> receiver.defer(m).thenReturn(m))
            .next().block(TIMEOUT);

        assertNotNull(receivedMessage);

        // Assert & Act
        final ServiceBusReceivedMessage receivedDeferredMessage = receiver
            .receiveDeferredMessage(receivedMessage.getSequenceNumber())
            .flatMap(m -> {
                final Mono<Void> operation;
                switch (dispositionStatus) {
                    case ABANDONED:
                        operation = receiver.abandon(m);
                        break;
                    case SUSPENDED:
                        operation = receiver.deadLetter(m);
                        break;
                    case COMPLETED:
                        operation = receiver.complete(m);
                        break;
                    default:
                        throw LOGGER.logExceptionAsError(new IllegalArgumentException(
                            "Disposition status not recognized for this test case: " + dispositionStatus));
                }
                return operation.thenReturn(m);

            })
            .block(TIMEOUT);

        assertNotNull(receivedDeferredMessage);
        assertEquals(receivedMessage.getSequenceNumber(), receivedDeferredMessage.getSequenceNumber());

        if (dispositionStatus != DispositionStatus.COMPLETED) {
            messagesPending.decrementAndGet();
        }
    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void sendReceiveMessageWithVariousPropertyTypes(MessagingEntityType entityType) {
        // Arrange
        final boolean isSessionEnabled = true;
        setSender(entityType, TestUtils.USE_CASE_SEND_RECEIVE_WITH_PROPERTIES, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage messageToSend = getMessage(messageId, isSessionEnabled);
        Map<String, Object> sentProperties = messageToSend.getApplicationProperties();
        sentProperties.put("NullProperty", null);
        sentProperties.put("BooleanProperty", true);
        sentProperties.put("ByteProperty", (byte) 1);
        sentProperties.put("ShortProperty", (short) 2);
        sentProperties.put("IntProperty", 3);
        sentProperties.put("LongProperty", 4L);
        sentProperties.put("FloatProperty", 5.5f);
        sentProperties.put("DoubleProperty", 6.6f);
        sentProperties.put("CharProperty", 'z');
        sentProperties.put("UUIDProperty", UUID.fromString("38400000-8cf0-11bd-b23e-10b96e4ef00d"));
        sentProperties.put("StringProperty", "string");

        sendMessage(messageToSend).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_SEND_RECEIVE_WITH_PROPERTIES, isSessionEnabled);

        // Assert & Act
        StepVerifier.create(receiver.receiveMessages().flatMap(receivedMessage -> receiver.complete(receivedMessage).thenReturn(receivedMessage)).take(1))
            .assertNext(receivedMessage -> {
                messagesPending.decrementAndGet();
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);

                final Map<String, Object> received = receivedMessage.getApplicationProperties();

                assertEquals(sentProperties.size(), received.size());

                for (Map.Entry<String, Object> sentEntry : sentProperties.entrySet()) {
                    if (sentEntry.getValue() != null && sentEntry.getValue().getClass().isArray()) {
                        assertArrayEquals((Object[]) sentEntry.getValue(), (Object[]) received.get(sentEntry.getKey()));
                    } else {
                        final Object expected = sentEntry.getValue();
                        final Object actual = received.get(sentEntry.getKey());

                        assertEquals(expected, actual, String.format(
                            "Key '%s' does not match. Expected: '%s'. Actual: '%s'", sentEntry.getKey(), expected,
                            actual));
                    }
                }
            })
            .thenCancel()
            .verify();
    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void setAndGetSessionState(MessagingEntityType entityType) {
        // Arrange
        setSender(entityType, TestUtils.USE_CASE_DEFAULT, true);

        final byte[] sessionState = "Finished".getBytes(UTF_8);
        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage messageToSend = getMessage(messageId, true);

        sendMessage(messageToSend).block(Duration.ofSeconds(10));

        // Act
        setReceiver(entityType, TestUtils.USE_CASE_DEFAULT, true);

        StepVerifier.create(receiver.receiveMessages()
            .flatMap(message -> {
                    LOGGER.info("SessionId: {}. LockToken: {}. LockedUntil: {}. Message received.",
                        message.getSessionId(), message.getLockToken(), message.getLockedUntil());
                    assertMessageEquals(message, messageId, isSessionEnabled);
                    messagesPending.decrementAndGet();
                    return receiver.abandon(message)
                        .then(receiver.setSessionState(sessionState))
                        .then(receiver.getSessionState());
                }
            ).take(1))
            .assertNext(state -> {
                LOGGER.info("State received: {}", new String(state, UTF_8));
                assertArrayEquals(sessionState, state);
            })
            .verifyComplete();
    }

    /**
     * Verifies that we can receive a message from dead letter queue.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityWithSessions")
    @ParameterizedTest
    void receiveFromDeadLetter(MessagingEntityType entityType, boolean isSessionEnabled) {
        // Arrange
        final Duration shortWait = Duration.ofSeconds(2);
        final int entityIndex = 0;

        if (isSessionEnabled && sessionId == null) {
            sessionId = UUID.randomUUID().toString();
        }

        setSender(entityType, entityIndex, isSessionEnabled);

        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);

        sendMessage(message).block();

        setReceiver(entityType, entityIndex, isSessionEnabled);

        receiver.receiveMessages()
            .filter(receivedMessage -> messageId.equals(receivedMessage.getMessageId()))
            .map(receivedMessage -> {
                assertMessageEquals(receivedMessage, messageId, isSessionEnabled);
                messagesPending.decrementAndGet();
                receiver.deadLetter(receivedMessage).block();
                return receivedMessage;
            }).next().block(OPERATION_TIMEOUT);

        final ServiceBusReceiverAsyncClient deadLetterReceiver;
        switch (entityType) {
            case QUEUE:
                final String queueName = isSessionEnabled ? getSessionQueueName(entityIndex) : getQueueName(entityIndex);
                assertNotNull(queueName, "'queueName' cannot be null.");

                deadLetterReceiver = getBuilder(false).receiver()
                    .queueName(queueName)
                    .subQueue(SubQueue.DEAD_LETTER_QUEUE)
                    .buildAsyncClient();
                break;
            case SUBSCRIPTION:
                final String topicName = getTopicName(entityIndex);
                final String subscriptionName = isSessionEnabled ? getSessionSubscriptionBaseName() : getSubscriptionBaseName();
                assertNotNull(topicName, "'topicName' cannot be null.");
                assertNotNull(subscriptionName, "'subscriptionName' cannot be null.");

                deadLetterReceiver = getBuilder(false).receiver()
                    .topicName(topicName)
                    .subscriptionName(subscriptionName)
                    .subQueue(SubQueue.DEAD_LETTER_QUEUE)
                    .buildAsyncClient();
                break;
            default:
                throw LOGGER.logExceptionAsError(new IllegalArgumentException("Unknown entity type: " + entityType));
        }

        // Assert & Act
        try {
            deadLetterReceiver.receiveMessages()
                .filter(serviceBusReceivedMessage -> messageId.equals(serviceBusReceivedMessage.getMessageId()))
                .map(serviceBusReceivedMessage -> {
                    assertMessageEquals(serviceBusReceivedMessage, messageId, isSessionEnabled);
                    return serviceBusReceivedMessage;
                })
                .next()
                .block(OPERATION_TIMEOUT);
        } finally {
            // close dead letter receiver.
            deadLetterReceiver.close();
        }

    }

    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void renewMessageLock(MessagingEntityType entityType) {
        // Arrange
        final boolean isSessionEnabled = false;
        setSenderAndReceiver(entityType, TestUtils.USE_CASE_DEFAULT, isSessionEnabled);

        final Duration maximumDuration = Duration.ofSeconds(35);
        final Duration sleepDuration = maximumDuration.plusMillis(500);
        final String messageId = UUID.randomUUID().toString();
        final ServiceBusMessage message = getMessage(messageId, isSessionEnabled);
        final AtomicInteger numberCompleted = new AtomicInteger(0);

        final ServiceBusReceivedMessage receivedMessage = sendMessage(message)
            .then(receiver.receiveMessages().next())
            .block();
        assertNotNull(receivedMessage);

        final OffsetDateTime lockedUntil = receivedMessage.getLockedUntil();
        assertNotNull(lockedUntil);

        // Assert & Act
        StepVerifier.create(receiver.renewMessageLock(receivedMessage, maximumDuration))
            .thenAwait(sleepDuration)
            .then(() -> receiver.receiveMessages()
                .filter(m -> messageId.equals(m.getMessageId()))
                .flatMap(m -> {
                    LOGGER.info("Completing message.");
                    numberCompleted.addAndGet(completeMessages(receiver, Collections.singletonList(m)));
                    messagesPending.addAndGet(-numberCompleted.get());
                    return Mono.just(m);
                }).subscribe())
            .expectComplete()
            .verify(Duration.ofMinutes(3));
    }

    /**
     * Verifies that we can receive a message which have different section set (i.e header, footer, annotations,
     * application properties etc).
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void receiveAndValidateProperties(MessagingEntityType entityType) {
        // Arrange
        final boolean isSessionEnabled = false;
        final int totalMessages = 1;
        final String subject = "subject";
        final Map<String, Object> footer = new HashMap<>();
        footer.put("footer-key-1", "footer-value-1");
        footer.put("footer-key-2", "footer-value-2");

        final Map<String, Object> applicationProperties = new HashMap<>();
        applicationProperties.put("ap-key-1", "ap-value-1");
        applicationProperties.put("ap-key-2", "ap-value-2");

        final Map<String, Object> deliveryAnnotation = new HashMap<>();
        deliveryAnnotation.put("delivery-annotations-key-1", "delivery-annotations-value-1");
        deliveryAnnotation.put("delivery-annotations-key-2", "delivery-annotations-value-2");

        final String messageId = UUID.randomUUID().toString();
        final AmqpAnnotatedMessage expectedAmqpProperties = new AmqpAnnotatedMessage(
            AmqpMessageBody.fromData(CONTENTS_BYTES));
        expectedAmqpProperties.getProperties().setSubject(subject);
        expectedAmqpProperties.getProperties().setReplyToGroupId("r-gid");
        expectedAmqpProperties.getProperties().setReplyTo(new AmqpAddress("reply-to"));
        expectedAmqpProperties.getProperties().setContentType("content-type");
        expectedAmqpProperties.getProperties().setCorrelationId(new AmqpMessageId("correlation-id"));
        expectedAmqpProperties.getProperties().setTo(new AmqpAddress("to"));
        expectedAmqpProperties.getProperties().setAbsoluteExpiryTime(OffsetDateTime.now().plusSeconds(60));
        expectedAmqpProperties.getProperties().setUserId("user-id-1".getBytes());
        expectedAmqpProperties.getProperties().setContentEncoding("string");
        expectedAmqpProperties.getProperties().setGroupSequence(2L);
        expectedAmqpProperties.getProperties().setCreationTime(OffsetDateTime.now().plusSeconds(30));

        expectedAmqpProperties.getHeader().setPriority((short) 2);
        expectedAmqpProperties.getHeader().setFirstAcquirer(true);
        expectedAmqpProperties.getHeader().setDurable(true);

        expectedAmqpProperties.getFooter().putAll(footer);
        expectedAmqpProperties.getDeliveryAnnotations().putAll(deliveryAnnotation);
        expectedAmqpProperties.getApplicationProperties().putAll(applicationProperties);

        final ServiceBusMessage message = TestUtils.getServiceBusMessage(CONTENTS_BYTES, messageId);

        final AmqpAnnotatedMessage amqpAnnotatedMessage = message.getRawAmqpMessage();
        amqpAnnotatedMessage.getMessageAnnotations().putAll(expectedAmqpProperties.getMessageAnnotations());
        amqpAnnotatedMessage.getApplicationProperties().putAll(expectedAmqpProperties.getApplicationProperties());
        amqpAnnotatedMessage.getDeliveryAnnotations().putAll(expectedAmqpProperties.getDeliveryAnnotations());
        amqpAnnotatedMessage.getFooter().putAll(expectedAmqpProperties.getFooter());

        final AmqpMessageHeader header = amqpAnnotatedMessage.getHeader();
        header.setFirstAcquirer(expectedAmqpProperties.getHeader().isFirstAcquirer());
        header.setTimeToLive(expectedAmqpProperties.getHeader().getTimeToLive());
        header.setDurable(expectedAmqpProperties.getHeader().isDurable());
        header.setDeliveryCount(expectedAmqpProperties.getHeader().getDeliveryCount());
        header.setPriority(expectedAmqpProperties.getHeader().getPriority());

        final AmqpMessageProperties amqpMessageProperties = amqpAnnotatedMessage.getProperties();
        amqpMessageProperties.setReplyTo((expectedAmqpProperties.getProperties().getReplyTo()));
        amqpMessageProperties.setContentEncoding((expectedAmqpProperties.getProperties().getContentEncoding()));
        amqpMessageProperties.setAbsoluteExpiryTime((expectedAmqpProperties.getProperties().getAbsoluteExpiryTime()));
        amqpMessageProperties.setSubject((expectedAmqpProperties.getProperties().getSubject()));
        amqpMessageProperties.setContentType(expectedAmqpProperties.getProperties().getContentType());
        amqpMessageProperties.setCorrelationId(expectedAmqpProperties.getProperties().getCorrelationId());
        amqpMessageProperties.setTo(expectedAmqpProperties.getProperties().getTo());
        amqpMessageProperties.setGroupSequence(expectedAmqpProperties.getProperties().getGroupSequence());
        amqpMessageProperties.setUserId(expectedAmqpProperties.getProperties().getUserId());
        amqpMessageProperties.setAbsoluteExpiryTime(expectedAmqpProperties.getProperties().getAbsoluteExpiryTime());
        amqpMessageProperties.setCreationTime(expectedAmqpProperties.getProperties().getCreationTime());
        amqpMessageProperties.setReplyToGroupId(expectedAmqpProperties.getProperties().getReplyToGroupId());

        setSender(entityType, TestUtils.USE_CASE_VALIDATE_AMQP_PROPERTIES, isSessionEnabled);

        // Send the message
        sendMessage(message).block(TIMEOUT);

        setReceiver(entityType, TestUtils.USE_CASE_VALIDATE_AMQP_PROPERTIES, isSessionEnabled);
        StepVerifier.create(receiver.receiveMessages()/*.take(totalMessages)*/)
            .assertNext(received -> {
                assertNotNull(received.getLockToken());
                AmqpAnnotatedMessage actual = received.getRawAmqpMessage();
                try {
                    assertArrayEquals(CONTENTS_BYTES, message.getBody().toBytes());
                    assertEquals(expectedAmqpProperties.getHeader().getPriority(), actual.getHeader().getPriority());
                    assertEquals(expectedAmqpProperties.getHeader().isFirstAcquirer(), actual.getHeader().isFirstAcquirer());
                    assertEquals(expectedAmqpProperties.getHeader().isDurable(), actual.getHeader().isDurable());

                    assertEquals(expectedAmqpProperties.getProperties().getSubject(), actual.getProperties().getSubject());
                    assertEquals(expectedAmqpProperties.getProperties().getReplyToGroupId(), actual.getProperties().getReplyToGroupId());
                    assertEquals(expectedAmqpProperties.getProperties().getReplyTo(), actual.getProperties().getReplyTo());
                    assertEquals(expectedAmqpProperties.getProperties().getContentType(), actual.getProperties().getContentType());
                    assertEquals(expectedAmqpProperties.getProperties().getCorrelationId(), actual.getProperties().getCorrelationId());
                    assertEquals(expectedAmqpProperties.getProperties().getTo(), actual.getProperties().getTo());
                    assertEquals(expectedAmqpProperties.getProperties().getAbsoluteExpiryTime().toEpochSecond(), actual.getProperties().getAbsoluteExpiryTime().toEpochSecond());
                    assertEquals(expectedAmqpProperties.getProperties().getSubject(), actual.getProperties().getSubject());
                    assertEquals(expectedAmqpProperties.getProperties().getContentEncoding(), actual.getProperties().getContentEncoding());
                    assertEquals(expectedAmqpProperties.getProperties().getGroupSequence(), actual.getProperties().getGroupSequence());
                    assertEquals(expectedAmqpProperties.getProperties().getCreationTime().toEpochSecond(), actual.getProperties().getCreationTime().toEpochSecond());
                    assertArrayEquals(expectedAmqpProperties.getProperties().getUserId(), actual.getProperties().getUserId());

                    assertMapValues(expectedAmqpProperties.getDeliveryAnnotations(), actual.getDeliveryAnnotations());
                    assertMapValues(expectedAmqpProperties.getMessageAnnotations(), actual.getMessageAnnotations());
                    assertMapValues(expectedAmqpProperties.getApplicationProperties(), actual.getApplicationProperties());
                    assertMapValues(expectedAmqpProperties.getFooter(), actual.getFooter());
                } finally {
                    LOGGER.info("Completing message.");
                    receiver.complete(received).block(Duration.ofSeconds(15));
                    messagesPending.decrementAndGet();
                }
            })
            .thenCancel()
            .verify(Duration.ofMinutes(2));
           // .expectComplete();
    }

    /**
     * Verifies we can autocomplete for a queue.
     *
     * @param entityType Entity Type.
     */
    @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider")
    @ParameterizedTest
    void autoComplete(MessagingEntityType entityType) {
        // Arrange
        final Duration shortWait = Duration.ofSeconds(2);
        final int index = TestUtils.USE_CASE_AUTO_COMPLETE;
        setSender(entityType, index, false);

        final int numberOfEvents = 3;
        final String messageId = UUID.randomUUID().toString();
        final List<ServiceBusMessage> messages = getServiceBusMessages(numberOfEvents, messageId);

        setReceiver(entityType, index, false);

        // lastMessage: This is to make sure, if there is left over from previous run.
        final ServiceBusReceivedMessage lastMessage = receiver.peekMessage().block(TIMEOUT);

        // Send messages.
        Mono.when(messages.stream().map(this::sendMessage)
            .collect(Collectors.toList()))
            .block(TIMEOUT);

        final ServiceBusReceiverAsyncClient autoCompleteReceiver =
            getReceiverBuilder(false, entityType, index, false)
                .buildAsyncClient();

        // Act
        // Expecting that as we receive these messages, they'll be completed.
        try {
            StepVerifier.create(autoCompleteReceiver.receiveMessages())
                .assertNext(receivedMessage -> {
                    if (lastMessage != null) {
                        assertEquals(lastMessage.getMessageId(), receivedMessage.getMessageId());
                    } else {
                        assertEquals(messageId, receivedMessage.getMessageId());
                    }
                })
                .assertNext(context -> {
                    if (lastMessage == null) {
                        assertEquals(messageId, context.getMessageId());
                    }
                })
                .assertNext(context -> {
                    if (lastMessage == null) {
                        assertEquals(messageId, context.getMessageId());
                    }
                })
                .thenAwait(shortWait) // Give time for autoComplete to finish
                .thenCancel()
                .verify(TIMEOUT);

        } finally {
            autoCompleteReceiver.close();
        }

        // Assert
        final ServiceBusReceivedMessage newLastMessage = receiver.peekMessage().block(TIMEOUT);
        if (lastMessage == null) {
            assertNull(newLastMessage,
                String.format("Actual messageId[%s]", newLastMessage != null ? newLastMessage.getMessageId() : "n/a"));
        } else {
            assertNotNull(newLastMessage);
            assertEquals(lastMessage.getSequenceNumber(), newLastMessage.getSequenceNumber());
        }
    }

    /**
     * Asserts the length and values with in the map.
     */
    private void assertMapValues(Map<String, Object> expectedMap, Map<String, Object> actualMap) {
        assertTrue(actualMap.size() >= expectedMap.size());
        for (String key : expectedMap.keySet()) {
            assertEquals(expectedMap.get(key), actualMap.get(key), "Value is not equal for Key " + key);
        }
    }

    /**
     * Sets the sender and receiver. If session is enabled, then a single-named session receiver is created.
     */
    private void setSenderAndReceiver(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled) {
        setSender(entityType, entityIndex, isSessionEnabled);
        setReceiver(entityType, entityIndex, isSessionEnabled);
    }

    private void setReceiver(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled) {
        setReceiver(entityType, entityIndex, isSessionEnabled, defaultClientCreationOptions);
    }

    private void setReceiver(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled,
        ClientCreationOptions options) {
        final boolean shareConnection = false;
        final boolean useCredentials = false;
        if (isSessionEnabled) {
            assertNotNull(sessionId, "'sessionId' should have been set.");
            sessionReceiver = getSessionReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .maxAutoLockRenewDuration(options.getMaxAutoLockRenewDuration())
                .disableAutoComplete()
                .buildAsyncClient();

            this.receiver = sessionReceiver.acceptSession(sessionId).block();

        } else {
            this.receiver = getReceiverBuilder(useCredentials, entityType, entityIndex, shareConnection)
                .maxAutoLockRenewDuration(options.getMaxAutoLockRenewDuration())
                .disableAutoComplete()
                .buildAsyncClient();
        }
    }

    private void setSender(MessagingEntityType entityType, int entityIndex, boolean isSessionEnabled) {
        final boolean shareConnection = false;
        final boolean useCredentials = false;
        this.sender = getSenderBuilder(useCredentials, entityType, entityIndex, isSessionEnabled, shareConnection)
            .buildAsyncClient();
    }

    private Mono<Void> sendMessage(ServiceBusMessage message) {
        return sender.sendMessage(message).doOnSuccess(aVoid -> {
            int number = messagesPending.incrementAndGet();
            LOGGER.info("Message Id {}. Number sent: {}", message.getMessageId(), number);
        });
    }

    private int completeMessages(ServiceBusReceiverAsyncClient client, List<ServiceBusReceivedMessage> messages) {
        Mono.when(messages.stream().map(client::complete)
            .collect(Collectors.toList()))
            .block();

        return messages.size();
    }

    /**
     * Class represents various options while creating receiver/sender client.
     */
    public static class ClientCreationOptions {
        Duration maxAutoLockRenewDuration;

        ClientCreationOptions setMaxAutoLockRenewDuration(Duration maxAutoLockRenewDuration) {
            this.maxAutoLockRenewDuration = maxAutoLockRenewDuration;
            return this;
        }

        Duration getMaxAutoLockRenewDuration() {
            return this.maxAutoLockRenewDuration;
        }
    }
}


text:
var test2 = true;

window.onload = function(){

	if( test1 && test2 && test3 ){
		document.body.innerHTML = '
Everything seems in order.
';
	}

};

redpajama_set_name: RedPajamaGithub


text:
require 'yaml'

class MultaArcana
  DEFAULT_SECRET_FILE = 'secrets.yml'
  @@secrets = nil
  @@secret_file = nil

  #
  # Load secrets from the default file (+secrets.yml+, or +config/secrets.yml+
  # for a Rails application) or from the file specified.  The file is sticky;
  # the first call to +secret_for+ sets the file in use, and afterwards all
  # following secret_for calls use the data from a cache.
  #
  def self.secret_for(tag, fname = nil)
    load_file(fname)
    @@secrets[tag.to_s]
  end

  private

  def self.secret_file(fname = nil)
    return @@secret_file if @@secret_file

    @@secret_file ||= fname
    unless @@secret_file
      if defined?Rails
        @@secret_file = Rails.root.join('config', fname || DEFAULT_SECRET_FILE)
      else
        @@secret_file = DEFAULT_SECRET_FILE
      end
    end
    @@secret_file
  end

  def self.load_file(fname = nil)
    @@secrets ||= YAML::load(File.read(secret_file(fname)))
  end
end


text:
jQuery.ajaxSetup({
  'beforeSend': function(xhr) { xhr.setRequestHeader("Accept", "text/javascript"); }
});

var text = ""

$(function(){

  text = $("#textarea").val();

  $(window).resize(function(){
    $("#textarea").css("height", $(window).height() - 55 + "px" );
    $("#textarea").css("width", ($(window).width()/2) - 20 + "px" );
    $("#result").css("height", $(window).height() - 55 + "px" );
    $("#result").css("width", ($(window).width()/2) - 21 + "px" );
  });

  $("#textarea").focus();
  $(window).trigger("resize");

  // hide button
  $("#submit").hide();

  // ajaxify on change
  $("#textarea").keyup(function(){
    if ($(this).val() !== text) {
      text = $(this).val();
      $.post("/virastar", { text: $("#textarea").val() }, function(data){
        $("#result").html(data);
      });
    }
  });

  $("#highlight").change(function(){
    $("#html").toggle();
    $("#plain").toggle();
    $("#options .del").toggleClass("visible");
  });

  $("#show_deleted").change(function(){
    $("del").toggleClass("visible_del");
  });

  $(".close").click(function(){
    $(this).parent().parent().parent().hide();
    return false;
  });

  $(".about").click(function(){
    $("#about").toggle();
    return false;
  });

  $(".bug").click(function(){
    $("#bug").toggle();
    return false;
  });

});


text:
import CalendarLocale from 'rc-calendar/lib/locale/zh_CN';
import TimePickerLocale from '../../time-picker/locale/zh_CN';

// 备注:以下代码无法完全按最初设计运行,但为了保证兼容性,需要保留,直至 antd 默认语言改为英文
//  1. 如果用户不给时间类组件传入 value defaultValue,则运行符合预期
//  2. 如果用户传入 value defaultValue,因为这段代码没有在用户代码之前运行,所以用户调用 moment 时,默认语言依然为英文
// To set the default locale of moment to zh-cn globally.
import moment from 'moment';
import 'moment/locale/zh-cn';
moment.locale('zh-cn');

// Merge into a locale object
const locale = {
  lang: {
    placeholder: '请选择日期',
    rangePlaceholder: ['开始日期', '结束日期'],
    ...CalendarLocale,
  },
  timePickerLocale: {
    ...TimePickerLocale,
  },
};

// should add whitespace between char in Button
locale.lang.ok = '确 定';

// All settings at:
// https://github.com/ant-design/ant-design/blob/master/components/date-picker/locale/example.json

export default locale;


text: Some utils and tools


text:
#ifndef TIIO_MOV_H
#define TIIO_MOV_H

#ifdef __LP64__
#include "tiio_mov_proxy.h"
#else

//#include <Carbon/Carbon.h>
//#include <Movies.h>
//#include <ImageCompression.h>
//#include "QuickTimeComponents.h"

#include <QuickTime/QuickTime.h>

#include "tquicktime.h"
#include "tlevel_io.h"
#include "tthread.h"
#include "tthreadmessage.h"

bool IsQuickTimeInstalled();

class TLevelWriterMov : public TLevelWriter
{

	std::vector<std::pair<int, TimeValue>> m_savedFrames;

public:
	TLevelWriterMov(const TFilePath &path, TPropertyGroup *winfo);
	~TLevelWriterMov();
	TImageWriterP getFrameWriter(TFrameId fid);

	int m_IOError;

	void save(const TImageP &img, int frameIndex);

private:
	Movie m_movie;
	Track m_videoTrack;
	Media m_videoMedia;
	Track m_soundTrack;
	Media m_soundMedia;

	GWorldPtr m_gworld;
	PixMapHandle m_pixmap;
	//Handle       m_compressedData;
	short m_refNum;

	PixelXRGB *buf;
	int buf_lx;
	int buf_ly;
	int m_firstFrame;
	TThread::Mutex m_mutex;
	ComponentInstance m_ci;

public:
	static TLevelWriter *create(const TFilePath &f, TPropertyGroup *winfo)
	{
		return new TLevelWriterMov(f, winfo);
	};
	void saveSoundTrack(TSoundTrack *st);
};

class TLevelReaderMov : public TLevelReader
{
	bool m_readAsToonzOutput;

public:
	TLevelReaderMov(const TFilePath &path);
	~TLevelReaderMov();
	TImageReaderP getFrameReader(TFrameId fid);
	//friend class TImageReaderMov;
	TLevelP loadInfo();

	void enableRandomAccessRead(bool enable) { m_readAsToonzOutput = enable; }
	void load(const TRasterP &rasP, int frameIndex, const TPoint &pos, int shrinkX = 1, int shrinkY = 1);
	TDimension getSize() const { return TDimension(m_lx, m_ly); }
	TRect getBBox() const { return TRect(0, 0, m_lx - 1, m_ly - 1); }

	int m_IOError;

private:
	short m_refNum;
	Movie m_movie;
	short m_resId;
	Track m_track;
	long m_depth;
	map<int, TimeValue> currentTimes;
	int m_lx, m_ly;

public:
	static TLevelReader *create(const TFilePath &f)
	{
		return new TLevelReaderMov(f);
	};
	TThread::Mutex m_mutex;

	TLevelP loadToonzOutputFormatInfo();
};

//------------------------------------------------------------------------------
/*
class TWriterInfoMov : public TWriterInfo {
friend class TLevelWriterMov;
public:
  static TWriterInfo *create(const string &ext);
  ~TWriterInfoMov();
  TWriterInfo *clone() const;
private:
  TWriterInfoMov();

  TWriterInfoMov(const TWriterInfoMov&);

  TWriterInfoMov&operator=(const TWriterInfoMov&);   // not implemented

  typedef map<int, CodecQ> CodecTable;
  typedef map<int, CodecType> QualityTable;
  CodecTable m_codecTable;
  QualityTable m_qualityTable;
};
*/

namespace Tiio
{

class MovWriterProperties : public TPropertyGroup
{
	//  std::map<wstring, CodecType> m_codecMap;
	//  std::map<wstring, CodecQ> m_qualityMap;

public:
	MovWriterProperties();
	//	TPropertyGroup* clone() const;
	//  TEnumProperty m_codec;
	//  TEnumProperty m_quality;
	//	CodecType getCurrentCodec()const;
	//  wstring getCurrentNameFromCodec(CodecType &cCodec)const;
	//  wstring getCurrentQualityFromCodeQ(CodecQ &cCodeQ)const;
	//	CodecQ getCurrentQuality()const;
};
}

#endif //__LP64__

#endif //TIIO_MOV_H


text:
require 'spec_helper'

describe 'parseyaml' do
  it 'exists' do
    is_expected.not_to eq(nil)
  end

  it 'raises an error if called without any arguments' do
    is_expected.to run.with_params
                      .and_raise_error(%r{wrong number of arguments}i)
  end

  context 'with correct YAML data' do
    it 'is able to parse a YAML data with a String' do
      actual_array = ['--- just a string', 'just a string']
      actual_array.each do |actual|
        is_expected.to run.with_params(actual).and_return('just a string')
      end
    end

    it 'is able to parse YAML data with a Hash' do
      is_expected.to run.with_params("---\na: '1'\nb: '2'\n")
                        .and_return('a' => '1', 'b' => '2')
    end

    it 'is able to parse YAML data with an Array' do
      is_expected.to run.with_params("---\n- a\n- b\n- c\n")
                        .and_return(['a', 'b', 'c'])
    end

    it 'is able to parse YAML data with a mixed structure' do
      is_expected.to run.with_params("---\na: '1'\nb: 2\nc:\n  d:\n  - :a\n  - true\n  - false\n")
                        .and_return('a' => '1', 'b' => 2, 'c' => { 'd' => [:a, true, false] })
    end

    it 'is able to parse YAML data with a UTF8 and double byte characters' do
      is_expected.to run.with_params("---\na: ×\nこれ: 記号\nです:\n  ©:\n  - Á\n  - ß\n")
                        .and_return('a' => '×', 'これ' => '記号', 'です' => { '©' => ['Á', 'ß'] })
    end

    it 'does not return the default value if the data was parsed correctly' do
      is_expected.to run.with_params("---\na: '1'\n", 'default_value')
                        .and_return('a' => '1')
    end
  end

  context 'on a modern ruby', :unless => RUBY_VERSION == '1.8.7' do
    it 'raises an error with invalid YAML and no default' do
      is_expected.to run.with_params('["one"')
                        .and_raise_error(Psych::SyntaxError)
    end
  end

  context 'when running on ruby 1.8.7, which does not have Psych', :if => RUBY_VERSION == '1.8.7' do
    it 'raises an error with invalid YAML and no default' do
      is_expected.to run.with_params('["one"')
                        .and_raise_error(ArgumentError)
    end
  end

  context 'with incorrect YAML data' do
    it 'supports a structure for a default value' do
      is_expected.to run.with_params('', 'a' => '1')
                        .and_return('a' => '1')
    end

    [1, 1.2, nil, true, false, [], {}, :yaml].each do |value|
      it "should return the default value for an incorrect #{value.inspect} (#{value.class}) parameter" do
        is_expected.to run.with_params(value, 'default_value')
                          .and_return('default_value')
      end
    end

    context 'when running on modern rubies', :unless => RUBY_VERSION == '1.8.7' do
      ['---', '...', '*8', ''].each do |value|
        it "should return the default value for an incorrect #{value.inspect} string parameter" do
          is_expected.to run.with_params(value, 'default_value')
                            .and_return('default_value')
        end
      end
    end
  end
end


text:
title: Add RadAjaxManager Programmatically
page_title: Add RadAjaxManager Programmatically | RadAjax for ASP.NET AJAX Documentation
description: Add RadAjaxManager Programmatically
slug: ajax/radajaxmanager/how-to/add-radajaxmanager-programmatically
tags: add,radajaxmanager,programmatically
published: True
position: 7
---

# Add RadAjaxManager Programmatically

##

There are some cases where you might need to create the RadAjaxManager control programmatically. This is an easy task, however note this should be done in the **Page_Init** event but later. Below is a sample code you can use:

````C#
protected void Page_Init(object sender, EventArgs e)
{
	RadAjaxManager manager = new RadAjaxManager();
	manager.ID = "RadAjaxManager1";
	this.Page.Form.Controls.Add(manager);
}
````
````VB
Protected Sub Page_Init(ByVal sender As Object, ByVal e As EventArgs) Handles Me.Init
	Dim manager As New RadAjaxManager()
	manager.ID = "RadAjaxManager1"
	Me.Page.Form.Controls.Add(manager)
End Sub
````

Then, to access the RadAjaxManager instance you can call the **GetCurrent()** static method of the RadAjaxManager class:

**Server-Side**

````C#
protected void Page_Load(object sender, EventArgs e)
{
	RadAjaxManager manager = RadAjaxManager.GetCurrent(Page);
}

````
````VB
Protected Sub Page_Load(ByVal sender As Object, ByVal e As EventArgs) Handles Me.Load
	Dim manager As RadAjaxManager = RadAjaxManager.GetCurrent(Page)
End Sub
````

````JavaScript
<telerik:RadCodeBlock ID="RadCodeBlock1" runat="server">
	<script type="text/javascript">
	    function GetAjaxManager() {
	        var manager = $find("<%= RadAjaxManager.GetCurrent(Page).ClientID %>")
	    }
	</script>
</telerik:RadCodeBlock>
````

Additionally, note that if you are creating the ScriptManager dynamically as well, you should add it to the Page.Form.Controls collection before the RadAjaxManager control.


text:
<?php
/**
 * PDO implementing application-level nested transactions.
 *
 * @since 1.0
 * @package Components
 * @author Konstantinos Filios <konfilios@gmail.com>
 */
class CBPdo extends PDO
{
	use TBPdoWithNestedTransactions;
}


text:
<?xml version="1.0" encoding="UTF-8"?>
<project
        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
        xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>com.polyhedral.security.testing</groupId>
		<artifactId>com.polyhedral.security.testing.parent</artifactId>
		<version>0.3.5.RELEASE</version>
		<relativePath>../com.polyhedral.security.testing.parent/pom.xml</relativePath>
	</parent>
	<artifactId>com.polyhedral.security.testing.site</artifactId>
	<packaging>eclipse-repository</packaging>
</project>


text:
package org.codingmatters.value.objects.generation;

import com.squareup.javapoet.*;
import org.codingmatters.value.objects.spec.PropertyCardinality;
import org.codingmatters.value.objects.spec.PropertySpec;

import javax.lang.model.element.Modifier;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;

public class OptionalValue {
    private final ValueConfiguration types;
    private final List<PropertySpec> propertySpecs;
    private final OptionalHelper optionalHelper;

    public OptionalValue(ValueConfiguration types, List<PropertySpec> propertySpecs) {
        this.types = types;
        this.propertySpecs = propertySpecs;
        this.optionalHelper = new OptionalHelper();
    }

    public List<TypeSpec> types() {
        List<TypeSpec> result = new LinkedList<>();

        result.add(TypeSpec.classBuilder(this.types.optionalValueType())
                .addModifiers(Modifier.PUBLIC)
                .addMethod(this.staticConstrucor())
                .addMethod(this.construcor())
                .addFields(this.fields())
                .addMethods(this.optionalGetters())
                .addMethods(this.optionalMethods())
                .build()
        );

        return result;
    }

    private MethodSpec staticConstrucor() {
        return MethodSpec.methodBuilder("of")
                .addModifiers(Modifier.STATIC, Modifier.PUBLIC)
                .addParameter(this.types.valueType(), "value")
                .returns(this.types.optionalValueType())
                .addStatement("return new $T(value)", this.types.optionalValueType())
                .build();
    }

    private MethodSpec construcor() {
        MethodSpec.Builder result = MethodSpec.constructorBuilder()
                .addModifiers(Modifier.PRIVATE)
                .addParameter(this.types.valueType(), "value")
                .addStatement("this.optional = $T.ofNullable(value)", Optional.class);

        for (PropertySpec propertySpec : this.propertySpecs) {
            if (!propertySpec.typeSpec().cardinality().isCollection()) {
                if (!propertySpec.typeSpec().typeKind().isValueObject()) {
                    result.addStatement("this.$L = $T.ofNullable(value != null ? value.$L() : null)",
                            propertySpec.name(),
                            Optional.class,
                            propertySpec.name());
                }
            } else {
                if(propertySpec.typeSpec().cardinality() == PropertyCardinality.LIST) {
                    if (propertySpec.typeSpec().typeKind().isValueObject()) {
                        result.addStatement("this.$L = new $T<>(value != null ? value.$L() : null, e -> $T.of(e))",
                                propertySpec.name(),
                                ClassName.get(this.types.rootPackage() + ".optional", "OptionalValueList"),
                                propertySpec.name(),
                                this.types.propertySingleOptionalType(propertySpec));
                    } else {
                        result.addStatement("this.$L = new $T<>(value != null ? value.$L() : null, e -> $T.ofNullable(e))",
                                propertySpec.name(),
                                ClassName.get(this.types.rootPackage() + ".optional", "OptionalValueList"),
                                propertySpec.name(),
                                Optional.class);
                    }
                } else {
                    result.addStatement("this.$L = new $T<>(value != null ? value.$L() : null)",
                            propertySpec.name(),
                            ClassName.get(this.types.rootPackage() + ".optional", "OptionalValueSet"),
                            propertySpec.name());
                }
            }
        }

        return result
                .build();
    }

    private Iterable<FieldSpec> fields() {
        List<FieldSpec> results = new LinkedList<>();

        results.add(FieldSpec.builder(
                ParameterizedTypeName.get(ClassName.get(Optional.class), this.types.valueType()),
                "optional")
                .addModifiers(Modifier.PRIVATE, Modifier.FINAL)
                .build());

        for (PropertySpec propertySpec : this.propertySpecs) {
            if(! propertySpec.typeSpec().cardinality().isCollection()) {
                if (!propertySpec.typeSpec().typeKind().isValueObject()) {
                    results.add(FieldSpec.builder(
                            ParameterizedTypeName.get(ClassName.get(Optional.class), this.types.propertyType(propertySpec)),
                            propertySpec.name())
                            .addModifiers(Modifier.PRIVATE, Modifier.FINAL)
                            .build());
                } else {
                    results.add(FieldSpec.builder(
                            this.types.propertyOptionalType(propertySpec),
                            propertySpec.name())
                            .addModifiers(Modifier.PRIVATE)
                            .initializer("this.$L", propertySpec.name())
                            .build());
                }
            } else {
                results.add(FieldSpec.builder(
                        this.types.propertyOptionalType(propertySpec),
                        propertySpec.name())
                        .addModifiers(Modifier.PRIVATE, Modifier.FINAL)
                        .build());
            }
        }

        return results;
    }

    private Iterable<MethodSpec> optionalGetters() {
        List<MethodSpec> results = new LinkedList<>();

        for (PropertySpec propertySpec : this.propertySpecs) {
            if(! propertySpec.typeSpec().cardinality().isCollection()) {
                if (!propertySpec.typeSpec().typeKind().isValueObject()) {
                    results.add(MethodSpec.methodBuilder(propertySpec.name())
                            .addModifiers(Modifier.PUBLIC)
                            .returns(ParameterizedTypeName.get(ClassName.get(Optional.class), this.types.propertyType(propertySpec)))
                            .addStatement("return this.$L", propertySpec.name())
                            .build());
                } else {
                    results.add(MethodSpec.methodBuilder(propertySpec.name())
                            .addModifiers(Modifier.PUBLIC, Modifier.SYNCHRONIZED)
                            .returns(this.types.propertyOptionalType(propertySpec))
                            .beginControlFlow("if(this.$L == null)", propertySpec.name())
                            .addStatement("this.$L = $T.of(this.optional.isPresent() ? this.optional.get().$L() : null)",
                                    propertySpec.name(),
                                    this.types.propertyOptionalType(propertySpec),
                                    propertySpec.name()
                            )
                            .endControlFlow()
                            .addStatement("return this.$L", propertySpec.name())
                            .build());
                }
            } else {
                results.add(MethodSpec.methodBuilder(propertySpec.name())
                        .addModifiers(Modifier.PUBLIC)
                        .returns(this.types.propertyOptionalType(propertySpec))
                        .addStatement("return this.$L", propertySpec.name())
                        .build());
            }
        }

        return results;
    }

    private Iterable<MethodSpec> optionalMethods() {
        ClassName valueType = this.types.valueType();
        List<MethodSpec> results = this.optionalHelper.optionalMethods(valueType);
        return results;
    }

}


text:
require 'rails_helper'

RSpec.describe ContributionReport, type: :model do
  skip "add some examples to (or delete) #{__FILE__}"
end


text:
$ErrorActionPreference = 'Stop'

$pp = Get-PackageParameters
$toolsPath = Split-Path -parent $MyInvocation.MyCommand.Definition
$iniFile = Join-Path $toolsPath 'Mp3tagSetup.ini'

$desktop = if ($pp.NoDesktopShortcut) { 0 } else { 1 }
$explorer = if ($pp.NoContextMenu) { 0 } else { 1 }
# Automatic language selection
$LCID = (Get-Culture).LCID
$iniContent = @"
[shortcuts]
startmenu=1
desktop=$desktop
explorer=$explorer

[language]
language=$LCID
"@

# Create the ini file for the installer
New-Item $iniFile -type file -force -value $iniContent

$packageArgs = @{
  packageName    = $env:ChocolateyPackageName
  fileType       = 'exe'
  file           = "$toolsPath\mp3tagv304bsetup.exe"
  silentArgs     = "/S"
  validExitCodes = @(0)
}

Install-ChocolateyInstallPackage @packageArgs

Get-ChildItem $toolsPath\*.exe | ForEach-Object { Remove-Item $_ -ea 0; if (Test-Path $_) { Set-Content "$_.ignore" } }


text:
import { Injectable } from '@angular/core';

import { TranslateService } from '@ngx-translate/core';

import { DataSendService } from 'app/core/core-services/data-send.service';
import { RelationManagerService } from 'app/core/core-services/relation-manager.service';
import { ViewModelStoreService } from 'app/core/core-services/view-model-store.service';
import { RelationDefinition } from 'app/core/definitions/relations';
import { AssignmentVote } from 'app/shared/models/assignments/assignment-vote';
import { ViewAssignmentOption } from 'app/site/assignments/models/view-assignment-option';
import { ViewAssignmentVote } from 'app/site/assignments/models/view-assignment-vote';
import { ViewUser } from 'app/site/users/models/view-user';
import { BaseRepository } from '../base-repository';
import { CollectionStringMapperService } from '../../core-services/collection-string-mapper.service';
import { DataStoreService } from '../../core-services/data-store.service';

const AssignmentVoteRelations: RelationDefinition[] = [
    {
        type: 'M2O',
        ownIdKey: 'user_id',
        ownKey: 'user',
        foreignViewModel: ViewUser
    },
    {
        type: 'M2O',
        ownIdKey: 'option_id',
        ownKey: 'option',
        foreignViewModel: ViewAssignmentOption
    }
];

/**
 * Repository Service for Assignments.
 *
 * Documentation partially provided in {@link BaseRepository}
 */
@Injectable({
    providedIn: 'root'
})
export class AssignmentVoteRepositoryService extends BaseRepository<ViewAssignmentVote, AssignmentVote, object> {
    /**
     * @param DS DataStore access
     * @param dataSend Sending data
     * @param mapperService Map models to object
     * @param viewModelStoreService Access view models
     * @param translate Translate string
     * @param httpService make HTTP Requests
     */
    public constructor(
        DS: DataStoreService,
        dataSend: DataSendService,
        mapperService: CollectionStringMapperService,
        viewModelStoreService: ViewModelStoreService,
        translate: TranslateService,
        relationManager: RelationManagerService
    ) {
        super(
            DS,
            dataSend,
            mapperService,
            viewModelStoreService,
            translate,
            relationManager,
            AssignmentVote,
            AssignmentVoteRelations
        );
    }

    public getTitle = (titleInformation: object) => {
        return 'Vote';
    };

    public getVerboseName = (plural: boolean = false) => {
        return this.translate.instant(plural ? 'Votes' : 'Vote');
    };

    public getVotesForUser(pollId: number, userId: number): ViewAssignmentVote[] {
        return this.getViewModelList().filter(vote => vote.option.poll_id === pollId && vote.user_id === userId);
    }
}


text:
import tempfile
import unittest

from mock import mock

from conans.errors import ConanException
from conans.model.scm import SCM

class SCMDetectRepoTest(unittest.TestCase):

    def setUp(self):
        self.folder = tempfile.gettempdir()
        # Be sure there is no repo in the folder to test
        for name, candidate in SCM.availables.items():
            try:
                candidate(folder=self.folder).check_repo()
            except ConanException:
                pass
            else:
                self.fail("There is a repo of type '{}' in the folder to test".format(name))

    def test_svn(self):
        with mock.patch("conans.client.tools.scm.SVN.check_repo", return_value=None):
            r = SCM.detect_scm(folder=self.folder)
            self.assertEqual(r, "svn")

    def test_git(self):
        with mock.patch("conans.client.tools.scm.Git.check_repo", return_value=None):
            r = SCM.detect_scm(folder=self.folder)
            self.assertEqual(r, "git")

    def test_none(self):
        r = SCM.detect_scm(folder=self.folder)
        self.assertEqual(r, None)


text:
package fixture.owners.scenario;

import org.apache.isis.applib.fixturescripts.FixtureScript;

import fixture.owners.objects.OwnerForKate;
import fixture.owners.objects.OwnerForKyle;
import fixture.owners.objects.OwnerForJordan;
import fixture.owners.objects.OwnerForDeon;
import fixture.pets.PetClinicAppTearDownFixture;
import fixture.pets.objects.PetForLucky;
import fixture.pets.objects.PetForCloud;
import fixture.pets.objects.Puddles;

public class OwnersFixture extends FixtureScript {

    public OwnersFixture() {
        withDiscoverability(Discoverability.DISCOVERABLE);
    }

    @Override
    protected void execute(ExecutionContext executionContext) {

        executionContext.executeChild(this, new PetClinicAppTearDownFixture());

        executionContext.executeChild(this, new PetForCloud());
        executionContext.executeChild(this, new PetForLucky());
        executionContext.executeChild(this, new Puddles());

        executionContext.executeChild(this, new OwnerForKyle());
        executionContext.executeChild(this, new OwnerForKate());
        executionContext.executeChild(this, new OwnerForJordan());
        executionContext.executeChild(this, new OwnerForDeon());
    }

}