/// <summary>
        /// Asynchronous method that delays execution until the specified pool reaches the specified state.
        /// </summary>
        /// <param name="client">A fully intitialized <see cref="BatchClient"/>.</param>
        /// <param name="poolId">The ID of the pool to monitor for the specified <see cref="AllocationState"/>.</param>
        /// <param name="targetAllocationState">The allocation state to monitor.</param>
        /// <param name="timeout">The maximum time to wait for the pool to reach the specified state.</param>
        /// <returns>A <see cref="System.Threading.Tasks.Task"/> object that represents the asynchronous operation.</returns>
        public static async Task WaitForPoolToReachStateAsync(BatchClient client, string poolId, AllocationState targetAllocationState, TimeSpan timeout)
        {
            Console.WriteLine("Waiting for pool {0} to reach allocation state {1}", poolId, targetAllocationState);

            DateTime startTime = DateTime.UtcNow;
            DateTime timeoutAfterThisTimeUtc = startTime.Add(timeout);

            ODATADetailLevel detail = new ODATADetailLevel(selectClause: "id,allocationState");
            CloudPool pool = await client.PoolOperations.GetPoolAsync(poolId, detail);

            while (pool.AllocationState != targetAllocationState)
            {
                Console.Write(".");

                await Task.Delay(TimeSpan.FromSeconds(10));
                await pool.RefreshAsync(detail);

                if (DateTime.UtcNow > timeoutAfterThisTimeUtc)
                {
                    Console.WriteLine();
                    Console.WriteLine("Timed out waiting for pool {0} to reach state {1}", poolId, targetAllocationState);

                    return;
                }
            }

            Console.WriteLine();
        }

 static void DeletePool(BatchClient client)
 {
     client.PoolOperations.DeletePool("testpool1");
     Console.WriteLine("Pool was deleted.");
     Console.WriteLine("Press Enter to continue.");
     Console.ReadLine();
 }

        /// <summary>
        /// Waits for all tasks under the specified job to complete and then prints each task's output to the console.
        /// </summary>
        /// <param name="batchClient">The BatchClient to use when interacting with the Batch service.</param>
        /// <param name="tasks">The tasks to wait for.</param>
        /// <param name="timeout">The timeout.  After this time has elapsed if the job is not complete and exception will be thrown.</param>
        /// <returns>An asynchronous <see cref="Task"/> representing the operation.</returns>
        public static async Task WaitForTasksAndPrintOutputAsync(BatchClient batchClient, IEnumerable<CloudTask> tasks, TimeSpan timeout)
        {
            // We use the task state monitor to monitor the state of our tasks -- in this case we will wait for them all to complete.
            TaskStateMonitor taskStateMonitor = batchClient.Utilities.CreateTaskStateMonitor();

            // Wait until the tasks are in completed state.
            List<CloudTask> ourTasks = tasks.ToList();

            bool timedOut = await taskStateMonitor.WaitAllAsync(ourTasks, TaskState.Completed, timeout);

            if (timedOut)
            {
                throw new TimeoutException("Timed out waiting for tasks");
            }

            // dump task output
            foreach (CloudTask t in ourTasks)
            {
                Console.WriteLine("Task {0}", t.Id);

                //Read the standard out of the task
                NodeFile standardOutFile = await t.GetNodeFileAsync(Constants.StandardOutFileName);
                string standardOutText = await standardOutFile.ReadAsStringAsync();
                Console.WriteLine("Standard out:");
                Console.WriteLine(standardOutText);

                //Read the standard error of the task
                NodeFile standardErrorFile = await t.GetNodeFileAsync(Constants.StandardErrorFileName);
                string standardErrorText = await standardErrorFile.ReadAsStringAsync();
                Console.WriteLine("Standard error:");
                Console.WriteLine(standardErrorText);

                Console.WriteLine();
            }
        }

        /// <summary>
        /// Prints task information to the console for each of the nodes in the specified pool.
        /// </summary>
        /// <param name="poolId">The ID of the <see cref="CloudPool"/> containing the nodes whose task information should be printed to the console.</param>
        /// <returns>A <see cref="System.Threading.Tasks.Task"/> object that represents the asynchronous operation.</returns>
        public static async Task PrintNodeTasksAsync(BatchClient batchClient, string poolId)
        {
            Console.WriteLine("Listing Node Tasks");
            Console.WriteLine("==================");

            ODATADetailLevel nodeDetail = new ODATADetailLevel(selectClause: "id,recentTasks");
            IPagedEnumerable<ComputeNode> nodes = batchClient.PoolOperations.ListComputeNodes(poolId, nodeDetail);
            
            await nodes.ForEachAsync(node =>
            {
                Console.WriteLine();
                Console.WriteLine(node.Id + " tasks:");

                if (node.RecentTasks != null && node.RecentTasks.Any())
                {
                    foreach (TaskInformation task in node.RecentTasks)
                    {
                        Console.WriteLine("\t{0}: {1}", task.TaskId, task.TaskState);
                    }
                }
                else
                {
                    // No tasks found for the node
                    Console.WriteLine("\tNone");
                }
            }).ConfigureAwait(continueOnCapturedContext: false);

            Console.WriteLine("==================");
        }

 private static void DeleteJob(BatchClient client)
 {
     client.JobOperations.DeleteJob("testjob1");
     Console.WriteLine("Job was deleted.");
     Console.WriteLine("Press Enter to continue.");
     Console.ReadLine();
 }

        public BatchService(BatchSharedKeyCredential credentials)
        {
            this.Client = BatchClient.Open(credentials);
            this.Credentials = credentials;
            this.retryPolicy = new LinearRetry(TimeSpan.FromSeconds(10), 5);

            this.Client.CustomBehaviors.Add(new RetryPolicyProvider(this.retryPolicy));
        }

 private static void DeleteTasks(BatchClient client)
 {
     CloudJob job = client.JobOperations.GetJob("testjob1");
     foreach (CloudTask task in job.ListTasks())
     {
         task.Delete();
     }
     Console.WriteLine("All tasks deleted.");
     Console.WriteLine("Press Enter to continue.");
     Console.ReadLine();
 }

 private static void ListTasks(BatchClient client)
 {
     IPagedEnumerable<CloudTask> tasks = client.JobOperations.ListTasks("testjob1");
     foreach (CloudTask task in tasks)
     {
         Console.WriteLine("Task id: " + task.Id);
         Console.WriteLine("   Task status: " + task.State);
         Console.WriteLine("   Task start: " + task.ExecutionInformation.StartTime);
     }
     Console.ReadLine();
 }

        /// <summary>
        /// Lists all the jobs in the Batch account.
        /// </summary>
        /// <param name="batchClient">The BatchClient to use when interacting with the Batch service.</param>
        /// <returns>An asynchronous <see cref="Task"/> representing the operation.</returns>
        public static async Task PrintJobsAsync(BatchClient batchClient)
        {
            Console.WriteLine("Listing Jobs");
            Console.WriteLine("============");

            IPagedEnumerable<CloudJob> jobs = batchClient.JobOperations.ListJobs(new ODATADetailLevel(selectClause: "id,state"));
            await jobs.ForEachAsync(job =>
            {
                Console.WriteLine("State of job " + job.Id + " is " + job.State);
            });

            Console.WriteLine("============");
        }

        /// <summary>
        /// Lists all the pools in the Batch account.
        /// </summary>
        /// <param name="batchClient">The BatchClient to use when interacting with the Batch service.</param>
        /// <returns>An asynchronous <see cref="Task"/> representing the operation.</returns>
        public static async Task PrintPoolsAsync(BatchClient batchClient)
        {
            Console.WriteLine("Listing Pools");
            Console.WriteLine("=============");

            // Using optional select clause to return only the properties of interest. Makes query faster and reduces HTTP packet size impact
            IPagedEnumerable<CloudPool> pools = batchClient.PoolOperations.ListPools(new ODATADetailLevel(selectClause: "id,state,currentDedicated,vmSize"));

            await pools.ForEachAsync(pool =>
            {
                Console.WriteLine("State of pool {0} is {1} and it has {2} nodes of size {3}", pool.Id, pool.State, pool.CurrentDedicated, pool.VirtualMachineSize);
            });
            Console.WriteLine("=============");
        }

        /// <summary>
        /// Creates a CloudJob in the specified pool if a job with the specified ID is not found
        /// in the pool, otherwise returns the existing job.
        /// </summary>
        /// <param name="batchClient">A fully initialized <see cref="BatchClient"/>.</param>
        /// <param name="poolId">The ID of the CloudPool in which the job should be created.</param>
        /// <param name="jobId">The ID of the CloudJob.</param>
        /// <returns>A bound version of the newly created CloudJob.</returns>
        public static async Task<CloudJob> CreateJobAsync(BatchClient batchClient, string poolId, string jobId)
        {
            CloudJob job = await SampleHelpers.GetJobIfExistAsync(batchClient, jobId);

            if (job == null)
            {
                Console.WriteLine("Job {0} not found, creating...", jobId);

                CloudJob unboundJob = batchClient.JobOperations.CreateJob(jobId, new PoolInformation() { PoolId = poolId });
                await unboundJob.CommitAsync();

                // Get the bound version of the job with all of its properties populated
                job = await batchClient.JobOperations.GetJobAsync(jobId);
            }

            return job;
        }

        /// <summary>
        /// Creates a <see cref="CloudPool"/> associated with the specified Batch account. If an existing pool with the
        /// specified ID is found, the pool is resized to match the specified node count.
        /// </summary>
        /// <param name="batchClient">A fully initialized <see cref="BatchClient"/>.</param>
        /// <param name="poolId">The ID of the <see cref="CloudPool"/>.</param>
        /// <param name="nodeSize">The size of the nodes within the pool.</param>
        /// <param name="nodeCount">The number of nodes to create within the pool.</param>
        /// <param name="maxTasksPerNode">The maximum number of tasks to run concurrently on each node.</param>
        /// <returns>A bound <see cref="CloudPool"/> with the specified properties.</returns>
        public async static Task<CloudPool> CreatePoolIfNotExistAsync(BatchClient batchClient, string poolId, string nodeSize, int nodeCount, int maxTasksPerNode)
        {
            // Create and configure an unbound pool with the specified ID
            CloudPool pool = batchClient.PoolOperations.CreatePool(poolId: poolId,
                                                                   osFamily: "3",
                                                                   virtualMachineSize: nodeSize,
                                                                   targetDedicated: nodeCount);
            
            pool.MaxTasksPerComputeNode = maxTasksPerNode;

            // We want each node to be completely filled with tasks (i.e. up to maxTasksPerNode) before
            // tasks are assigned to the next node in the pool
            pool.TaskSchedulingPolicy = new TaskSchedulingPolicy(ComputeNodeFillType.Pack);
            
            await GettingStartedCommon.CreatePoolIfNotExistAsync(batchClient, pool).ConfigureAwait(continueOnCapturedContext: false);

            return await batchClient.PoolOperations.GetPoolAsync(poolId).ConfigureAwait(continueOnCapturedContext: false);
        }

        static void AddTasks(BatchClient client)
        {
            CloudJob job = client.JobOperations.GetJob("testjob1");
            ResourceFile programFile = new ResourceFile(
                "https://mystorage00.blob.core.windows.net/testcon1/ProcessTaskData.exe",
                "ProcessTaskData.exe"
                );
            ResourceFile assemblyFile = new ResourceFile(
                  "https://mystorage00.blob.core.windows.net/testcon1/Microsoft.WindowsAzure.Storage.dll",
                  "Microsoft.WindowsAzure.Storage.dll"
                  );
            for (int i = 1; i < 4; ++i)
            {
                string blobName = "taskdata" + i;
                string taskName = "mytask" + i;
                ResourceFile taskData = new ResourceFile("https://mystorage00.blob.core.windows.net/testcon1/" +
                  blobName, blobName);
                CloudTask task = new CloudTask(
                    taskName,
                    "ProcessTaskData.exe https://mystorage00.blob.core.windows.net/testcon1/" +
                    blobName + " 3");
                List<ResourceFile> taskFiles = new List<ResourceFile>();
                taskFiles.Add(taskData);
                taskFiles.Add(programFile);
                taskFiles.Add(assemblyFile);
                task.ResourceFiles = taskFiles;
                job.AddTask(task);
                job.Commit();
                job.Refresh();
            }

            client.Utilities.CreateTaskStateMonitor().WaitAll(job.ListTasks(),
            TaskState.Completed, new TimeSpan(0, 30, 0));
            Console.WriteLine("The tasks completed successfully.");
            foreach (CloudTask task in job.ListTasks())
            {
                Console.WriteLine("Task " + task.Id + " says:\n" + task.GetNodeFile(Constants.StandardOutFileName).ReadAsString());
            }
            Console.WriteLine("Press Enter to continue.");
            Console.ReadLine();
        }

        private static void CreatePoolIfNeeded(BatchClient client, string poolId)
        {
            // go through all the pools and see if the specified pool already exists
            bool found = false;

            // use an OData based select clause to limit the amount of data returned. This will result in incomplete
            // client objects but reduces the amount of data on the wire leading to faster completion when there are
            // a lot of objects for a given query. No spaces are allowed in the string and property names are case-sensitive.
            foreach (CloudPool p in client.PoolOperations.ListPools(new ODATADetailLevel(selectClause: "id,currentDedicated")))
            {
                // pools are uniquely identified by their ID
                if (string.Equals(p.Id, poolId))
                {
                    Console.WriteLine("Using existing pool {0}", poolId);
                    found = true;

                    if (p.CurrentDedicated == 0)
                    {
                        Console.WriteLine("There are no compute nodes in this pool. No tasks will be run until at least one node has been added via resizing.");
                        Console.WriteLine("Resizing pool to add 3 nodes. This might take a while...");
                        p.Resize(3);
                    }

                    break;
                }
            }

            if (!found)
            {
                Console.WriteLine("Creating pool: {0}", poolId);
                // if pool not found, call CreatePool. You can learn more about os families and versions at:
                // https://azure.microsoft.com/en-us/documentation/articles/cloud-services-guestos-update-matrix/
                CloudPool pool = client.PoolOperations.CreatePool(poolId, targetDedicated: 3, virtualMachineSize: "small", osFamily: "3");
                pool.Commit();
            }
        }

        /// <summary>
        /// Prints task information to the console for each of the nodes in the specified pool.
        /// </summary>
        /// <param name="poolId">The ID of the <see cref="CloudPool"/> containing the nodes whose task information
        /// should be printed to the console.</param>
        private static void PrintNodeTasks(BatchClient batchClient, string poolId)
        {
            ODATADetailLevel nodeDetail = new ODATADetailLevel(selectClause: "id,recentTasks");

            // Obtain and print the task information for each of the compute nodes in the pool.
            foreach (ComputeNode node in batchClient.PoolOperations.ListComputeNodes(poolId, nodeDetail))
            {
                Console.WriteLine();
                Console.WriteLine(node.Id + " tasks:");

                if (node.RecentTasks != null && node.RecentTasks.Any())
                {
                    foreach (TaskInformation task in node.RecentTasks)
                    {
                        Console.WriteLine("\t{0}: {1}", task.TaskId, task.TaskState);
                    }
                }
                else
                {
                    // No tasks found for the node
                    Console.WriteLine("\tNone");
                }

            }
        }

        /// <summary>
        /// Queries and prints task information for the specified job.
        /// </summary>
        /// <param name="batchClient">A fully initialized <see cref="BatchClient"/>.</param>
        /// <param name="jobId">The ID of the job whose tasks should be queried.</param>
        /// <param name="detail">An <see cref="ODATADetailLevel"/> configured with one or more of expand, filter, select clauses.</param>
        private static void QueryTasks(BatchClient batchClient, string jobId, ODATADetailLevel detail)
        {
            List<CloudTask> taskList = new List<CloudTask>();

            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();

            try
            {
                taskList.AddRange(batchClient.JobOperations.ListTasks(jobId, detail).ToList());
            }
            catch (AggregateException ex)
            {
                AggregateException ax = ex.Flatten();

                Console.WriteLine(ax.Message);
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
            }
            finally
            {
                stopwatch.Stop();
            }
            
            Console.WriteLine("{0} tasks retrieved in {1} (ExpandClause: {2} | FilterClause: {3} | SelectClause: {4})",
                        taskList.Count, 
                        stopwatch.Elapsed, 
                        detail.ExpandClause, 
                        detail.FilterClause, 
                        detail.SelectClause);
        }

        /// <summary>
        /// Creates a CloudPool with a single compute node associated with the Batch account.
        /// </summary>
        /// <param name="batchClient">A fully initialized <see cref="BatchClient"/>.</param>
        /// <param name="poolId">The ID of the <see cref="CloudPool"/>.</param>
        private async static Task CreatePool(BatchClient batchClient, string poolId)
        {
            // Create and configure an unbound pool with the specified ID
            CloudPool pool = batchClient.PoolOperations.CreatePool(poolId: poolId,
                                                                   osFamily: "3",
                                                                   virtualMachineSize: "small",
                                                                   targetDedicated: 1);

            await GettingStartedCommon.CreatePoolIfNotExistAsync(batchClient, pool);
        }

        /// <summary>
        /// Monitors the specified tasks for completion and returns a value indicating whether all tasks completed successfully
        /// within the timeout period.
        /// </summary>
        /// <param name="batchClient">A <see cref="BatchClient"/>.</param>
        /// <param name="jobId">The id of the job containing the tasks that should be monitored.</param>
        /// <param name="timeout">The period of time to wait for the tasks to reach the completed state.</param>
        /// <returns><c>true</c> if all tasks in the specified job completed with an exit code of 0 within the specified timeout period, otherwise <c>false</c>.</returns>
        private static async Task<bool> MonitorTasks(BatchClient batchClient, string jobId, TimeSpan timeout)
        {
            bool allTasksSuccessful = true;
            const string successMessage = "All tasks reached state Completed.";
            const string failureMessage = "One or more tasks failed to reach the Completed state within the timeout period.";

            // Obtain the collection of tasks currently managed by the job. Note that we use a detail level to
            // specify that only the "id" property of each task should be populated. Using a detail level for
            // all list operations helps to lower response time from the Batch service.
            ODATADetailLevel detail = new ODATADetailLevel(selectClause: "id");
            List<CloudTask> tasks = await batchClient.JobOperations.ListTasks(JobId, detail).ToListAsync();

            Console.WriteLine("Awaiting task completion, timeout in {0}...", timeout.ToString());

            // We use a TaskStateMonitor to monitor the state of our tasks. In this case, we will wait for all tasks to
            // reach the Completed state.
            TaskStateMonitor taskStateMonitor = batchClient.Utilities.CreateTaskStateMonitor();
            bool timedOut = await taskStateMonitor.WhenAllAsync(tasks, TaskState.Completed, timeout);

            if (timedOut)
            {
                allTasksSuccessful = false;

                await batchClient.JobOperations.TerminateJobAsync(jobId, failureMessage);

                Console.WriteLine(failureMessage);
            }
            else
            {
                await batchClient.JobOperations.TerminateJobAsync(jobId, successMessage);

                // All tasks have reached the "Completed" state, however, this does not guarantee all tasks completed successfully.
                // Here we further check each task's ExecutionInfo property to ensure that it did not encounter a scheduling error
                // or return a non-zero exit code.

                // Update the detail level to populate only the task id and executionInfo properties.
                // We refresh the tasks below, and need only this information for each task.
                detail.SelectClause = "id, executionInfo";

                foreach (CloudTask task in tasks)
                {
                    // Populate the task's properties with the latest info from the Batch service
                    await task.RefreshAsync(detail);

                    if (task.ExecutionInformation.SchedulingError != null)
                    {
                        // A scheduling error indicates a problem starting the task on the node. It is important to note that
                        // the task's state can be "Completed," yet still have encountered a scheduling error.

                        allTasksSuccessful = false;

                        Console.WriteLine("WARNING: Task [{0}] encountered a scheduling error: {1}", task.Id, task.ExecutionInformation.SchedulingError.Message);
                    }
                    else if (task.ExecutionInformation.ExitCode != 0)
                    {
                        // A non-zero exit code may indicate that the application executed by the task encountered an error
                        // during execution. As not every application returns non-zero on failure by default (e.g. robocopy),
                        // your implementation of error checking may differ from this example.

                        allTasksSuccessful = false;

                        Console.WriteLine("WARNING: Task [{0}] returned a non-zero exit code - this may indicate task execution or completion failure.", task.Id);
                    }
                }
            }

            if (allTasksSuccessful)
            {
                Console.WriteLine("Success! All tasks completed successfully within the specified timeout period.");
            }

            return allTasksSuccessful;
        }

        /// <summary>
        /// Creates tasks to process each of the specified input files, and submits them to the
        /// specified job for execution.
        /// </summary>
        /// <param name="batchClient">A <see cref="BatchClient"/>.</param>
        /// <param name="jobId">The id of the job to which the tasks should be added.</param>
        /// <param name="inputFiles">A collection of <see cref="ResourceFile"/> objects representing the input files to be
        /// processed by the tasks executed on the compute nodes.</param>
        /// <param name="outputContainerSasUrl">The shared access signature URL for the container within Azure Storage that
        /// will receive the output files created by the tasks.</param>
        /// <returns>A collection of the submitted tasks.</returns>
        private static async Task<List<CloudTask>> AddTasksAsync(BatchClient batchClient, string jobId, List<ResourceFile> inputFiles, string outputContainerSasUrl)
        {
            Console.WriteLine("Adding {0} tasks to job [{1}]...", inputFiles.Count, jobId);

            // Create a collection to hold the tasks that we'll be adding to the job
            List<CloudTask> tasks = new List<CloudTask>();

            // Create each of the tasks. Because we copied the task application to the
            // node's shared directory with the pool's StartTask, we can access it via
            // the shared directory on whichever node each task will run.
            foreach (ResourceFile inputFile in inputFiles)
            {
                string taskId = "topNtask" + inputFiles.IndexOf(inputFile);
                string taskCommandLine = String.Format("cmd /c %AZ_BATCH_NODE_SHARED_DIR%\\TaskApplication.exe {0} 3 \"{1}\"", inputFile.FilePath, outputContainerSasUrl);

                CloudTask task = new CloudTask(taskId, taskCommandLine);
                task.ResourceFiles = new List<ResourceFile> { inputFile };
                tasks.Add(task);
            }

            // Add the tasks as a collection opposed to a separate AddTask call for each. Bulk task submission
            // helps to ensure efficient underlying API calls to the Batch service.
            await batchClient.JobOperations.AddTaskAsync(jobId, tasks);

            return tasks;
        }

        /// <summary>
        /// Creates a job in the specified pool.
        /// </summary>
        /// <param name="batchClient">A <see cref="BatchClient"/>.</param>
        /// <param name="jobId">The id of the job to be created.</param>
        /// <param name="poolId">The id of the <see cref="CloudPool"/> in which to create the job.</param>
        /// <returns>A <see cref="System.Threading.Tasks.Task"/> object that represents the asynchronous operation.</returns>
        private static async Task CreateJobAsync(BatchClient batchClient, string jobId, string poolId)
        {
            Console.WriteLine("Creating job [{0}]...", jobId);

            CloudJob job = batchClient.JobOperations.CreateJob();
            job.Id = jobId;
            job.PoolInformation = new PoolInformation { PoolId = poolId };

            await job.CommitAsync();
        }