<commit_before>﻿// ----------------------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
// ----------------------------------------------------------------------------

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json.Linq;

namespace Microsoft.WindowsAzure.MobileServices.Sync
{
    /// <summary>
    /// Responsible for executing a batch of operations from queue until bookmark is found
    /// </summary>
    internal class PushAction: SyncAction
    {
        private BookmarkOperation bookmark;
        private IMobileServiceSyncHandler syncHandler;
        private MobileServiceJsonSerializerSettings serializerSettings;

        public PushAction(OperationQueue operationQueue, IMobileServiceLocalStore store, IMobileServiceSyncHandler syncHandler, MobileServiceJsonSerializerSettings  serializerSettings, CancellationToken cancellationToken, BookmarkOperation bookmark): base(operationQueue, store, cancellationToken)
        {
            this.syncHandler = syncHandler;
            this.bookmark = bookmark;
            this.serializerSettings = serializerSettings;
        }        

        public override async Task ExecuteAsync()
        {
            var batch = new OperationBatch(this.syncHandler, this.Store);

            // when cancellation is requested, abort the batch
            this.CancellationToken.Register(() => batch.Abort(MobileServicePushStatus.CancelledByToken));

            MobileServiceTableOperation operation = this.OperationQueue.Peek();

            // keep taking out operations and executing them until queue is empty or operation finds the bookmark or batch is aborted 
            while (operation != null && operation != this.bookmark)
            {
                await this.ExecuteOperationAsync(operation, batch);

                if (batch.AbortReason.HasValue)
                {
                    break;
                }

                // we successfuly executed an operation so remove it from queue
                await this.OperationQueue.DequeueAsync();

                // get next operation
                operation = this.OperationQueue.Peek();
            }

            // if sync operation found the bookmark in operation queue, remove it
            if (operation == this.bookmark)
            {
                await this.OperationQueue.DequeueAsync();
            }

            MobileServicePushStatus batchStatus = batch.AbortReason.GetValueOrDefault(MobileServicePushStatus.Complete);
            IEnumerable<MobileServiceTableOperationError> syncErrors = Enumerable.Empty<MobileServiceTableOperationError>();
            try
            {
                syncErrors = await batch.LoadSyncErrorsAsync(this.serializerSettings);
            }
            catch (Exception ex)
            {

                batch.OtherErrors.Add(new MobileServiceLocalStoreException(Resources.SyncStore_FailedToLoadError, ex));
            }

            var result = new MobileServicePushCompletionResult(syncErrors, batchStatus);

            try
            {
                await batch.SyncHandler.OnPushCompleteAsync(result);

                // now that we've successfully given the errors to user, we can delete them from store
                await batch.DeleteErrorsAsync();
            }
            catch (Exception ex)
            {
                batch.OtherErrors.Add(ex);
            }

            if (batch.AbortReason.HasValue || batch.HasErrors(syncErrors))
            {
                List<MobileServiceTableOperationError> unhandledSyncErrors = syncErrors.Where(e=>!e.Handled).ToList();
                
                Exception inner;
                if (batch.OtherErrors.Count == 1)
                {
                    inner = batch.OtherErrors[0];
                }
                else
                {
                    inner = batch.OtherErrors.Any() ? new AggregateException(batch.OtherErrors) : null;
                }

                // create a new result with only unhandled errors
                result = new MobileServicePushCompletionResult(unhandledSyncErrors, batchStatus);
                this.TaskSource.TrySetException(new MobileServicePushFailedException(result, inner));
            }
            else
            {
                this.TaskSource.SetResult(0);
            }
        }

        private async Task ExecuteOperationAsync(MobileServiceTableOperation operation, OperationBatch batch)
        {
            if (operation is BookmarkOperation)
            {
                return; // nothing to execute in a bookmark
            }

            using (await this.OperationQueue.LockItemAsync(operation.ItemId, this.CancellationToken))
            {
                if (operation.IsCancelled || this.CancellationToken.IsCancellationRequested)
                {
                    return;
                }

                Exception error = null;

                await this.LoadOperationItem(operation, batch);

                if (this.CancellationToken.IsCancellationRequested)
                {
                    return;
                }

                JObject result = null;
                try
                {
                    result = await batch.SyncHandler.ExecuteTableOperationAsync(operation);                    <commit_msg>Do not leak exception out of ExecuteAsync() of PushAction<commit_after>﻿// ----------------------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
// ----------------------------------------------------------------------------

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json.Linq;

namespace Microsoft.WindowsAzure.MobileServices.Sync
{
    /// <summary>
    /// Responsible for executing a batch of operations from queue until bookmark is found
    /// </summary>
    internal class PushAction: SyncAction
    {
        private BookmarkOperation bookmark;
        private IMobileServiceSyncHandler syncHandler;
        private MobileServiceJsonSerializerSettings serializerSettings;

        public PushAction(OperationQueue operationQueue, IMobileServiceLocalStore store, IMobileServiceSyncHandler syncHandler, MobileServiceJsonSerializerSettings  serializerSettings, CancellationToken cancellationToken, BookmarkOperation bookmark): base(operationQueue, store, cancellationToken)
        {
            this.syncHandler = syncHandler;
            this.bookmark = bookmark;
            this.serializerSettings = serializerSettings;
        }        

        public override async Task ExecuteAsync()
        {
            var batch = new OperationBatch(this.syncHandler, this.Store);
            List<MobileServiceTableOperationError> syncErrors = new List<MobileServiceTableOperationError>();
            MobileServicePushStatus batchStatus = MobileServicePushStatus.InternalError;

            try
            {
                batchStatus = await ExecuteBatchAsync(batch, syncErrors);
            }
            catch (Exception ex)
            {
                batch.OtherErrors.Add(ex);
            }

            await FinalizePush(batch, batchStatus, syncErrors);
        }

        private async Task<MobileServicePushStatus> ExecuteBatchAsync(OperationBatch batch, List<MobileServiceTableOperationError> syncErrors)
        {
            // when cancellation is requested, abort the batch
            this.CancellationToken.Register(() => batch.Abort(MobileServicePushStatus.CancelledByToken));

            try
            {
                await ExecuteAllOperationsAsync(batch);
            }
            catch (Exception ex)
            {
                batch.OtherErrors.Add(ex);
            }

            MobileServicePushStatus batchStatus = batch.AbortReason.GetValueOrDefault(MobileServicePushStatus.Complete);
            try
            {
                syncErrors.AddRange(await batch.LoadSyncErrorsAsync(this.serializerSettings));
            }
            catch (Exception ex)
            {

                batch.OtherErrors.Add(new MobileServiceLocalStoreException(Resources.SyncStore_FailedToLoadError, ex));
            }
            return batchStatus;
        }

        private async Task FinalizePush(OperationBatch batch, MobileServicePushStatus batchStatus, IEnumerable<MobileServiceTableOperationError> syncErrors)
        {
            var result = new MobileServicePushCompletionResult(syncErrors, batchStatus);

            try
            {
                await batch.SyncHandler.OnPushCompleteAsync(result);

                // now that we've successfully given the errors to user, we can delete them from store
                await batch.DeleteErrorsAsync();
            }
            catch (Exception ex)
            {
                batch.OtherErrors.Add(ex);
            }

            if (batchStatus != MobileServicePushStatus.Complete || batch.HasErrors(syncErrors))
            {
                List<MobileServiceTableOperationError> unhandledSyncErrors = syncErrors.Where(e => !e.Handled).ToList();

                Exception inner;
                if (batch.OtherErrors.Count == 1)
                {
                    inner = batch.OtherErrors[0];
                }
                else
                {
                    inner = batch.OtherErrors.Any() ? new AggregateException(batch.OtherErrors) : null;
                }

                // create a new result with only unhandled errors
                result = new MobileServicePushCompletionResult(unhandledSyncErrors, batchStatus);
                this.TaskSource.TrySetException(new MobileServicePushFailedException(result, inner));
            }
            else
            {
                this.TaskSource.SetResult(0);
            }
        }

        private async Task ExecuteAllOperationsAsync(OperationBatch batch)
        {
            MobileServiceTableOperation operation = this.OperationQueue.Peek();

            // keep taking out operations and executing them until queue is empty or operation finds the bookmark or batch is aborted 
            while (operation != null && operation != this.bookmark)
            {
                await this.ExecuteOperationAsync(operation, batch);

                if (batch.AbortReason.HasValue)
                {
                    break;
                }

                // we successfuly executed an operation so remove it from queue
                await this.OperationQueue.DequeueAsync();

                // get next operation
                operation = this.OperationQueue.Peek();
            }

            // if sync operation found the bookmark in operation queue, remove it
            if (operation == this.bookmark)
            {
                await this.OperationQueue.DequeueAsync();
            }
        }

        private async Task ExecuteOperationAsync(MobileServiceTableOperation operation, OperationBatch batch)
        {
            if (operation is BookmarkOperation)
            {
                return; // nothing to execute in a bookmark
            }

            using (await this.OperationQueue.LockItemAsync(operation.ItemId, this.CancellationToken))
            {
                if (operation.IsCancelled || this.CancellationToken.IsCancellationRequested)
                {
                    return;
                }

                Exception error = null;

                await this.LoadOperationItem(operation, batch);

                if (this.CancellationToken.IsCancellationRequested)
                {
                    return;
                }

                JObject result = null;
                try
                {
                    result = await batch.SyncHandler.ExecuteTableOperationAsync(operation);                    
{'fraction_non_alphanumeric': 0.09058541028401093, 'fraction_numerical': 0.0004968121222157821, 'mean_word_length': 1.7014090807425633, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3439901', 'n_tokens_mistral': 2699, 'n_tokens_neox': 2575, 'n_words': 630}