/// <summary>
        /// set the access level of specified container
        /// </summary>
        /// <param name="name">container name</param>
        /// <param name="accessLevel">access level in ("off", "blob", "container")</param>
        internal async Task SetContainerAcl(long taskId, IStorageBlobManagement localChannel, string name, BlobContainerPublicAccessType accessLevel)
        {
            if (!NameUtil.IsValidContainerName(name))
            {
                throw new ArgumentException(String.Format(Resources.InvalidContainerName, name));
            }

            BlobContainerPermissions permissions = new BlobContainerPermissions();
            permissions.PublicAccess = accessLevel;

            BlobRequestOptions requestOptions = RequestOptions;
            AccessCondition accessCondition = null;

            CloudBlobContainer container = localChannel.GetContainerReference(name);

            if (!await localChannel.DoesContainerExistAsync(container, requestOptions, OperationContext, CmdletCancellationToken))
            {
                throw new ResourceNotFoundException(String.Format(Resources.ContainerNotFound, name));
            }

            await localChannel.SetContainerPermissionsAsync(container, permissions, accessCondition, requestOptions, OperationContext, CmdletCancellationToken);

            if (PassThru)
            {
                WriteCloudContainerObject(taskId, localChannel, container, permissions);
            }
        }

        /// <summary>
        /// Checks for the existence of a specific storage container, if it doesn't exist it will create it.
        /// </summary>
        /// <param name="client">The <see cref="StorageManagementClient"/> that is performing the operation.</param>
        /// <param name="accountName">The name of the storage account that we want to create the container in.</param>
        /// <param name="containerName">The name of the container that we are about to create.</param>
        /// <param name="publicAccess">The public access level for the container.</param>
        /// <param name="permissions">The set of permissions that the ACL for this container must have.</param>
        /// <returns>The async <see cref="Task"/> wrapper.</returns>
        public static async Task CreateContainerIfNotExistsAsync(
            this StorageManagementClient client,
            string accountName,
            string containerName,
            BlobContainerPublicAccessType publicAccess,
            SharedAccessBlobPermissions permissions)
        {
            Contract.Requires(client != null);
            Contract.Requires(!string.IsNullOrWhiteSpace(accountName));
            Contract.Requires(!string.IsNullOrWhiteSpace(containerName));
            Contract.Requires(containerName.Length >= 3);
            Contract.Requires(containerName.Length <= 63);

            var key = (await client.StorageAccounts.GetKeysAsync(accountName)).PrimaryKey;

            var storageAccount = CloudStorageAccount.Parse($"DefaultEndpointsProtocol=https;AccountName={accountName};AccountKey={key}");
            var container = storageAccount.CreateCloudBlobClient().GetContainerReference(containerName);

            await container.CreateIfNotExistsAsync();
            await container.SetPermissionsAsync(new BlobContainerPermissions { PublicAccess = publicAccess });
            FlexStreams.BuildEventsObserver.OnNext(new ProvisionEvent(AzureResource.StorageContainer, containerName));

            var acl = container.GetSharedAccessSignature(new SharedAccessBlobPolicy { Permissions = permissions });
            FlexStreams.BuildEventsObserver.OnNext(new StorageKeyEvent(accountName, containerName, acl));
        }

        /// <summary>
        /// create a new azure container
        /// </summary>
        /// <param name="name">container name</param>
        internal AzureStorageContainer CreateAzureContainer(string name, BlobContainerPublicAccessType accesslevel)
        {
            if (!NameUtil.IsValidContainerName(name))
            {
                throw new ArgumentException(String.Format(Resources.InvalidContainerName, name));
            }

            BlobRequestOptions requestOptions = RequestOptions;
            AccessCondition accessCondition = null;
            CloudBlobContainer container = Channel.GetContainerReference(name);

            bool created = Channel.CreateContainerIfNotExists(container, requestOptions, OperationContext);

            if (!created)
            {
                throw new ResourceAlreadyExistException(String.Format(Resources.ContainerAlreadyExists, name));
            }

            BlobContainerPermissions permissions = new BlobContainerPermissions();
            permissions.PublicAccess = accessLevel;

            if (accessLevel == BlobContainerPublicAccessType.Container || accessLevel == BlobContainerPublicAccessType.Blob)
            {
                Channel.SetContainerPermissions(container, permissions, accessCondition, requestOptions, OperationContext);
            }
            else
            {
                permissions = Channel.GetContainerPermissions(container, accessCondition, requestOptions, OperationContext);
            }

            AzureStorageContainer azureContainer = new AzureStorageContainer(container, permissions);

            return azureContainer;
        }

        /// <summary>
        /// set the access level of specified container
        /// </summary>
        /// <param name="name">container name</param>
        /// <param name="accessLevel">access level in ("off", "blob", "container")</param>
        internal void SetContainerAcl(string name, BlobContainerPublicAccessType accessLevel)
        {
            if (!NameUtil.IsValidContainerName(name))
            {
                throw new ArgumentException(String.Format(Resources.InvalidContainerName, name));
            }

            BlobContainerPermissions permissions = new BlobContainerPermissions();
            permissions.PublicAccess = accessLevel;

            BlobRequestOptions requestOptions = RequestOptions;
            AccessCondition accessCondition = null;

            CloudBlobContainer container = Channel.GetContainerReference(name);

            if (!Channel.DoesContainerExist(container, requestOptions, OperationContext))
            {
                throw new ResourceNotFoundException(String.Format(Resources.ContainerNotFound, name));
            }

            Channel.SetContainerPermissions(container, permissions, accessCondition, requestOptions, OperationContext);
            AzureStorageContainer azureContainer = new AzureStorageContainer(container, permissions);

            if (PassThru)
            {
                WriteObjectWithStorageContext(azureContainer);
            }
        }

 protected void SetContainerPermission(CloudBlobContainer container, BlobContainerPublicAccessType perimssion)
 {
     container.SetPermissionsAsync(new BlobContainerPermissions
     {
         PublicAccess = perimssion
     });
 }

 public CloudBlobManager(CloudBlobClient blobClient, string containerName, BlobContainerPublicAccessType containerAccess = BlobContainerPublicAccessType.Off)
 {
     BlobClient = blobClient;
     var container = blobClient.GetContainerReference(containerName);
     container.CreateIfNotExists();
     Container = container;
     SetPublicAccess(containerAccess);
 }

 public static async Task<SimpleHttpResponse> CreateContainer(string container, BlobContainerPublicAccessType access = BlobContainerPublicAccessType.Off, IEnumerable<CloudStorageAccount> excludeAccounts = null)
 {
     return await DoForAllContainersAsync(container,
         HttpStatusCode.Created,
         async containerObj => await containerObj.CreateAsync(access, null, null),
         false,
         excludeAccounts);
 }

 public void Create(BlobContainerPublicAccessType accessType, BlobRequestOptions requestOptions = null, OperationContext operationContext = null)
 {
     BlobRequestOptions modifiedOptions = BlobRequestOptions.ApplyDefaults(requestOptions, BlobType.Unspecified, this.ServiceClient);
     Executor.ExecuteSync(
         this.CreateContainerImpl(modifiedOptions, accessType),
         modifiedOptions.RetryPolicy,
         operationContext);
 }

 public void setPublicPermissions(CloudBlobContainer container, BlobContainerPublicAccessType Level)
 {
     container.SetPermissions(
         new BlobContainerPermissions
         {
             PublicAccess = Level
         });
 }

 public IAsyncAction CreateAsync(BlobContainerPublicAccessType accessType, BlobRequestOptions options, OperationContext operationContext)
 {
     BlobRequestOptions modifiedOptions = BlobRequestOptions.ApplyDefaults(options, BlobType.Unspecified, this.ServiceClient);
     return AsyncInfo.Run(async (token) => await Executor.ExecuteAsyncNullReturn(
         this.CreateContainerImpl(modifiedOptions, accessType),
         modifiedOptions.RetryPolicy,
         operationContext,
         token));
 }

        /// <summary>
        /// Constructs a web request to create a new container.
        /// </summary>
        /// <param name="uri">The absolute URI to the container.</param>
        /// <param name="timeout">The server timeout interval.</param>
        /// <param name="accessType">An <see cref="BlobContainerPublicAccessType"/> object that specifies whether data in the container may be accessed publicly and what level of access is to be allowed.</param>                                                                                        
        /// <returns>A web request to use to perform the operation.</returns>
        public static HttpRequestMessage Create(Uri uri, int? timeout, HttpContent content, OperationContext operationContext, BlobContainerPublicAccessType accessType)
        {
            UriQueryBuilder containerBuilder = GetContainerUriQueryBuilder();
            HttpRequestMessage request = HttpRequestMessageFactory.Create(uri, timeout, containerBuilder, content, operationContext);

            if (accessType != BlobContainerPublicAccessType.Off)
            {
                request.Headers.Add(Constants.HeaderConstants.ContainerPublicAccessType, accessType.ToString().ToLower());
            }

            return request;
        }

        /// <summary>
        /// Constructs a web request to create a new container.
        /// </summary>
        /// <param name="uri">A <see cref="System.Uri"/> specifying the absolute URI to the container.</param>
        /// <param name="timeout">An integer specifying the server timeout interval.</param>
        /// <param name="useVersionHeader">A boolean value indicating whether to set the <i>x-ms-version</i> HTTP header.</param>
        /// <param name="operationContext">An <see cref="OperationContext"/> object that represents the context for the current operation.</param>
        /// <param name="accessType">An <see cref="BlobContainerPublicAccessType"/> object that specifies whether data in the container may be accessed publicly and the level of access.</param>                
        /// <returns>A <see cref="System.Net.HttpWebRequest"/> object.</returns>
        public static HttpWebRequest Create(Uri uri, int? timeout, bool useVersionHeader, OperationContext operationContext, BlobContainerPublicAccessType accessType)
        {
            UriQueryBuilder containerBuilder = GetContainerUriQueryBuilder();
            HttpWebRequest request = HttpWebRequestFactory.Create(uri, timeout, containerBuilder, useVersionHeader, operationContext);

            if (accessType != BlobContainerPublicAccessType.Off)
            {
                request.Headers.Add(Constants.HeaderConstants.ContainerPublicAccessType, accessType.ToString().ToLower(CultureInfo.InvariantCulture));
            }

            return request;
        }

        public override void Create(BlobContainerPublicAccessType publicAccessType, Action<CloudOperationResponse<bool>> callback)
        {
            var requestUri = BuildUri(this.Uri, string.Empty, "restype=container&timeout=10000");
            var request = this.ResolveRequest(requestUri);

            request.Method = "PUT";

            // Set the permissions for the specified container.
            request.Headers.AddContainerPublicAccessTypeHeader(publicAccessType);

            // Add metadata for the specified container.
            request.Headers.AddMetadataHeaders(this.Metadata);

            this.SendRequest(request, HttpStatusCode.Created, r => true, callback);
        }

        private static void TestAccessTask(BlobContainerPublicAccessType accessType, CloudBlobContainer container, CloudBlob inputBlob)
        {
            StorageCredentials credentials = new StorageCredentials();
            container = new CloudBlobContainer(container.Uri, credentials);
            CloudPageBlob blob = new CloudPageBlob(inputBlob.Uri, credentials);

            if (accessType.Equals(BlobContainerPublicAccessType.Container))
            {
                blob.FetchAttributesAsync().Wait();
                BlobContinuationToken token = null;
                do
                {
                    BlobResultSegment results = container.ListBlobsSegmented(token);
                    results.Results.ToArray();
                    token = results.ContinuationToken;
                }
                while (token != null);
                container.FetchAttributesAsync().Wait();
            }
            else if (accessType.Equals(BlobContainerPublicAccessType.Blob))
            {
                blob.FetchAttributesAsync().Wait();

                TestHelper.ExpectedExceptionTask(
                    container.ListBlobsSegmentedAsync(null),
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedExceptionTask(
                    container.FetchAttributesAsync(),
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
            else
            {
                TestHelper.ExpectedExceptionTask(
                    blob.FetchAttributesAsync(),
                    "Fetch blob attributes while public access does not allow",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedExceptionTask(
                    container.ListBlobsSegmentedAsync(null),
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedExceptionTask(
                    container.FetchAttributesAsync(),
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
        }

        private static async Task TestAccessAsync(BlobContainerPublicAccessType accessType, CloudBlobContainer container, CloudBlob inputBlob)
        {
            StorageCredentials credentials = new StorageCredentials();
            container = new CloudBlobContainer(container.Uri, credentials);
            CloudPageBlob blob = new CloudPageBlob(inputBlob.Uri, credentials);
            OperationContext context = new OperationContext();
            BlobRequestOptions options = new BlobRequestOptions();


            if (accessType.Equals(BlobContainerPublicAccessType.Container))
            {
                await blob.FetchAttributesAsync();
                await container.ListBlobsSegmentedAsync(null, true, BlobListingDetails.All, null, null, options, context);
                await container.FetchAttributesAsync();
            }
            else if (accessType.Equals(BlobContainerPublicAccessType.Blob))
            {
                await blob.FetchAttributesAsync();
                await TestHelper.ExpectedExceptionAsync(
                    async () => await container.ListBlobsSegmentedAsync(null, true, BlobListingDetails.All, null, null, options, context),
                    context,
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                await TestHelper.ExpectedExceptionAsync(
                    async () => await container.FetchAttributesAsync(null, options, context),
                    context,
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
            else
            {
                await TestHelper.ExpectedExceptionAsync(
                    async () => await blob.FetchAttributesAsync(null, options, context),
                    context,
                    "Fetch blob attributes while public access does not allow",
                    HttpStatusCode.NotFound);
                await TestHelper.ExpectedExceptionAsync(
                    async () => await container.ListBlobsSegmentedAsync(null, true, BlobListingDetails.All, null, null, options, context),
                    context,
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                await TestHelper.ExpectedExceptionAsync(
                    async () => await container.FetchAttributesAsync(null, options, context),
                    context,
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
        }

        public void CreateContainer(string containerName, BlobContainerPublicAccessType publicAccessType, IDictionary<string, string> metadata, Action<CloudOperationResponse<Uri>> callback)
        {
            if (string.IsNullOrWhiteSpace(containerName))
                throw new ArgumentException("The container name cannot be null, empty or white space.", "containerName");

            var requestUri = BuildUri(
                this.sasServiceEndpoint,
                BuilUriPath(ContainersOperation, containerName),
                string.Empty);
            var request = this.ResolveRequest(requestUri);

            request.Method = "PUT";

            // Set the permissions for the specified container.
            request.Headers.AddContainerPublicAccessTypeHeader(publicAccessType);

            // Add metadata for the specified container.
            request.Headers.AddMetadataHeaders(metadata);

            this.SendRequest(request, HttpStatusCode.Created, UriResponseMapper, callback);
        }

        private static void TestAccess(BlobContainerPublicAccessType accessType, CloudBlobContainer container, CloudBlob inputBlob)
        {
            StorageCredentials credentials = new StorageCredentials();
            container = new CloudBlobContainer(container.Uri, credentials);
            CloudPageBlob blob = new CloudPageBlob(inputBlob.Uri, credentials);

            if (accessType.Equals(BlobContainerPublicAccessType.Container))
            {
                blob.FetchAttributes();
                container.ListBlobs().ToArray();
                container.FetchAttributes();
            }
            else if (accessType.Equals(BlobContainerPublicAccessType.Blob))
            {
                blob.FetchAttributes();
                TestHelper.ExpectedException(
                    () => container.ListBlobs().ToArray(),
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedException(
                    () => container.FetchAttributes(),
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
            else
            {
                TestHelper.ExpectedException(
                    () => blob.FetchAttributes(),
                    "Fetch blob attributes while public access does not allow",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedException(
                    () => container.ListBlobs().ToArray(),
                    "List blobs while public access does not allow for listing",
                    HttpStatusCode.NotFound);
                TestHelper.ExpectedException(
                    () => container.FetchAttributes(),
                    "Fetch container attributes while public access does not allow",
                    HttpStatusCode.NotFound);
            }
        }

        public static BlobStorage Create(CloudStorageAccount storageAccount, string containerReference, 
                                         BlobContainerPublicAccessType publicAccessPermission = BlobContainerPublicAccessType.Container)
        {
            try
            {
                // Get blob container
                var blobStorage = storageAccount.CreateCloudBlobClient();
                var container = blobStorage.GetContainerReference(containerReference);
                container.CreateIfNotExists();

                // Set blob container access permission
                var permission = container.GetPermissions();
                permission.PublicAccess = publicAccessPermission;
                container.SetPermissions(permission);

                return new BlobStorage(blobStorage, container);

            }
            catch (Exception exception)
            {
                throw new CloudStorageClientException(exception);
            }
        }

        /// <summary>
        /// create a new azure container
        /// </summary>
        /// <param name="name">container name</param>
        internal async Task CreateAzureContainer(long taskId, IStorageBlobManagement localChannel, string name, BlobContainerPublicAccessType accesslevel)
        {
            if (!NameUtil.IsValidContainerName(name))
            {
                throw new ArgumentException(String.Format(Resources.InvalidContainerName, name));
            }

            BlobRequestOptions requestOptions = RequestOptions;
            CloudBlobContainer container = localChannel.GetContainerReference(name);

            BlobContainerPermissions permissions = new BlobContainerPermissions();

            permissions.PublicAccess = accesslevel;

            bool created = await localChannel.CreateContainerIfNotExistsAsync(container, permissions.PublicAccess, requestOptions, OperationContext, CmdletCancellationToken);

            if (!created)
            {
                throw new ResourceAlreadyExistException(String.Format(Resources.ContainerAlreadyExists, name));
            }

            WriteCloudContainerObject(taskId, localChannel, container, permissions);
        }

 public static AzureBlobStorageFolder CreateContainerIfNotExists(CloudBlobContainer containerReference, BlobContainerPublicAccessType blobContainerPublicAccessType)
 {
     containerReference.CreateIfNotExists(blobContainerPublicAccessType);
     return new AzureBlobStorageFolder(containerReference);
 }