sdk/dotnet/src/Microsoft.Datasync.Client/Offline/SyncContext.cs
/// <summary>
/// Pulls the items matching the provided query from the remote table.
/// </summary>
/// <param name="tableName">The name of the offline table.</param>
/// <param name="query">The OData query that determines which items to pull from the remote table.</param>
/// <param name="options">The options used to configure the pull operation.</param>
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe.</param>
/// <returns>A task that completes when the pull operation has finished.</returns>
Pulls the items matching the provided query from the remote table.
Pulls the items matching the provided query from the remote table .
public async Task PullItemsAsync(string tableName, string query, PullOptions options, CancellationToken cancellationToken = default)
        {
            Arguments.IsValidTableName(tableName, nameof(tableName));
            Arguments.IsNotNull(query, nameof(query));
            Arguments.IsNotNull(options, nameof(options));
            EnsureContextIsInitialized();
            var table = ServiceClient.GetRemoteTable(tableName);
            var queryId = options.QueryId ?? GetQueryIdFromQuery(tableName, query);
            var queryDescription = QueryDescription.Parse(tableName, query);
            string[] relatedTables = options.PushOtherTables ? null : new string[] { tableName };
            if (queryDescription.Selection.Count > 0 || queryDescription.Projections.Count > 0)
            {
                throw new ArgumentException("Pull query with select clause is not supported.", nameof(query));
            }
            queryDescription.Ordering.Clear();
            queryDescription.Top = null;
            queryDescription.Skip = null;
            queryDescription.IncludeTotalCount = false;
            if (await TableIsDirtyAsync(tableName, cancellationToken).ConfigureAwait(false))
            {
                await PushContext.PushItemsAsync(relatedTables, cancellationToken).ConfigureAwait(false);
                if (await TableIsDirtyAsync(tableName, cancellationToken).ConfigureAwait(false))
                {
                    throw new DatasyncInvalidOperationException($"There are still pending operations for table '{tableName}' after a push");
                }
            }
            var deltaToken = await DeltaTokenStore.GetDeltaTokenAsync(tableName, queryId, cancellationToken).ConfigureAwait(false);
            var deltaTokenFilter = new BinaryOperatorNode(BinaryOperatorKind.GreaterThan, new MemberAccessNode(null, SystemProperties.JsonUpdatedAtProperty), new ConstantNode(deltaToken));
            queryDescription.Filter = queryDescription.Filter == null ? deltaTokenFilter : new BinaryOperatorNode(BinaryOperatorKind.And, queryDescription.Filter, deltaTokenFilter);
            Dictionary<string, string> parameters = new()
            {
                { ODataOptions.IncludeDeleted, "true" }
            };
            var odataString = queryDescription.ToODataString(parameters);
            await foreach (var instance in table.GetAsyncItems(odataString).WithCancellation(cancellationToken))
            {
                cancellationToken.ThrowIfCancellationRequested();
                if (instance is not JObject item)
                {
                    throw new DatasyncInvalidOperationException("Received item is not an object");
                }
                string itemId = ServiceSerializer.GetId(item);
                if (itemId == null)
                {
                    throw new DatasyncInvalidOperationException("Received an item without an ID");
                }
                var pendingOperation = await OperationsQueue.GetOperationByItemIdAsync(tableName, itemId, cancellationToken).ConfigureAwait(false);
                if (pendingOperation != null)
                {
                    throw new InvalidOperationException("Received an item for which there is a pending operation.");
                }
                DateTimeOffset? updatedAt = ServiceSerializer.GetUpdatedAt(item)?.ToUniversalTime();
                if (ServiceSerializer.IsDeleted(item))
                {
                    await OfflineStore.DeleteAsync(tableName, new[] { itemId }, cancellationToken).ConfigureAwait(false);
                }
                else
                {
                    await OfflineStore.UpsertAsync(tableName, new[] { item }, true, cancellationToken).ConfigureAwait(false);
                }
                if (updatedAt.HasValue)
                {
                    await DeltaTokenStore.SetDeltaTokenAsync(tableName, queryId, updatedAt.Value, cancellationToken).ConfigureAwait(false);
                }
            }
        }
public async Task PullItemsAsync ( string tableName , string query , PullOptions options , CancellationToken cancellationToken = default ) { Arguments . IsValidTableName ( tableName , nameof ( tableName ) ) ; Arguments . IsNotNull ( query , nameof ( query ) ) ; Arguments . IsNotNull ( options , nameof ( options ) ) ; EnsureContextIsInitialized ( ) ; var table = ServiceClient . GetRemoteTable ( tableName ) ; var queryId = options . QueryId ?? GetQueryIdFromQuery ( tableName , query ) ; var queryDescription = QueryDescription . Parse ( tableName , query ) ; string [ ] relatedTables = options . PushOtherTables ? null : new string [ ] { tableName } ; if ( queryDescription . Selection . Count > 0 || queryDescription . Projections . Count > 0 ) { throw new ArgumentException ( " " , nameof ( query ) ) ; } queryDescription . Ordering . Clear ( ) ; queryDescription . Top = null ; queryDescription . Skip = null ; queryDescription . IncludeTotalCount = false ; if ( await TableIsDirtyAsync ( tableName , cancellationToken ) . ConfigureAwait ( false ) ) { await PushContext . PushItemsAsync ( relatedTables , cancellationToken ) . ConfigureAwait ( false ) ; if ( await TableIsDirtyAsync ( tableName , cancellationToken ) . ConfigureAwait ( false ) ) { throw new DatasyncInvalidOperationException ( $" There are still pending operations for table ' { tableName } ' after a push " ) ; } } var deltaToken = await DeltaTokenStore . GetDeltaTokenAsync ( tableName , queryId , cancellationToken ) . ConfigureAwait ( false ) ; var deltaTokenFilter = new BinaryOperatorNode ( BinaryOperatorKind . GreaterThan , new MemberAccessNode ( null , SystemProperties . JsonUpdatedAtProperty ) , new ConstantNode ( deltaToken ) ) ; queryDescription . Filter = queryDescription . Filter == null ? deltaTokenFilter : new BinaryOperatorNode ( BinaryOperatorKind . And , queryDescription . Filter , deltaTokenFilter ) ; Dictionary < string , string > parameters = new ( ) { { ODataOptions . IncludeDeleted , " " } } ; var odataString = queryDescription . ToODataString ( parameters ) ; await foreach ( var instance in table . GetAsyncItems ( odataString ) . WithCancellation ( cancellationToken ) ) { cancellationToken . ThrowIfCancellationRequested ( ) ; if ( instance is not JObject item ) { throw new DatasyncInvalidOperationException ( " " ) ; } string itemId = ServiceSerializer . GetId ( item ) ; if ( itemId == null ) { throw new DatasyncInvalidOperationException ( " " ) ; } var pendingOperation = await OperationsQueue . GetOperationByItemIdAsync ( tableName , itemId , cancellationToken ) . ConfigureAwait ( false ) ; if ( pendingOperation != null ) { throw new InvalidOperationException ( " " ) ; } DateTimeOffset ? updatedAt = ServiceSerializer . GetUpdatedAt ( item ) ? . ToUniversalTime ( ) ; if ( ServiceSerializer . IsDeleted ( item ) ) { await OfflineStore . DeleteAsync ( tableName , new [ ] { itemId } , cancellationToken ) . ConfigureAwait ( false ) ; } else { await OfflineStore . UpsertAsync ( tableName , new [ ] { item } , true , cancellationToken ) . ConfigureAwait ( false ) ; } if ( updatedAt . HasValue ) { await DeltaTokenStore . SetDeltaTokenAsync ( tableName , queryId , updatedAt . Value , cancellationToken ) . ConfigureAwait ( false ) ; } } }
Pulls the items matching the provided query from the remote table.
Pulls the items matching the provided query from the remote table .
// If the table is still dirty, then throw an error.
A task that completes when the pull operation has finished.
A task that completes when the pull operation has finished .
tableName
string
The name of the offline table.
The name of the offline table .
query
string
The OData query that determines which items to pull from the remote table.
The OData query that determines which items to pull from the remote table .
options
PullOptions
The options used to configure the pull operation.
The options used to configure the pull operation .
cancellationToken
CancellationToken
A  to observe.
A to observe .