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

package com.azure.storage.blob.specialized;

import com.azure.core.annotation.ReturnType;
import com.azure.core.annotation.ServiceClient;
import com.azure.core.annotation.ServiceMethod;
import com.azure.core.http.HttpPipeline;
import com.azure.core.http.HttpRange;
import com.azure.core.http.HttpResponse;
import com.azure.core.http.RequestConditions;
import com.azure.core.http.rest.PagedFlux;
import com.azure.core.http.rest.PagedResponse;
import com.azure.core.http.rest.PagedResponseBase;
import com.azure.core.http.rest.Response;
import com.azure.core.http.rest.ResponseBase;
import com.azure.core.http.rest.SimpleResponse;
import com.azure.core.util.Context;
import com.azure.core.util.FluxUtil;
import com.azure.core.util.UrlBuilder;
import com.azure.core.util.logging.ClientLogger;
import com.azure.storage.blob.BlobAsyncClient;
import com.azure.storage.blob.BlobServiceVersion;
import com.azure.storage.blob.implementation.models.EncryptionScope;
import com.azure.storage.blob.implementation.models.PageBlobsClearPagesHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsCreateHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsGetPageRangesDiffHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsGetPageRangesHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsResizeHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsUpdateSequenceNumberHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsUploadPagesFromURLHeaders;
import com.azure.storage.blob.implementation.models.PageBlobsUploadPagesHeaders;
import com.azure.storage.blob.implementation.models.PageListHelper;
import com.azure.storage.blob.implementation.util.ModelHelper;
import com.azure.storage.blob.models.BlobHttpHeaders;
import com.azure.storage.blob.models.BlobImmutabilityPolicy;
import com.azure.storage.blob.models.BlobRange;
import com.azure.storage.blob.models.BlobRequestConditions;
import com.azure.storage.blob.models.BlobStorageException;
import com.azure.storage.blob.models.ClearRange;
import com.azure.storage.blob.models.CopyStatusType;
import com.azure.storage.blob.models.CpkInfo;
import com.azure.storage.blob.models.CustomerProvidedKey;
import com.azure.storage.blob.models.PageBlobCopyIncrementalRequestConditions;
import com.azure.storage.blob.models.PageBlobItem;
import com.azure.storage.blob.models.PageBlobRequestConditions;
import com.azure.storage.blob.models.PageList;
import com.azure.storage.blob.models.PageRange;
import com.azure.storage.blob.models.PageRangeItem;
import com.azure.storage.blob.models.SequenceNumberActionType;
import com.azure.storage.blob.options.ListPageRangesDiffOptions;
import com.azure.storage.blob.options.ListPageRangesOptions;
import com.azure.storage.blob.options.PageBlobCopyIncrementalOptions;
import com.azure.storage.blob.options.PageBlobCreateOptions;
import com.azure.storage.blob.options.PageBlobUploadPagesFromUrlOptions;
import com.azure.storage.common.implementation.Constants;
import com.azure.storage.common.implementation.StorageImplUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.MalformedURLException;
import java.net.URL;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static com.azure.core.util.FluxUtil.monoError;
import static com.azure.core.util.FluxUtil.withContext;
import static com.azure.core.util.tracing.Tracer.AZ_TRACING_NAMESPACE_KEY;
import static com.azure.storage.common.Utility.STORAGE_TRACING_NAMESPACE_VALUE;

/**
 * Client to a page blob. It may only be instantiated through a {@link SpecializedBlobClientBuilder} or via the method
 * {@link BlobAsyncClient#getPageBlobAsyncClient()}. This class does not hold any state about a particular blob, but is
 * instead a convenient way of sending appropriate requests to the resource on the service.
 *
 *

 * Please refer to the Azure Docs for more information.
 *
 *

 * Note this client is an async client that returns reactive responses from Spring Reactor Core project
 * (https://projectreactor.io/). Calling the methods in this client will NOT start the actual network
 * operation, until {@code .subscribe()} is called on the reactive response. You can simply convert one of these
 * responses to a {@link java.util.concurrent.CompletableFuture} object through {@link Mono#toFuture()}.
 */
@ServiceClient(builder = SpecializedBlobClientBuilder.class, isAsync = true)
public final class PageBlobAsyncClient extends BlobAsyncClientBase {
    /**
     * Indicates the number of bytes in a page.
     */
    public static final int PAGE_BYTES = 512;

    /**
     * Indicates the maximum number of bytes that may be sent in a call to putPage.
     */
    public static final int MAX_PUT_PAGES_BYTES = 4 * Constants.MB;

    private static final ClientLogger LOGGER = new ClientLogger(PageBlobAsyncClient.class);

    /**
     * Package-private constructor for use by {@link SpecializedBlobClientBuilder}.
     *
     * @param pipeline The pipeline used to send and receive service requests.
     * @param url The endpoint where to send service requests.
     * @param serviceVersion The version of the service to receive requests.
     * @param accountName The storage account name.
     * @param containerName The container name.
     * @param blobName The blob name.
     * @param snapshot The snapshot identifier for the blob, pass {@code null} to interact with the blob directly.
     * @param customerProvidedKey Customer provided key used during encryption of the blob's data on the server, pass
     * {@code null} to allow the service to use its own encryption.
     * @param encryptionScope Encryption scope used during encryption of the blob's data on the server, pass
     * {@code null} to allow the service to use its own encryption.
     * @param versionId The version identifier for the blob, pass {@code null} to interact with the latest blob version.
     */
    PageBlobAsyncClient(HttpPipeline pipeline, String url, BlobServiceVersion serviceVersion,
        String accountName, String containerName, String blobName, String snapshot, CpkInfo customerProvidedKey,
        EncryptionScope encryptionScope, String versionId) {
        super(pipeline, url, serviceVersion, accountName, containerName, blobName, snapshot, customerProvidedKey,
            encryptionScope, versionId);
    }

    /**
     * Creates a new {@link PageBlobAsyncClient} with the specified {@code encryptionScope}.
     *
     * @param encryptionScope the encryption scope for the blob, pass {@code null} to use no encryption scope.
     * @return a {@link PageBlobAsyncClient} with the specified {@code encryptionScope}.
     */
    @Override
    public PageBlobAsyncClient getEncryptionScopeAsyncClient(String encryptionScope) {
        EncryptionScope finalEncryptionScope = null;
        if (encryptionScope != null) {
            finalEncryptionScope = new EncryptionScope().setEncryptionScope(encryptionScope);
        }
        return new PageBlobAsyncClient(getHttpPipeline(), getAccountUrl(), getServiceVersion(), getAccountName(),
            getContainerName(), getBlobName(), getSnapshotId(), getCustomerProvidedKey(), finalEncryptionScope,
            getVersionId());
    }

    /**
     * Creates a new {@link PageBlobAsyncClient} with the specified {@code customerProvidedKey}.
     *
     * @param customerProvidedKey the {@link CustomerProvidedKey} for the blob,
     * pass {@code null} to use no customer provided key.
     * @return a {@link PageBlobAsyncClient} with the specified {@code customerProvidedKey}.
     */
    @Override
    public PageBlobAsyncClient getCustomerProvidedKeyAsyncClient(CustomerProvidedKey customerProvidedKey) {
        CpkInfo finalCustomerProvidedKey = null;
        if (customerProvidedKey != null) {
            finalCustomerProvidedKey = new CpkInfo()
                .setEncryptionKey(customerProvidedKey.getKey())
                .setEncryptionKeySha256(customerProvidedKey.getKeySha256())
                .setEncryptionAlgorithm(customerProvidedKey.getEncryptionAlgorithm());
        }
        return new PageBlobAsyncClient(getHttpPipeline(), getAccountUrl(), getServiceVersion(), getAccountName(),
            getContainerName(), getBlobName(), getSnapshotId(), finalCustomerProvidedKey, encryptionScope,
            getVersionId());
    }

    private static String pageRangeToString(PageRange pageRange) {
        if (pageRange.getStart() < 0 || pageRange.getEnd() <= 0) {
            throw new IllegalArgumentException("PageRange's start and end values must be greater than or equal to "
                + "0 if specified.");
        }
        if (pageRange.getStart() % PAGE_BYTES != 0) {
            throw new IllegalArgumentException("PageRange's start value must be a multiple of 512.");
        }
        if (pageRange.getEnd() % PAGE_BYTES != PAGE_BYTES - 1) {
            throw new IllegalArgumentException("PageRange's end value must be 1 less than a multiple of 512.");
        }
        if (pageRange.getEnd() <= pageRange.getStart()) {
            throw new IllegalArgumentException("PageRange's End value must be after the start.");
        }
        return "bytes=" + pageRange.getStart() + '-' + pageRange.getEnd();
    }

    /**
     * Creates a page blob of the specified length. By default, this method will not overwrite an existing blob.
     * Call PutPage to upload data to a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * client.create(size).subscribe(response -> System.out.printf(
     *     "Created page blob with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param size Specifies the maximum size for the page blob, up to 8 TB. The page blob size must be aligned to a
     * 512-byte boundary.
     *
     * @return A reactive response containing the information of the created page blob.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> create(long size) {
        return create(size, false);
    }

    /**
     * Creates a page blob of the specified length. Call PutPage to upload data to a page blob. For more
     * information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * boolean overwrite = false; // Default behavior
     * client.create(size, overwrite).subscribe(response -> System.out.printf(
     *     "Created page blob with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param size Specifies the maximum size for the page blob, up to 8 TB. The page blob size must be aligned to a
     * 512-byte boundary.
     * @param overwrite Whether to overwrite, should data exist on the blob.
     * @return A reactive response containing the information of the created page blob.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> create(long size, boolean overwrite) {
        BlobRequestConditions blobRequestConditions = new BlobRequestConditions();
        if (!overwrite) {
            blobRequestConditions.setIfNoneMatch(Constants.HeaderConstants.ETAG_WILDCARD);
        }
        return createWithResponse(size, null, null, null, blobRequestConditions).flatMap(FluxUtil::toMono);
    }

    /**
     * Creates a page blob of the specified length. Call PutPage to upload data to a page blob. For more
     * information, see the
     * Azure Docs.
     *

     * To avoid overwriting, pass "*" to {@link BlobRequestConditions#setIfNoneMatch(String)}.
     *
     *
Code Samples

     *
     *
     *

     * BlobHttpHeaders headers = new BlobHttpHeaders()
     *     .setContentLanguage("en-US")
     *     .setContentType("binary");
     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.createWithResponse(size, sequenceNumber, headers, metadata, blobRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Created page blob with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *
     *

     *
     *
     * @param size Specifies the maximum size for the page blob, up to 8 TB. The page blob size must be aligned to a
     * 512-byte boundary.
     * @param sequenceNumber A user-controlled value that you can use to track requests. The value of the sequence
     * number must be between 0 and 2^63 - 1.The default value is 0.
     * @param headers {@link BlobHttpHeaders}
     * @param metadata Metadata to associate with the blob. If there is leading or trailing whitespace in any
     * metadata key or value, it must be removed or encoded.
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response containing the information of the created page blob.
     *
     * @throws IllegalArgumentException If {@code size} isn't a multiple of {@link PageBlobAsyncClient#PAGE_BYTES} or
     * {@code sequenceNumber} isn't null and is less than 0.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> createWithResponse(long size, Long sequenceNumber, BlobHttpHeaders headers,
        Map<String, String> metadata, BlobRequestConditions requestConditions) {
        return this.createWithResponse(new PageBlobCreateOptions(size).setSequenceNumber(sequenceNumber)
            .setHeaders(headers).setMetadata(metadata).setRequestConditions(requestConditions));
    }

    /**
     * Creates a page blob of the specified length. Call PutPage to upload data to a page blob. For more
     * information, see the
     * Azure Docs.
     *

     * To avoid overwriting, pass "*" to {@link BlobRequestConditions#setIfNoneMatch(String)}.
     *
     *
Code Samples

     *
     *
     *

     * BlobHttpHeaders headers = new BlobHttpHeaders()
     *     .setContentLanguage("en-US")
     *     .setContentType("binary");
     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.createWithResponse(new PageBlobCreateOptions(size).setSequenceNumber(sequenceNumber)
     *     .setHeaders(headers).setMetadata(metadata).setTags(tags).setRequestConditions(blobRequestConditions))
     *     .subscribe(response -> System.out.printf(
     *         "Created page blob with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *
     *

     *
     *
     * @param options {@link PageBlobCreateOptions}
     * @return A reactive response containing the information of the created page blob.
     *
     * @throws IllegalArgumentException If {@code size} isn't a multiple of {@link PageBlobAsyncClient#PAGE_BYTES} or
     * {@code sequenceNumber} isn't null and is less than 0.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> createWithResponse(PageBlobCreateOptions options) {
        try {
            return withContext(context -> createWithResponse(options, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> createWithResponse(PageBlobCreateOptions options, Context context) {
        StorageImplUtils.assertNotNull("options", options);
        BlobRequestConditions requestConditions = options.getRequestConditions() == null ? new BlobRequestConditions()
            : options.getRequestConditions();

        if (options.getSize() % PAGE_BYTES != 0) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(
                new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES."));
        }
        if (options.getSequenceNumber() != null && options.getSequenceNumber() < 0) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(
                new IllegalArgumentException("SequenceNumber must be greater than or equal to 0."));
        }
        context = context == null ? Context.NONE : context;
        BlobImmutabilityPolicy immutabilityPolicy = options.getImmutabilityPolicy() == null
            ? new BlobImmutabilityPolicy() : options.getImmutabilityPolicy();

        return this.azureBlobStorage.getPageBlobs().createWithResponseAsync(containerName, blobName, 0,
            options.getSize(), null, null, options.getMetadata(), requestConditions.getLeaseId(),
            requestConditions.getIfModifiedSince(), requestConditions.getIfUnmodifiedSince(),
            requestConditions.getIfMatch(), requestConditions.getIfNoneMatch(), requestConditions.getTagsConditions(),
            options.getSequenceNumber(), null, tagsToString(options.getTags()),
            immutabilityPolicy.getExpiryTime(), immutabilityPolicy.getPolicyMode(), options.isLegalHold(),
            options.getHeaders(), getCustomerProvidedKey(), encryptionScope,
            context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsCreateHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), hd.getContentMD5(),
                    hd.isXMsRequestServerEncrypted(), hd.getXMsEncryptionKeySha256(), hd.getXMsEncryptionScope(),
                    null, hd.getXMsVersionId());
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Creates a page blob of the specified length if it does not exist.
     * Call PutPage to upload data to a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * client.createIfNotExists(size).subscribe(response ->
     *     System.out.printf("Created page blob with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param size Specifies the maximum size for the page blob, up to 8 TB. The page blob size must be aligned to a
     * 512-byte boundary.
     *
     * @return A reactive response {@link Mono} signaling completion. {@link PageBlobItem} contains information of
     * the newly created page blob.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> createIfNotExists(long size) {
        return createIfNotExistsWithResponse(new PageBlobCreateOptions(size)).flatMap(FluxUtil::toMono);
    }

    /**
     * Creates a page blob of the specified length if it does not exist. Call PutPage to upload data to a page blob.
     * For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobHttpHeaders headers = new BlobHttpHeaders()
     *     .setContentLanguage("en-US")
     *     .setContentType("binary");
     *
     * client.createIfNotExistsWithResponse(new PageBlobCreateOptions(size).setSequenceNumber(sequenceNumber)
     *     .setHeaders(headers).setMetadata(metadata).setTags(tags)).subscribe(response -> {
     *         if (response.getStatusCode() == 409) {
     *             System.out.println("Already exists.");
     *         } else {
     *             System.out.println("successfully created.");
     *         }
     *     });
     *

     *
     *
     * @param options {@link PageBlobCreateOptions}
     * @return A {@link Mono} containing {@link Response} signaling completion, whose {@link Response#getValue() value}
     * contains a {@link PageBlobItem} containing information about the page blob. If {@link Response}'s status code is
     * 201, a new page blob was successfully created. If status code is 409, a page blob already existed at this location.
     *
     * @throws IllegalArgumentException If {@code size} isn't a multiple of {@link PageBlobAsyncClient#PAGE_BYTES} or
     * {@code sequenceNumber} isn't null and is less than 0.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> createIfNotExistsWithResponse(PageBlobCreateOptions options) {
        try {
            return createIfNotExistsWithResponse(options, null);
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> createIfNotExistsWithResponse(PageBlobCreateOptions options, Context context) {
        try {
            options.setRequestConditions(new BlobRequestConditions().setIfNoneMatch(Constants.HeaderConstants.ETAG_WILDCARD)
                .setIfNoneMatch(Constants.HeaderConstants.ETAG_WILDCARD));
            return createWithResponse(options, context).onErrorResume(t -> t instanceof BlobStorageException
                && ((BlobStorageException) t).getStatusCode() == 409,
                t -> {
                    HttpResponse response = ((BlobStorageException) t).getResponse();
                    return Mono.just(new SimpleResponse<>(response.getRequest(), response.getStatusCode(),
                        response.getHeaders(), null));
                });
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    /**
     * Writes one or more pages to the page blob. Write size must be a multiple of 512. For more information, see
     * the Azure Docs.
     *

     * Note that the data passed must be replayable if retries are enabled (the default). In other words, the
     * {@code Flux} must produce the same data each time it is subscribed to.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     *
     * client.uploadPages(pageRange, body).subscribe(response -> System.out.printf(
     *     "Uploaded page blob with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param pageRange A {@link PageRange} object. Given that pages must be aligned with 512-byte boundaries, the start
     * offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte ranges
     * are 0-511, 512-1023, etc.
     * @param body The data to upload. Note that this {@code Flux} must be replayable if retries are enabled (the
     * default). In other words, the Flowable must produce the same data each time it is subscribed to.
     *
     * @return A reactive response containing the information of the uploaded pages.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> uploadPages(PageRange pageRange, Flux<ByteBuffer> body) {
        return uploadPagesWithResponse(pageRange, body, null, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Writes one or more pages to the page blob. Write size must be a multiple of 512. For more information, see
     * the Azure Docs.
     *

     * Note that the data passed must be replayable if retries are enabled (the default). In other words, the
     * {@code Flux} must produce the same data each time it is subscribed to.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     *
     * byte[] md5 = MessageDigest.getInstance("MD5").digest("data".getBytes(StandardCharsets.UTF_8));
     * PageBlobRequestConditions pageBlobRequestConditions = new PageBlobRequestConditions().setLeaseId(leaseId);
     *
     * client.uploadPagesWithResponse(pageRange, body, md5, pageBlobRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Uploaded page blob with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param pageRange A {@link PageRange} object. Given that pages must be aligned with 512-byte boundaries, the start
     * offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte ranges
     * are 0-511, 512-1023, etc.
     * @param body The data to upload. Note that this {@code Flux} must be replayable if retries are enabled (the
     * default). In other words, the Flowable must produce the same data each time it is subscribed to.
     * @param contentMd5 An MD5 hash of the page content. This hash is used to verify the integrity of the page during
     * transport. When this header is specified, the storage service compares the hash of the content that has arrived
     * with this header value. Note that this MD5 hash is not stored with the blob. If the two hashes do not match, the
     * operation will fail.
     * @param pageBlobRequestConditions {@link PageBlobRequestConditions}
     * @return A reactive response containing the information of the uploaded pages.
     *
     * @throws IllegalArgumentException If {@code pageRange} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> uploadPagesWithResponse(PageRange pageRange, Flux<ByteBuffer> body,
        byte[] contentMd5, PageBlobRequestConditions pageBlobRequestConditions) {
        try {
            return withContext(context -> uploadPagesWithResponse(pageRange, body, contentMd5,
                pageBlobRequestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> uploadPagesWithResponse(PageRange pageRange, Flux<ByteBuffer> body, byte[] contentMd5,
        PageBlobRequestConditions pageBlobRequestConditions, Context context) {
        pageBlobRequestConditions = pageBlobRequestConditions == null
            ? new PageBlobRequestConditions()
            : pageBlobRequestConditions;

        if (pageRange == null) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(new IllegalArgumentException("pageRange cannot be null."));
        }
        String pageRangeStr = pageRangeToString(pageRange);
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().uploadPagesWithResponseAsync(containerName, blobName,
            pageRange.getEnd() - pageRange.getStart() + 1, body, contentMd5, null, null, pageRangeStr,
            pageBlobRequestConditions.getLeaseId(),
            pageBlobRequestConditions.getIfSequenceNumberLessThanOrEqualTo(),
            pageBlobRequestConditions.getIfSequenceNumberLessThan(),
            pageBlobRequestConditions.getIfSequenceNumberEqualTo(), pageBlobRequestConditions.getIfModifiedSince(),
            pageBlobRequestConditions.getIfUnmodifiedSince(), pageBlobRequestConditions.getIfMatch(),
            pageBlobRequestConditions.getIfNoneMatch(), pageBlobRequestConditions.getTagsConditions(), null,
            getCustomerProvidedKey(), encryptionScope,
            context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsUploadPagesHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), hd.getContentMD5(),
                    hd.isXMsRequestServerEncrypted(), hd.getXMsEncryptionKeySha256(), hd.getXMsEncryptionScope(),
                    hd.getXMsBlobSequenceNumber());
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Writes one or more pages from the source page blob to this page blob. Write size must be a multiple of 512.
     * For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     *
     * client.uploadPagesFromUrl(pageRange, url, sourceOffset)
     *     .subscribe(response -> System.out.printf(
     *         "Uploaded page blob from URL with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param range A {@link PageRange} object. Given that pages must be aligned with 512-byte boundaries, the start
     * offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte ranges
     * are 0-511, 512-1023, etc.
     * @param sourceUrl The url to the blob that will be the source of the copy.  A source blob in the same storage
     * account can be authenticated via Shared Key. However, if the source is a blob in another account, the source blob
     * must either be public or must be authenticated via a shared access signature. If the source blob is public, no
     * authentication is required to perform the operation.
     * @param sourceOffset The source offset to copy from.  Pass null or 0 to copy from the beginning of source page
     * blob.
     *
     * @return A reactive response containing the information of the uploaded pages.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> uploadPagesFromUrl(PageRange range, String sourceUrl, Long sourceOffset) {
        return uploadPagesFromUrlWithResponse(range, sourceUrl, sourceOffset, null, null, null)
            .flatMap(FluxUtil::toMono);
    }

    /**
     * Writes one or more pages from the source page blob to this page blob. Write size must be a multiple of 512.
     * For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     * InputStream dataStream = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
     * byte[] sourceContentMD5 = new byte[512];
     * PageBlobRequestConditions pageBlobRequestConditions = new PageBlobRequestConditions().setLeaseId(leaseId);
     * BlobRequestConditions sourceRequestConditions = new BlobRequestConditions()
     *     .setIfUnmodifiedSince(OffsetDateTime.now().minusDays(3));
     *
     * client.uploadPagesFromUrlWithResponse(pageRange, url, sourceOffset, sourceContentMD5, pageBlobRequestConditions,
     *         sourceRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Uploaded page blob from URL with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param range The destination {@link PageRange} range. Given that pages must be aligned with 512-byte boundaries,
     * the start offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte
     * ranges are 0-511, 512-1023, etc.
     * @param sourceUrl The url to the blob that will be the source of the copy.  A source blob in the same storage
     * account can be authenticated via Shared Key. However, if the source is a blob in another account, the source blob
     * must either be public or must be authenticated via a shared access signature. If the source blob is public, no
     * authentication is required to perform the operation.
     * @param sourceOffset The source offset to copy from.  Pass null or 0 to copy from the beginning of source blob.
     * @param sourceContentMd5 An MD5 hash of the page content. This hash is used to verify the integrity of the page
     * during transport. When this header is specified, the storage service compares the hash of the content that has
     * arrived with this header value. Note that this MD5 hash is not stored with the blob. If the two hashes do not
     * match, the operation will fail.
     * @param destRequestConditions {@link PageBlobRequestConditions}
     * @param sourceRequestConditions {@link BlobRequestConditions}
     * @return A reactive response containing the information of the uploaded pages.
     *
     * @throws IllegalArgumentException If {@code range} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> uploadPagesFromUrlWithResponse(PageRange range, String sourceUrl,
            Long sourceOffset, byte[] sourceContentMd5, PageBlobRequestConditions destRequestConditions,
            BlobRequestConditions sourceRequestConditions) {
        return uploadPagesFromUrlWithResponse(new PageBlobUploadPagesFromUrlOptions(range, sourceUrl)
            .setSourceOffset(sourceOffset).setSourceContentMd5(sourceContentMd5)
            .setDestinationRequestConditions(destRequestConditions)
            .setSourceRequestConditions(sourceRequestConditions));
    }

    /**
     * Writes one or more pages from the source page blob to this page blob. Write size must be a multiple of 512.
     * For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     * InputStream dataStream = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
     * byte[] sourceContentMD5 = new byte[512];
     * PageBlobRequestConditions pageBlobRequestConditions = new PageBlobRequestConditions().setLeaseId(leaseId);
     * BlobRequestConditions sourceRequestConditions = new BlobRequestConditions()
     *     .setIfUnmodifiedSince(OffsetDateTime.now().minusDays(3));
     *
     * client.uploadPagesFromUrlWithResponse(new PageBlobUploadPagesFromUrlOptions(pageRange, url)
     *     .setSourceOffset(sourceOffset).setSourceContentMd5(sourceContentMD5)
     *     .setDestinationRequestConditions(pageBlobRequestConditions)
     *     .setSourceRequestConditions(sourceRequestConditions))
     *     .subscribe(response -> System.out.printf(
     *         "Uploaded page blob from URL with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param options Parameters for the operation.
     * @return A reactive response containing the information of the uploaded pages.
     *
     * @throws IllegalArgumentException If {@code range} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> uploadPagesFromUrlWithResponse(PageBlobUploadPagesFromUrlOptions options) {
        try {
            return withContext(context -> uploadPagesFromUrlWithResponse(options, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> uploadPagesFromUrlWithResponse(PageBlobUploadPagesFromUrlOptions options, Context context) {
        if (options.getRange() == null) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(new IllegalArgumentException("range cannot be null."));
        }

        String rangeString = pageRangeToString(options.getRange());

        long sourceOffset = options.getSourceOffset() == null ? 0L : options.getSourceOffset();

        String sourceRangeString = pageRangeToString(new PageRange()
            .setStart(sourceOffset)
            .setEnd(sourceOffset + (options.getRange().getEnd() - options.getRange().getStart())));

        PageBlobRequestConditions destRequestConditions = (options.getDestinationRequestConditions() == null)
            ? new PageBlobRequestConditions() : options.getDestinationRequestConditions();
        BlobRequestConditions sourceRequestConditions = (options.getSourceRequestConditions() == null)
            ? new BlobRequestConditions() : options.getSourceRequestConditions();

        try {
            new URL(options.getSourceUrl());
        } catch (MalformedURLException ex) {
            throw LOGGER.logExceptionAsError(new IllegalArgumentException("'sourceUrl' is not a valid url.", ex));
        }
        context = context == null ? Context.NONE : context;
        String sourceAuth = options.getSourceAuthorization() == null
            ? null : options.getSourceAuthorization().toString();

        return this.azureBlobStorage.getPageBlobs().uploadPagesFromURLWithResponseAsync(
            containerName, blobName, options.getSourceUrl(), sourceRangeString, 0, rangeString, options.getSourceContentMd5(), null, null,
            destRequestConditions.getLeaseId(), destRequestConditions.getIfSequenceNumberLessThanOrEqualTo(),
            destRequestConditions.getIfSequenceNumberLessThan(), destRequestConditions.getIfSequenceNumberEqualTo(),
            destRequestConditions.getIfModifiedSince(), destRequestConditions.getIfUnmodifiedSince(),
            destRequestConditions.getIfMatch(), destRequestConditions.getIfNoneMatch(),
            destRequestConditions.getTagsConditions(), sourceRequestConditions.getIfModifiedSince(),
            sourceRequestConditions.getIfUnmodifiedSince(), sourceRequestConditions.getIfMatch(),
            sourceRequestConditions.getIfNoneMatch(), null, sourceAuth, getCustomerProvidedKey(),
            encryptionScope, context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsUploadPagesFromURLHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), hd.getContentMD5(),
                    hd.isXMsRequestServerEncrypted(), hd.getXMsEncryptionKeySha256(), hd.getXMsEncryptionScope(), null);
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Frees the specified pages from the page blob. The size of the range must be a multiple of 512. For more
     * information, see the Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     *
     * client.clearPages(pageRange).subscribe(response -> System.out.printf(
     *     "Cleared page blob with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param pageRange A {@link PageRange} object. Given that pages must be aligned with 512-byte boundaries, the start
     * offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte ranges
     * are 0-511, 512-1023, etc.
     *
     * @return A reactive response containing the information of the cleared pages.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> clearPages(PageRange pageRange) {
        return clearPagesWithResponse(pageRange, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Frees the specified pages from the page blob. The size of the range must be a multiple of 512. For more
     * information, see the Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * PageRange pageRange = new PageRange()
     *     .setStart(0)
     *     .setEnd(511);
     * PageBlobRequestConditions pageBlobRequestConditions = new PageBlobRequestConditions().setLeaseId(leaseId);
     *
     * client.clearPagesWithResponse(pageRange, pageBlobRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Cleared page blob with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param pageRange A {@link PageRange} object. Given that pages must be aligned with 512-byte boundaries, the start
     * offset must be a modulus of 512 and the end offset must be a modulus of 512 - 1. Examples of valid byte ranges
     * are 0-511, 512-1023, etc.
     * @param pageBlobRequestConditions {@link PageBlobRequestConditions}
     * @return A reactive response containing the information of the cleared pages.
     *
     * @throws IllegalArgumentException If {@code pageRange} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> clearPagesWithResponse(PageRange pageRange,
        PageBlobRequestConditions pageBlobRequestConditions) {
        try {
            return withContext(context -> clearPagesWithResponse(pageRange, pageBlobRequestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> clearPagesWithResponse(PageRange pageRange,
        PageBlobRequestConditions pageBlobRequestConditions, Context context) {
        pageBlobRequestConditions = pageBlobRequestConditions == null
            ? new PageBlobRequestConditions()
            : pageBlobRequestConditions;
        if (pageRange == null) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(new IllegalArgumentException("pageRange cannot be null."));
        }
        String pageRangeStr = pageRangeToString(pageRange);
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().clearPagesWithResponseAsync(containerName, blobName, 0,
            null, pageRangeStr, pageBlobRequestConditions.getLeaseId(),
            pageBlobRequestConditions.getIfSequenceNumberLessThanOrEqualTo(),
            pageBlobRequestConditions.getIfSequenceNumberLessThan(),
            pageBlobRequestConditions.getIfSequenceNumberEqualTo(), pageBlobRequestConditions.getIfModifiedSince(),
            pageBlobRequestConditions.getIfUnmodifiedSince(), pageBlobRequestConditions.getIfMatch(),
            pageBlobRequestConditions.getIfNoneMatch(), pageBlobRequestConditions.getTagsConditions(), null,
            getCustomerProvidedKey(), encryptionScope,
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsClearPagesHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), hd.getContentMD5(),
                    hd.isXMsRequestServerEncrypted(), hd.getXMsEncryptionKeySha256(), null,
                    hd.getXMsBlobSequenceNumber());
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Returns the list of valid page ranges for a page blob or snapshot of a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     *
     * client.getPageRanges(blobRange).subscribe(response -> {
     *     System.out.println("Valid Page Ranges are:");
     *     for (PageRange pageRange : response.getPageRange()) {
     *         System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *     }
     * });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     *
     * @return A reactive response containing the information of the cleared pages.
     * @deprecated Use {@link #listPageRanges(BlobRange)}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    @Deprecated
    public Mono<PageList> getPageRanges(BlobRange blobRange) {
        return getPageRangesWithResponse(blobRange, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Returns the list of valid page ranges for a page blob or snapshot of a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.getPageRangesWithResponse(blobRange, blobRequestConditions)
     *     .subscribe(response -> {
     *         System.out.println("Valid Page Ranges are:");
     *         for (PageRange pageRange : response.getValue().getPageRange()) {
     *             System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *         }
     *     });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response emitting all the page ranges.
     * @deprecated Use {@link #listPageRanges(ListPageRangesOptions)}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    @Deprecated
    public Mono<Response<PageList>> getPageRangesWithResponse(BlobRange blobRange,
        BlobRequestConditions requestConditions) {
        try {
            return withContext(context -> getPageRangesWithResponse(blobRange, requestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageList>> getPageRangesWithResponse(BlobRange blobRange, BlobRequestConditions requestConditions,
        Context context) {
        blobRange = blobRange == null ? new BlobRange(0) : blobRange;
        requestConditions = requestConditions == null ? new BlobRequestConditions() : requestConditions;
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().getPageRangesWithResponseAsync(containerName, blobName,
                getSnapshotId(), null, blobRange.toHeaderValue(), requestConditions.getLeaseId(),
                requestConditions.getIfModifiedSince(), requestConditions.getIfUnmodifiedSince(),
                requestConditions.getIfMatch(), requestConditions.getIfNoneMatch(),
                requestConditions.getTagsConditions(), null, null, null,
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(response -> new SimpleResponse<>(response.getRequest(), response.getStatusCode(),
                response.getHeaders(), response.getValue()));
    }

    /**
     * Returns the list of valid page ranges for a page blob or snapshot of a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     *
     * System.out.println("Valid Page Ranges are:");
     * client.listPageRanges(blobRange).subscribe(rangeItem -> System.out.printf("Offset: %s, Length: %s%n",
     *     rangeItem.getRange().getOffset(), rangeItem.getRange().getLength()));
     *

     *
     *
     * @param blobRange {@link BlobRange}
     *
     * @return A reactive response containing the information of the cleared pages.
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    public PagedFlux<PageRangeItem> listPageRanges(BlobRange blobRange) {
        return listPageRanges(new ListPageRangesOptions(blobRange));
    }

    /**
     * Returns the list of valid page ranges for a page blob or snapshot of a page blob. For more information, see the
     * Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * ListPageRangesOptions options = new ListPageRangesOptions(new BlobRange(offset))
     *     .setMaxResultsPerPage(1000).setRequestConditions(new BlobRequestConditions().setLeaseId(leaseId));
     *
     * client.listPageRanges(options)
     *     .subscribe(rangeItem -> System.out.printf("Offset: %s, Length: %s%n", rangeItem.getRange().getOffset(),
     *         rangeItem.getRange().getLength()));
     *

     *
     *
     * @param options {@link ListPageRangesOptions}
     * @return A reactive response emitting all the page ranges.
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    public PagedFlux<PageRangeItem> listPageRanges(ListPageRangesOptions options) {
        return new PagedFlux<>(
            pageSize -> withContext(context ->
                listPageRangesWithOptionalTimeout(options, null, context).apply(null, pageSize)),
            (continuationToken, pageSize) -> withContext(context ->
                listPageRangesWithOptionalTimeout(options, null, context).apply(continuationToken, pageSize)));
    }

    /*
     * Implementation for this paged listing operation, supporting an optional timeout provided by the synchronous
     * ContainerClient. Applies the given timeout to each Mono<ContainersListBlobHierarchySegmentResponse> backing the
     * PagedFlux.
     *
     * @param delimiter The delimiter for blob hierarchy, "/" for hierarchy based on directories
     * @param options {@link PageBlobGetPageRangesOptions}
     * @param timeout An optional timeout to be applied to the network asynchronous operations.
     * @return A reactive response emitting the listed blobs, flattened.
     */
    BiFunction<String, Integer, Mono<PagedResponse<PageRangeItem>>> listPageRangesWithOptionalTimeout(
        ListPageRangesOptions options, Duration timeout, Context context) {
        return (marker, pageSize) -> {
            ListPageRangesOptions finalOptions;
            /*
             If pageSize was not set in a .byPage(int) method, the page size from options will be preserved.
             Otherwise, prefer the new value.
             */
            if (pageSize != null) {
                finalOptions = new ListPageRangesOptions(options.getRange()).setMaxResultsPerPage(pageSize);
            } else {
                finalOptions = options;
            }
            return getPageRangesSegment(marker, finalOptions, timeout, context)
                .map(response -> {
                    List<PageRangeItem> value = response.getValue() == null
                        ? Collections.emptyList()
                        : Stream.concat(
                        response.getValue().getPageRange().stream().map(PageBlobAsyncClient::toPageBlobRange),
                        response.getValue().getClearRange().stream().map(PageBlobAsyncClient::toPageBlobRange)
                    ).collect(Collectors.toList());

                    return new PagedResponseBase<>(
                        response.getRequest(),
                        response.getStatusCode(),
                        response.getHeaders(),
                        value,
                        PageListHelper.getNextMarker(response.getValue()),
                        response.getDeserializedHeaders());
                });
        };
    }

    private Mono<ResponseBase<PageBlobsGetPageRangesHeaders, PageList>> getPageRangesSegment(String marker,
        ListPageRangesOptions options, Duration timeout, Context context) {
        BlobRequestConditions requestConditions = options.getRequestConditions() == null ? new BlobRequestConditions()
            : options.getRequestConditions();
        context = context == null ? Context.NONE : context;

        return StorageImplUtils.applyOptionalTimeout(
            this.azureBlobStorage.getPageBlobs().getPageRangesWithResponseAsync(containerName, blobName,
                getSnapshotId(), null, options.getRange().toHeaderValue(), requestConditions.getLeaseId(),
                requestConditions.getIfModifiedSince(), requestConditions.getIfUnmodifiedSince(),
                requestConditions.getIfMatch(), requestConditions.getIfNoneMatch(),
                requestConditions.getTagsConditions(), null, marker, options.getMaxResultsPerPage(),
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE)),
            timeout);
    }

    private static PageRangeItem toPageBlobRange(PageRange range) {
        return new PageRangeItem(new HttpRange(range.getStart(), range.getEnd() - range.getStart() + 1), false);
    }

    private static PageRangeItem toPageBlobRange(ClearRange range) {
        return new PageRangeItem(new HttpRange(range.getStart(), range.getEnd() - range.getStart() + 1), true);
    }

    /**
     * Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * final String prevSnapshot = "previous snapshot";
     *
     * client.getPageRangesDiff(blobRange, prevSnapshot).subscribe(response -> {
     *     System.out.println("Valid Page Ranges are:");
     *     for (PageRange pageRange : response.getPageRange()) {
     *         System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *     }
     * });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param prevSnapshot Specifies that the response will contain only pages that were changed between target blob and
     * previous snapshot. Changed pages include both updated and cleared pages. The target blob may be a snapshot, as
     * long as the snapshot specified by prevsnapshot is the older of the two.
     *
     * @return A reactive response emitting all the different page ranges.
     * @deprecated See {@link #listPageRangesDiff(BlobRange, String)}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    @Deprecated
    public Mono<PageList> getPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
        return getPageRangesDiffWithResponse(blobRange, prevSnapshot, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * final String prevSnapshot = "previous snapshot";
     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.getPageRangesDiffWithResponse(blobRange, prevSnapshot, blobRequestConditions)
     *     .subscribe(response -> {
     *         System.out.println("Valid Page Ranges are:");
     *         for (PageRange pageRange : response.getValue().getPageRange()) {
     *             System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *         }
     *     });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param prevSnapshot Specifies that the response will contain only pages that were changed between target blob and
     * previous snapshot. Changed pages include both updated and cleared pages. The target blob may be a snapshot, as
     * long as the snapshot specified by prevsnapshot is the older of the two.
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response emitting all the different page ranges.
     *
     * @throws IllegalArgumentException If {@code prevSnapshot} is {@code null}
     * @deprecated See {@link #listPageRangesDiff(ListPageRangesDiffOptions)}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    @Deprecated
    public Mono<Response<PageList>> getPageRangesDiffWithResponse(BlobRange blobRange, String prevSnapshot,
        BlobRequestConditions requestConditions) {
        try {
            return withContext(context -> getPageRangesDiffWithResponse(blobRange, prevSnapshot, null,
                requestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    /**
     * Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * String prevSnapshot = "previous snapshot";
     *
     * System.out.println("Valid Page Ranges are:");
     * client.listPageRangesDiff(blobRange, prevSnapshot).subscribe(rangeItem ->
     *     System.out.printf("Offset: %s, Length: %s, isClear: %s%n",
     *     rangeItem.getRange().getOffset(), rangeItem.getRange().getLength(), rangeItem.isClear()));
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param prevSnapshot Specifies that the response will contain only pages that were changed between target blob and
     * previous snapshot. Changed pages include both updated and cleared pages. The target blob may be a snapshot, as
     * long as the snapshot specified by prevsnapshot is the older of the two.
     *
     * @return A reactive response emitting all the different page ranges.
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    public PagedFlux<PageRangeItem> listPageRangesDiff(BlobRange blobRange, String prevSnapshot) {
        return listPageRangesDiff(new ListPageRangesDiffOptions(blobRange, prevSnapshot));
    }

    /**
     * Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * ListPageRangesDiffOptions options = new ListPageRangesDiffOptions(new BlobRange(offset), "previous snapshot")
     *     .setRequestConditions(new BlobRequestConditions().setLeaseId(leaseId))
     *     .setMaxResultsPerPage(1000);
     *
     * client.listPageRangesDiff(options)
     *     .subscribe(rangeItem -> System.out.printf("Offset: %s, Length: %s, isClear: %s%n",
     *         rangeItem.getRange().getOffset(), rangeItem.getRange().getLength(), rangeItem.isClear()));
     *

     *
     *
     * @param options {@link ListPageRangesDiffOptions}.
     * @return A reactive response emitting all the different page ranges.
     *
     * @throws IllegalArgumentException If {@code prevSnapshot} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.COLLECTION)
    public PagedFlux<PageRangeItem> listPageRangesDiff(ListPageRangesDiffOptions options) {
        return new PagedFlux<>(
            pageSize -> withContext(context ->
                listPageRangesDiffWithOptionalTimeout(options, null, context).apply(null, pageSize)),
            (continuationToken, pageSize) -> withContext(context ->
                listPageRangesDiffWithOptionalTimeout(options, null, context).apply(continuationToken, pageSize)));
    }

    /**
     * This API only works for managed disk accounts.
     *
Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.

     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * final String prevSnapshotUrl = "previous snapshot url";
     *
     * client.getPageRangesDiff(blobRange, prevSnapshotUrl).subscribe(response -> {
     *     System.out.println("Valid Page Ranges are:");
     *     for (PageRange pageRange : response.getPageRange()) {
     *         System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *     }
     * });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param prevSnapshotUrl Specifies the URL of a previous snapshot of the target blob. Specifies that the
     * response will contain only pages that were changed between target blob and previous snapshot. Changed pages
     * include both updated and cleared pages. The target blob may be a snapshot, as long as the snapshot specified by
     * prevsnapshot is the older of the two.
     *
     * @return A reactive response emitting all the different page ranges.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageList> getManagedDiskPageRangesDiff(BlobRange blobRange, String prevSnapshotUrl) {
        return getManagedDiskPageRangesDiffWithResponse(blobRange, prevSnapshotUrl, null).flatMap(FluxUtil::toMono);
    }

    /**
     * This API only works for managed disk accounts.
     *
Gets the collection of page ranges that differ between a specified snapshot and this page blob. For more
     * information, see the Azure
     * Docs.

     *
     *
Code Samples

     *
     *
     *

     * BlobRange blobRange = new BlobRange(offset);
     * final String prevSnapshotUrl = "previous snapshot url";
     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.getPageRangesDiffWithResponse(blobRange, prevSnapshotUrl, blobRequestConditions)
     *     .subscribe(response -> {
     *         System.out.println("Valid Page Ranges are:");
     *         for (PageRange pageRange : response.getValue().getPageRange()) {
     *             System.out.printf("Start: %s, End: %s%n", pageRange.getStart(), pageRange.getEnd());
     *         }
     *     });
     *

     *
     *
     * @param blobRange {@link BlobRange}
     * @param prevSnapshotUrl Specifies the URL of a previous snapshot of the target blob. Specifies that the
     * response will contain only pages that were changed between target blob and previous snapshot. Changed pages
     * include both updated and cleared pages. The target blob may be a snapshot, as long as the snapshot specified by
     * prevsnapshot is the older of the two.
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response emitting all the different page ranges.
     *
     * @throws IllegalArgumentException If {@code prevSnapshot} is {@code null}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageList>> getManagedDiskPageRangesDiffWithResponse(BlobRange blobRange,
        String prevSnapshotUrl, BlobRequestConditions requestConditions) {
        try {
            return withContext(context ->
                getPageRangesDiffWithResponse(blobRange, null, prevSnapshotUrl, requestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageList>> getPageRangesDiffWithResponse(BlobRange blobRange, String prevSnapshot,
        String prevSnapshotUrl, BlobRequestConditions requestConditions, Context context) {
        blobRange = blobRange == null ? new BlobRange(0) : blobRange;
        requestConditions = requestConditions == null ? new BlobRequestConditions() : requestConditions;

        if (prevSnapshotUrl == null && prevSnapshot == null) {
            throw LOGGER.logExceptionAsError(new IllegalArgumentException("prevSnapshot cannot be null"));
        }
        if (prevSnapshotUrl != null) {
            try {
                new URL(prevSnapshotUrl);
            } catch (MalformedURLException ex) {
                throw LOGGER.logExceptionAsError(new IllegalArgumentException("'prevSnapshotUrl' is not a valid url.", ex));
            }
        }
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().getPageRangesDiffWithResponseAsync(containerName, blobName,
                getSnapshotId(), null, prevSnapshot, prevSnapshotUrl, blobRange.toHeaderValue(),
                requestConditions.getLeaseId(), requestConditions.getIfModifiedSince(),
                requestConditions.getIfUnmodifiedSince(), requestConditions.getIfMatch(),
                requestConditions.getIfNoneMatch(), requestConditions.getTagsConditions(), null, null, null,
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(response -> new SimpleResponse<>(response.getRequest(), response.getStatusCode(),
                response.getHeaders(), response.getValue()));
    }

    /*
     * Implementation for this paged listing operation, supporting an optional timeout provided by the synchronous
     * ContainerClient. Applies the given timeout to each Mono<ContainersListBlobHierarchySegmentResponse> backing the
     * PagedFlux.
     *
     * @param delimiter The delimiter for blob hierarchy, "/" for hierarchy based on directories
     * @param options {@link PageBlobGetPageRangesDiffOptions}
     * @param timeout An optional timeout to be applied to the network asynchronous operations.
     * @return A reactive response emitting the listed blobs, flattened.
     */
    BiFunction<String, Integer, Mono<PagedResponse<PageRangeItem>>> listPageRangesDiffWithOptionalTimeout(ListPageRangesDiffOptions options,
        Duration timeout, Context context) {
        return (marker, pageSize) -> {
            ListPageRangesDiffOptions finalOptions;
            /*
             If pageSize was not set in a .byPage(int) method, the page size from options will be preserved.
             Otherwise, prefer the new value.
             */
            if (pageSize != null) {
                finalOptions =
                    new ListPageRangesDiffOptions(options.getRange(), options.getPreviousSnapshot())
                        .setRequestConditions(options.getRequestConditions()).setMaxResultsPerPage(pageSize);
            } else {
                finalOptions = options;
            }

            return getPageRangesDiffSegment(marker, finalOptions, timeout, context)
                .map(response -> {
                    List<PageRangeItem> value = response.getValue() == null
                        ? Collections.emptyList()
                        : Stream.concat(
                            response.getValue().getPageRange().stream().map(PageBlobAsyncClient::toPageBlobRange),
                            response.getValue().getClearRange().stream().map(PageBlobAsyncClient::toPageBlobRange))
                        .collect(Collectors.toList());

                    return new PagedResponseBase<>(
                        response.getRequest(),
                        response.getStatusCode(),
                        response.getHeaders(),
                        value,
                        PageListHelper.getNextMarker(response.getValue()),
                        response.getDeserializedHeaders());
                });
        };
    }

    private Mono<ResponseBase<PageBlobsGetPageRangesDiffHeaders, PageList>> getPageRangesDiffSegment(String marker,
        ListPageRangesDiffOptions options, Duration timeout, Context context) {
        BlobRequestConditions requestConditions = options.getRequestConditions() == null ? new BlobRequestConditions()
            : options.getRequestConditions();
        context = context == null ? Context.NONE : context;

        return StorageImplUtils.applyOptionalTimeout(
            this.azureBlobStorage.getPageBlobs().getPageRangesDiffWithResponseAsync(containerName, blobName,
                getSnapshotId(), null, options.getPreviousSnapshot(), null,
                options.getRange().toHeaderValue(), requestConditions.getLeaseId(),
                requestConditions.getIfModifiedSince(), requestConditions.getIfUnmodifiedSince(),
                requestConditions.getIfMatch(), requestConditions.getIfNoneMatch(),
                requestConditions.getTagsConditions(), null, marker, options.getMaxResultsPerPage(),
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE)),
            timeout);
    }

    /**
     * Resizes the page blob to the specified size (which must be a multiple of 512). For more information, see the Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * client.resize(size).subscribe(response -> System.out.printf(
     *     "Page blob resized with sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param size Resizes a page blob to the specified size. If the specified value is less than the current size of
     * the blob, then all pages above the specified value are cleared.
     *
     * @return A reactive response emitting the resized page blob.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> resize(long size) {
        return resizeWithResponse(size, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Resizes the page blob to the specified size (which must be a multiple of 512). For more information, see the Azure Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.resizeWithResponse(size, blobRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Page blob resized with sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param size Resizes a page blob to the specified size. If the specified value is less than the current size of
     * the blob, then all pages above the specified value are cleared.
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response emitting the resized page blob.
     *
     * @throws IllegalArgumentException If {@code size} isn't a multiple of {@link PageBlobAsyncClient#PAGE_BYTES}
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> resizeWithResponse(long size, BlobRequestConditions requestConditions) {
        try {
            return withContext(context -> resizeWithResponse(size, requestConditions, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> resizeWithResponse(long size, BlobRequestConditions requestConditions,
        Context context) {
        if (size % PAGE_BYTES != 0) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(
                new IllegalArgumentException("size must be a multiple of PageBlobAsyncClient.PAGE_BYTES."));
        }
        requestConditions = requestConditions == null ? new BlobRequestConditions() : requestConditions;
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().resizeWithResponseAsync(containerName, blobName, size, null,
            requestConditions.getLeaseId(), requestConditions.getIfModifiedSince(),
            requestConditions.getIfUnmodifiedSince(), requestConditions.getIfMatch(),
            requestConditions.getIfNoneMatch(), requestConditions.getTagsConditions(), null,
            getCustomerProvidedKey(), encryptionScope,
            context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsResizeHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), null, null, null, null,
                    hd.getXMsBlobSequenceNumber());
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Sets the page blob's sequence number. For more information, see the
     * Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * client.updateSequenceNumber(SequenceNumberActionType.INCREMENT, size)
     *     .subscribe(response -> System.out.printf(
     *         "Page blob updated to sequence number %s%n", response.getBlobSequenceNumber()));
     *

     *
     *
     * @param action Indicates how the service should modify the blob's sequence number.
     * @param sequenceNumber The blob's sequence number. The sequence number is a user-controlled property that you can
     * use to track requests and manage concurrency issues.
     *
     * @return A reactive response emitting the updated page blob.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<PageBlobItem> updateSequenceNumber(SequenceNumberActionType action, Long sequenceNumber) {
        return updateSequenceNumberWithResponse(action, sequenceNumber, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Sets the page blob's sequence number. For more information, see the Azure
     * Docs.
     *
     *
Code Samples

     *
     *
     *

     * BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setLeaseId(leaseId);
     *
     * client.updateSequenceNumberWithResponse(SequenceNumberActionType.INCREMENT, size, blobRequestConditions)
     *     .subscribe(response -> System.out.printf(
     *         "Page blob updated to sequence number %s%n", response.getValue().getBlobSequenceNumber()));
     *

     *
     *
     * @param action Indicates how the service should modify the blob's sequence number.
     * @param sequenceNumber The blob's sequence number. The sequence number is a user-controlled property that you can
     * use to track requests and manage concurrency issues.
     * @param requestConditions {@link BlobRequestConditions}
     * @return A reactive response emitting the updated page blob.
     *
     * @throws IllegalArgumentException If {@code sequenceNumber} isn't null and is less than 0
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<PageBlobItem>> updateSequenceNumberWithResponse(SequenceNumberActionType action,
        Long sequenceNumber, BlobRequestConditions requestConditions) {
        try {
            return withContext(context -> updateSequenceNumberWithResponse(action, sequenceNumber, requestConditions,
                context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<PageBlobItem>> updateSequenceNumberWithResponse(SequenceNumberActionType action, Long sequenceNumber,
        BlobRequestConditions requestConditions, Context context) {
        if (sequenceNumber != null && sequenceNumber < 0) {
            // Throwing is preferred to Single.error because this will error out immediately instead of waiting until
            // subscription.
            throw LOGGER.logExceptionAsError(
                new IllegalArgumentException("SequenceNumber must be greater than or equal to 0."));
        }
        requestConditions = requestConditions == null ? new BlobRequestConditions() : requestConditions;
        sequenceNumber = action == SequenceNumberActionType.INCREMENT ? null : sequenceNumber;
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().updateSequenceNumberWithResponseAsync(containerName, blobName, action, null,
            requestConditions.getLeaseId(), requestConditions.getIfModifiedSince(),
            requestConditions.getIfUnmodifiedSince(), requestConditions.getIfMatch(),
            requestConditions.getIfNoneMatch(), requestConditions.getTagsConditions(), sequenceNumber, null,
            context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> {
                PageBlobsUpdateSequenceNumberHeaders hd = rb.getDeserializedHeaders();
                PageBlobItem item = new PageBlobItem(hd.getETag(), hd.getLastModified(), null, null, null, null,
                    hd.getXMsBlobSequenceNumber());
                return new SimpleResponse<>(rb, item);
            });
    }

    /**
     * Begins an operation to start an incremental copy from one page blob's snapshot to this page blob. The snapshot
     * is
     * copied such that only the differential changes between the previously copied snapshot are transferred to the
     * destination. The copied snapshots are complete copies of the original snapshot and can be read or copied from as
     * usual. For more information, see the Azure Docs
     * here
     * and
     * here.
     *
     *
Code Samples

     *
     *
     *

     * final String snapshot = "copy snapshot";
     * client.copyIncremental(url, snapshot).subscribe(statusType -> {
     *     switch (statusType) {
     *         case SUCCESS:
     *             System.out.println("Page blob copied successfully");
     *             break;
     *         case FAILED:
     *             System.out.println("Page blob copied failed");
     *             break;
     *         case ABORTED:
     *             System.out.println("Page blob copied aborted");
     *             break;
     *         case PENDING:
     *             System.out.println("Page blob copied pending");
     *             break;
     *         default:
     *             break;
     *     }
     * });
     *

     *
     *
     * @param source The source page blob.
     * @param snapshot The snapshot on the copy source.
     *
     * @return A reactive response emitting the copy status.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<CopyStatusType> copyIncremental(String source, String snapshot) {
        return copyIncrementalWithResponse(source, snapshot, null).flatMap(FluxUtil::toMono);
    }

    /**
     * Begins an operation to start an incremental copy from one page blob's snapshot to this page blob. The snapshot
     * is
     * copied such that only the differential changes between the previously copied snapshot are transferred to the
     * destination. The copied snapshots are complete copies of the original snapshot and can be read or copied from as
     * usual. For more information, see the Azure Docs
     * here
     * and
     * here.
     *
     *
Code Samples

     *
     *
     *

     * final String snapshot = "copy snapshot";
     * RequestConditions modifiedRequestConditions = new RequestConditions()
     *     .setIfNoneMatch("snapshotMatch");
     *
     * client.copyIncrementalWithResponse(url, snapshot, modifiedRequestConditions)
     *     .subscribe(response -> {
     *         CopyStatusType statusType = response.getValue();
     *
     *         switch (statusType) {
     *             case SUCCESS:
     *                 System.out.println("Page blob copied successfully");
     *                 break;
     *             case FAILED:
     *                 System.out.println("Page blob copied failed");
     *                 break;
     *             case ABORTED:
     *                 System.out.println("Page blob copied aborted");
     *                 break;
     *             case PENDING:
     *                 System.out.println("Page blob copied pending");
     *                 break;
     *             default:
     *                 break;
     *         }
     *     });
     *

     *
     *
     * @param source The source page blob.
     * @param snapshot The snapshot on the copy source.
     * @param modifiedRequestConditions Standard HTTP Access conditions related to the modification of data. ETag and
     * LastModifiedTime are used to construct conditions related to when the blob was changed relative to the given
     * request. The request will fail if the specified condition is not satisfied.
     *
     * @return A reactive response emitting the copy status.
     *
     * @throws IllegalStateException If {@code source} and {@code snapshot} form a malformed URL.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<CopyStatusType>> copyIncrementalWithResponse(String source, String snapshot,
        RequestConditions modifiedRequestConditions) {
        return copyIncrementalWithResponse(new PageBlobCopyIncrementalOptions(source, snapshot)
                .setRequestConditions(ModelHelper.populateBlobDestinationRequestConditions(modifiedRequestConditions)));
    }

    /**
     * Begins an operation to start an incremental copy from one page blob's snapshot to this page blob. The snapshot
     * is
     * copied such that only the differential changes between the previously copied snapshot are transferred to the
     * destination. The copied snapshots are complete copies of the original snapshot and can be read or copied from as
     * usual. For more information, see the Azure Docs
     * here
     * and
     * here.
     *
     *
Code Samples

     *
     *
     *

     * final String snapshot = "copy snapshot";
     * PageBlobCopyIncrementalRequestConditions destinationRequestConditions = new PageBlobCopyIncrementalRequestConditions()
     *     .setIfNoneMatch("snapshotMatch");
     *
     * client.copyIncrementalWithResponse(new PageBlobCopyIncrementalOptions(url, snapshot)
     *     .setRequestConditions(destinationRequestConditions))
     *     .subscribe(response -> {
     *         CopyStatusType statusType = response.getValue();
     *
     *         switch (statusType) {
     *             case SUCCESS:
     *                 System.out.println("Page blob copied successfully");
     *                 break;
     *             case FAILED:
     *                 System.out.println("Page blob copied failed");
     *                 break;
     *             case ABORTED:
     *                 System.out.println("Page blob copied aborted");
     *                 break;
     *             case PENDING:
     *                 System.out.println("Page blob copied pending");
     *                 break;
     *             default:
     *                 break;
     *         }
     *     });
     *

     *
     *
     * @param options {@link PageBlobCopyIncrementalOptions}
     *
     * @return A reactive response emitting the copy status.
     *
     * @throws IllegalStateException If {@code source} and {@code snapshot} form a malformed URL.
     */
    @ServiceMethod(returns = ReturnType.SINGLE)
    public Mono<Response<CopyStatusType>> copyIncrementalWithResponse(PageBlobCopyIncrementalOptions options) {
        try {
            return withContext(context -> copyIncrementalWithResponse(options, context));
        } catch (RuntimeException ex) {
            return monoError(LOGGER, ex);
        }
    }

    Mono<Response<CopyStatusType>> copyIncrementalWithResponse(PageBlobCopyIncrementalOptions options,
        Context context) {
        StorageImplUtils.assertNotNull("options", options);
        UrlBuilder builder = UrlBuilder.parse(options.getSource());
        builder.setQueryParameter(Constants.UrlConstants.SNAPSHOT_QUERY_PARAMETER, options.getSnapshot());
        PageBlobCopyIncrementalRequestConditions modifiedRequestConditions = (options.getRequestConditions() == null)
            ? new PageBlobCopyIncrementalRequestConditions() : options.getRequestConditions();

        try {
            builder.toUrl();
        } catch (MalformedURLException e) {
            // We are parsing a valid url and adding a query parameter. If this fails, we can't recover.
            throw LOGGER.logExceptionAsError(new IllegalArgumentException(e));
        }
        context = context == null ? Context.NONE : context;

        return this.azureBlobStorage.getPageBlobs().copyIncrementalWithResponseAsync(containerName, blobName,
                builder.toString(), null, modifiedRequestConditions.getIfModifiedSince(),
                modifiedRequestConditions.getIfUnmodifiedSince(), modifiedRequestConditions.getIfMatch(),
                modifiedRequestConditions.getIfNoneMatch(), modifiedRequestConditions.getTagsConditions(), null,
                context.addData(AZ_TRACING_NAMESPACE_KEY, STORAGE_TRACING_NAMESPACE_VALUE))
            .map(rb -> new SimpleResponse<>(rb, rb.getDeserializedHeaders().getXMsCopyStatus()));
    }
}


text:
<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="com.cavillum.pirategame.android"
    android:versionCode="10"
    android:versionName="0.9.4" >

    <uses-sdk android:minSdkVersion="9" android:targetSdkVersion="23" />

    <uses-permission android:name="android.permission.INTERNET" />
	<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />

    <application
        android:allowBackup="true"
        android:icon="@drawable/ic_launcher"
        android:label="@string/app_name"
        android:theme="@style/GdxTheme" >
        <activity
            android:name="com.cavillum.pirategame.android.AndroidLauncher"
            android:label="@string/app_name"
            android:screenOrientation="portrait"
            android:configChanges="keyboard|keyboardHidden|orientation|screenSize">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />
                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>

        <activity android:name="com.nextpeer.android.NextpeerActivity" />

        <!-- GPGS Stuff -->
        <meta-data android:name="com.google.android.gms.games.APP_ID"
            android:value="@string/app_id" />
		<meta-data android:name="com.google.android.gms.version"
		    android:value="@integer/google_play_services_version" />

		 <!--Include the AdActivity configChanges and theme. -->
		 <activity
		     android:name="com.google.android.gms.ads.AdActivity"
		     android:configChanges="keyboard|keyboardHidden|orientation|screenLayout|uiMode|screenSize|smallestScreenSize"
		     android:theme="@android:style/Theme.Translucent" />

    </application>

</manifest>

redpajama_set_name: RedPajamaGithub


text:
\hypertarget{physical__communicator_8hpp}{}\section{lib/physical\+\_\+communicator.hpp File Reference}
\label{physical__communicator_8hpp}\index{lib/physical\+\_\+communicator.\+hpp@{lib/physical\+\_\+communicator.\+hpp}}
{\ttfamily \#include $<$functional$>$}\newline
{\ttfamily \#include \char`\"{}logical\+\_\+address.\+h\char`\"{}}\newline
{\ttfamily \#include \char`\"{}socket/client\+Socket.\+hpp\char`\"{}}\newline
{\ttfamily \#include \char`\"{}socket/server\+Socket.\+hpp\char`\"{}}\newline
{\ttfamily \#include \char`\"{}spa\+\_\+message.\+h\char`\"{}}\newline
{\ttfamily \#include $<$iostream$>$}\newline
\subsection*{Classes}
\begin{DoxyCompactItemize}
\item
class \hyperlink{classPhysicalCommunicator}{Physical\+Communicator}
\end{DoxyCompactItemize}


text:
echo "****MEMECARD****"

if shopt -q  login_shell ; then

	if [ ! -f "$HOME"/.pagekicker/config.txt ]; then
		echo "config file not found, creating /home/<user>/.pagekicker, put config file there"
		mkdir -p -m 755 "$HOME"/.pagekicker
		echo "exiting"
		exit 1
	else
		. "$HOME"/.pagekicker/config.txt
		echo "read config file from login shell $HOME""/.pagekicker/config.txt"
	fi
else
	. /home/$(whoami)/.pagekicker/config.txt #hard-coding /home is a hack
	echo "read config file from nonlogin shell /home/$(whoami)/.pagekicker/config.txt"
fi

cd $scriptpath

TMPDIR="/tmp/pagekicker/"
uuid="memecard"  && mkdir -p $TMPDIR"$uuid"
confdir="$confdir"
memewidth=1200
memeheight=630
memebackgroundcolor="white"
memefillcolor="black"
memeheadlinefont="DejaVu-Sans-Bold"

. includes/set-variables.sh

echo "version is" $SFB_VERSION

while :
do
case $1 in
--help | -\?)
echo "for help review source code for now"
exit 0  # This is not an error, the user requested help, so do not exit status 1.
;;
-U|--passuuid)
passuuid=$2
shift 2
;;
--passuuid=*)
passuuid=${1#*=}
shift
;;
-f|--infile)
infile=$2
shift 2
;;
--infile=*)
infile=${1#*=}
shift
;;
-t|--tldr)
tldr=$2
shift 2
;;
--tldr=*)
tldr=${1#*=}
shift
;;
-F|--font)
font=$2
shift 2
;;
--font=*)
font=${1#*=}
shift
;;
-w|--memewidth)
memewidth=$2
shift 2
;;
--memewidth=*)
memewidth=${1#*=}
shift
;;
-h|--memeheight)
memeheight=$2
shift 2
;;
--memeheight=*)
memeheight=${1#*=}
shift
;;
  --) # End of all options
            shift
            break
            ;;
        -*)
            echo "WARN: Unknown option (ignored): $1" >&2
            shift
            ;;
        *)  # no more options. Stop while loop
            break
            ;;

esac
done

if [ ! "$passuuid" ] ; then
	echo "creating uuid"
	uuid=$("$PYTHON_BIN"  -c 'import uuid; print(uuid.uuid1())')
	echo "uuid is" $uuid | tee --append $xform_log
	mkdir -p -m 777  "$TMPDIR"$uuid
else
	uuid=$passuuid
	echo "received uuid " $uuid
	mkdir -p -m 777  "$TMPDIR"$uuid
fi

cp "$infile" $TMPDIR$uuid/tmp.md

# create title bar & version label
convert -units pixelsperinch -density 300 -size 1000x50 -border 5 -background "$backgroundcolor" -font "$memeheadlinefont" -fill "$memefillcolor" -background "$memebackgroundcolor" -gravity center  caption:"$tldr" $TMPDIR$uuid/toplabel1.png
convert -units pixelsperinch -density 300 -size 500x25 -font "Utopia" -background white -fill black -gravity southeast caption:"$SFB_VERSION" $TMPDIR$uuid/version.png

# prepend pagenumbering to tmp file

echo -e '\pagenumbering{gobble}\n' | cat - $TMPDIR$uuid/"tmp.md" > $TMPDIR$uuid/out && mv $TMPDIR$uuid/out $TMPDIR$uuid/"tmp.md"

# make pdf

# make pdf

cat $TMPDIR$uuid/"tmp.md"  | \
 pandoc  --latex-engine=xelatex --template=$confdir"pandoc_templates/nonumtemplate.tex" \
-o $TMPDIR$uuid/memecard.pdf
# -V "geometry:paperheight=5.0in" -V "geometry:paperwidth=7.0in"

# make png of text
convert -density 400  $TMPDIR$uuid/memecard.pdf  -trim $TMPDIR$uuid/memecard.png
# "if error issued here see comments in includes/1000x3000skyscraper.sh for explanation"

# lay logo & version onto bottom label

convert $TMPDIR$uuid/memecard.png -border 30 $TMPDIR$uuid/memecard.png
# put logo on 1000 px wide & trim
convert $scriptpath"assets/pk35pc.jpg" -resize 20% $TMPDIR$uuid/pksmall.jpg
convert $TMPDIR$uuid"/pksmall.jpg" -gravity west -background white -extent 1024x50 \
$TMPDIR$uuid/memecardlogo.png
 convert -gravity center $TMPDIR$uuid/memecardlogo.png -gravity southeast $TMPDIR$uuid/version.png -composite   $TMPDIR$uuid/bottom.png

# lay image of text onto card background
montage $TMPDIR$uuid/toplabel1.png \
$TMPDIR$uuid"/memecard.png" \
$TMPDIR$uuid/bottom.png  \
-geometry "$memewidth"x"$memeheight" -border 10 -tile 1x10 -mode concatenate \
$TMPDIR$uuid/memecard.png

# create card for delivery to desitnation

convert $TMPDIR$uuid"/memecard.png" -trim -border 30 $TMPDIR$uuid/memecard_delivery.png

echo "built memecard and delivered it to $TMPDIR$uuid/memecard_delivery.png"


text:
{-# OPTIONS -XImpredicativeTypes -fno-warn-deprecated-flags #-}

module ShouldFail where

import Control.Concurrent

-- Attempt to put a polymorphic value in an MVar
-- Fails, but the error message is worth keeping an eye on
--
-- Actually (Dec 06) it succeeds now
--
-- In GHC 7.0 it fails again! (and rightly so)

foo = do var <- newEmptyMVar :: IO (MVar (forall a. Show a => a -> String))
         putMVar var (show :: forall b. Show b => b -> String)


text:
package com.ircclouds.irc.api.utils;

import java.io.*;
import java.net.*;
import java.util.*;

public class NetUtils
{
	public static int getRandDccPort()
	{
		return getRandPortIn(1024, 5000);
	}

	public static int getRandPortIn(int aMin, int aMax)
	{
		if (aMin >= aMax)
		{
			throw new RuntimeException("Please provide a valid [Min-Max] range.");
		}

		Random _rand = new Random();
		int _port = _rand.nextInt(aMax - aMin) + aMin;
		while (!NetUtils.available(_port))
		{
			_port = _rand.nextInt(aMax - aMin) + aMin;
		}

		return _port;
	}

	public static boolean available(int aPort)
	{
		ServerSocket _ss = null;
		try
		{
			_ss = new ServerSocket(aPort);
			_ss.setReuseAddress(true);

			return true;
		}
		catch (IOException aExc)
		{
		}
		finally
		{
			if (_ss != null)
			{
				try
				{
					_ss.close();
				}
				catch (IOException aExc)
				{
				}
			}
		}

		return false;
	}
}


text:
require "spec_helper"

describe Cobrato::Entities::PaymentConfig do
  let(:attributes) do
    {
      id: 1,
      bank_account_id: 1,
      name: "Configuração de pagamento",
      timezone: "Brasilia"
    }
  end

  subject { described_class.new(attributes) }

  it_behaves_like "entity_attributes", [:id, :name, :bank_account_id, :timezone]
end


text:
package org.osgi.service.tr069todmt;

/**
 * Maps to the TR-069 {@code ParameterValueStruct}
 */
public interface ParameterValue {

	/**
	 * This is the path of a Parameter. In TR-069 this is called the Parameter
	 * Name.
	 *
	 * @return The path of the parameter
	 */
	String getPath();

	/**
	 * This is the value of the parameter. The returned value must be in a
	 * representation defined by the TR-069 protocol.
	 *
	 * @return The value of the parameter
	 */
	public String getValue();

	/**
	 * The type of the parameter. One of {@link TR069Connector#TR069_INT},
	 * {@link TR069Connector#TR069_UNSIGNED_INT},
	 * {@link TR069Connector#TR069_LONG},
	 * {@link TR069Connector#TR069_UNSIGNED_LONG},
	 * {@link TR069Connector#TR069_STRING},
	 * {@link TR069Connector#TR069_DATETIME},
	 * {@link TR069Connector#TR069_BASE64},
	 * {@link TR069Connector#TR069_HEXBINARY}. This method is not part of the
	 * {@code ParameterValueStruct} but is necessary to encode the type in the
	 * XML.
	 *
	 * @return The parameter type
	 */
	public int getType();

}


text:
Follow {{ include.feed_name }}

      {% include icon/feedly.svg %}

        {% assign feed_path = site.url | append: include.rss_path | url_encode %}
        {% assign feedly_url = 'https://feedly.com/i/subscription/feed%2F' | append: feed_path %}
        via Feedly

      {% include icon/rss.svg %}

        via Atom RSS

    {% if include.twitter_url %}

        {% include icon/twitter.svg %}

          via {{ include.twitter_text }}

    {% endif %}
    {% if include.microblog_url %}

        {% include icon/microblog.svg %}

          via Micro.blog

    {% endif %}


text:
package com.oydnatech.jpipe;

import com.oydnatech.jpipe.impl.AsyncUniPipeline;
import com.oydnatech.jpipe.impl.lambda.*;

import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;

/**
 * Created by Trond Øydna on 04.06.2016.
 */
public interface UniPipe<O> {

    void setListener(UniListener<? super O> listener);

    default <R> UniPipeline<? super O, R> then(UniPipeline<? super O, R> next) {
        setListener(next);

        return next;
    }

    default <RL, RR> UniToBiPipeline<? super O, RL, RR> thenBi(UniToBiPipeline<? super O, RL, RR> next) {
        setListener(next);

        return next;
    }

    default void consume(UniListener<? super O> end) {
        setListener(end);
    }

    default <R> UniPipeline<? super O, R> map(Function<? super O, R> function) {
        return then(new PipelineFunction<>(function));
    }

    default <R> UniToBiPipeline<? super O, O, R> supplyRight(Supplier<R> supplier) {
        return thenBi(new PipelineRightSupplier<>(supplier));
    }

    default UniPipeline<? super O, O> action(Consumer<? super O> action) {
        return then(new PipelineAction<>(action));
    }

    default UniPipeline<? super O, O> filter(Predicate<? super O> filter) {
        return then(new PipelinePredicate<>(filter));
    }

    default void consume(Consumer<? super O> consumer) {
        consume(new PipelineConsumer<>(consumer));
    }

    default UniPipeline<? super O, O> async() {
        AsyncUniPipeline<O> async = new AsyncUniPipeline<>();

        then(async);

        return async;
    }

}