Skip to content

Commit d4fd2f7

Browse files
StefanCuypersStefanCuypers
committed
Added test PullAsync_List_FailedRequest_SynchronizationEventWorks Added Synchronization events for push operations and test for it.
1 parent e232d8c commit d4fd2f7

File tree

5 files changed

+191
-9
lines changed

5 files changed

+191
-9
lines changed

‎src/CommunityToolkit.Datasync.Client/Offline/Operations/PullOperationManager.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
187187
catch (DatasyncPullException ex)
188188
{
189189
result.AddFailedRequest(requestUri, ex.ServiceResponse);
190-
databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, Enumerable.Empty<object>(), totalCount, itemsProcessed, true, ex));
190+
databaseUpdateQueue.Enqueue(new PullResponse(pullRequest.EntityType, pullRequest.QueryId, [], totalCount, itemsProcessed, true, ex));
191191
}
192192
});
193193

‎src/CommunityToolkit.Datasync.Client/Offline/OperationsQueue/OperationsQueueManager.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -270,16 +270,40 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
270270

271271
// Determine the list of queued operations in scope.
272272
List<DatasyncOperation> queuedOperations = await GetQueuedOperationsAsync(entityTypeNames, cancellationToken).ConfigureAwait(false);
273+
274+
// Signal we started the push operation.
275+
this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
276+
{
277+
EventType = SynchronizationEventType.PushStarted,
278+
TotalNrItems = queuedOperations.Count
279+
});
280+
273281
if (queuedOperations.Count == 0)
274282
{
283+
// Signal we ended the push operation.
284+
this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
285+
{
286+
EventType = SynchronizationEventType.PushEnded
287+
});
275288
return pushResult;
276289
}
277290

291+
int nrItemsProcessed = 0;
292+
278293
// Push things in parallel, according to the PushOptions
279294
QueueHandler<DatasyncOperation> queueHandler = new(pushOptions.ParallelOperations, async operation =>
280295
{
281296
ServiceResponse? response = await PushOperationAsync(operation, cancellationToken).ConfigureAwait(false);
282297
pushResult.AddOperationResult(operation, response);
298+
// We can run on multiple threads, so use Interlocked to update the number of items processed.
299+
int newItemsProcessed = Interlocked.Increment(ref nrItemsProcessed);
300+
this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
301+
{
302+
EventType = SynchronizationEventType.PushItem,
303+
ItemsProcessed = newItemsProcessed,
304+
TotalNrItems = queuedOperations.Count,
305+
PushOperation = operation,
306+
});
283307
});
284308

285309
// Enqueue and process all the queued operations in scope
@@ -288,6 +312,14 @@ internal async Task<PushResult> PushAsync(IEnumerable<Type> entityTypes, PushOpt
288312

289313
// Save the changes, this time we don't update the queue.
290314
_ = await this._context.SaveChangesAsync(acceptAllChangesOnSuccess: true, addToQueue: false, cancellationToken).ConfigureAwait(false);
315+
316+
this._context.SendSynchronizationEvent(new SynchronizationEventArgs()
317+
{
318+
EventType = SynchronizationEventType.PushEnded,
319+
ItemsProcessed = nrItemsProcessed,
320+
TotalNrItems = queuedOperations.Count,
321+
});
322+
291323
return pushResult;
292324
}
293325

‎src/CommunityToolkit.Datasync.Client/Offline/SynchronizationEventArgs.cs‎

Lines changed: 31 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,18 @@ public enum SynchronizationEventType
3030
/// Pull for the given entiry ended.
3131
/// </summary>
3232
PullEnded,
33+
/// <summary>
34+
/// Push operation started.
35+
/// </summary>
36+
PushStarted,
37+
/// <summary>
38+
/// An item was pushed to the server
39+
/// </summary>
40+
PushItem,
41+
/// <summary>
42+
/// Push operation ended.
43+
/// </summary>
44+
PushEnded,
3345
}
3446

3547
/// <summary>
@@ -40,35 +52,46 @@ public class SynchronizationEventArgs
4052
/// <summary>
4153
/// The type of event.
4254
/// </summary>
55+
/// <remarks>
56+
/// On pull events, reporting occurs per entity type. With a start/stop per entity type.
57+
/// On push events, reporting occurs per push request, which may contain multiple entity types.
58+
/// </remarks>
4359
public required SynchronizationEventType EventType { get; init; }
4460

4561
/// <summary>
46-
/// The EntityType that is being processed.
62+
/// The EntityType that is being processed. Not used on push events.
4763
/// </summary>
48-
public required Type EntityType { get; init; }
64+
public Type? EntityType { get; init; }
4965

5066
/// <summary>
51-
/// When pulling records, the number of items that have been processed in the current pull request.
67+
/// When pulling records, the number of items for the given entiry that have been processed in the current pull request.
68+
/// When pushing records, the total number of items that have been processed in the current push request.
5269
/// </summary>
5370
public long ItemsProcessed { get; init; } = -1;
5471

5572
/// <summary>
56-
/// The total number of items in the current pull request.
73+
/// When pulling records, the total number of items to pull for the given entity in the current pull request
74+
/// When pushing records, the total number of items that are being pushed in the current push request.
5775
/// </summary>
5876
public long TotalNrItems { get; init; }
5977

6078
/// <summary>
61-
/// The query ID that is being processed
79+
/// The query ID that is being processed on pull operations. Not used on push events.
6280
/// </summary>
63-
public required string QueryId { get; init; }
81+
public string? QueryId { get; init; }
6482

6583
/// <summary>
66-
/// If not <c>null</c> on event type <see cref="SynchronizationEventType.PullEnded"/>, indicates pull failed with this exception.
84+
/// If not <c>null</c> on event type <see cref="SynchronizationEventType.PullEnded"/>, indicates pull failed with this exception. Currently not used on push.
6785
/// </summary>
6886
public Exception? Exception { get; init; }
6987

7088
/// <summary>
71-
/// If a <see cref="DatasyncException"/> occured in <see cref="Exception"/> during server call processing, this property has more detail on the server response.
89+
/// If a <see cref="DatasyncException"/> occured in <see cref="Exception"/> during server call processing, this property has more detail on the server response. Currently not used on push, use the returned <see cref="PushResult.FailedRequests"/> instead.
7290
/// </summary>
7391
public ServiceResponse? ServiceResponse { get; init; }
92+
93+
/// <summary>
94+
/// The operation that was executed. Not used on pull events.
95+
/// </summary>
96+
public DatasyncOperation? PushOperation { get; init; }
7497
}

‎tests/CommunityToolkit.Datasync.Client.Test/Offline/OfflineDbContext_Tests.cs‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
using CommunityToolkit.Datasync.Client.Test.Offline.Helpers;
1010
using CommunityToolkit.Datasync.TestCommon;
1111
using CommunityToolkit.Datasync.TestCommon.Databases;
12+
using CommunityToolkit.Datasync.TestCommon.Models;
1213
using Microsoft.Data.Sqlite;
1314
using Microsoft.EntityFrameworkCore;
1415
using System.Net;
@@ -1501,6 +1502,112 @@ public async Task SynchronizationProgress_Event_Works()
15011502
currentItemsCommited.Should().Be(20);
15021503
}
15031504

1505+
[Fact]
1506+
public async Task PullAsync_List_FailedRequest_SynchronizationEventWorks()
1507+
{
1508+
this.context.Handler.AddResponse(HttpStatusCode.BadRequest);
1509+
1510+
bool eventFiredForStart = false;
1511+
bool eventFiredForEnd = false;
1512+
1513+
this.context.SynchronizationProgress += (sender, args) =>
1514+
{
1515+
sender.Should().Be(this.context);
1516+
args.EntityType.Should().Be<ClientMovie>();
1517+
args.QueryId.Should().Be("CommunityToolkit.Datasync.TestCommon.Databases.ClientMovie");
1518+
switch (args.EventType)
1519+
{
1520+
case SynchronizationEventType.PullStarted:
1521+
eventFiredForStart.Should().BeFalse("PullStarted event should only fire once");
1522+
eventFiredForStart = true;
1523+
args.Exception.Should().BeNull();
1524+
args.ServiceResponse.Should().BeNull();
1525+
break;
1526+
case SynchronizationEventType.PullEnded:
1527+
eventFiredForEnd.Should().BeFalse("PullEnded event should only fire once");
1528+
eventFiredForEnd = true;
1529+
args.Exception.Should().NotBeNull();
1530+
args.Exception.Should().BeOfType<Client.Exceptions.DatasyncPullException>();
1531+
args.ServiceResponse.Should().NotBeNull();
1532+
args.ServiceResponse.StatusCode.Should().Be(400);
1533+
break;
1534+
default:
1535+
Assert.Fail($"Unexpected event type: {args.EventType}");
1536+
break;
1537+
}
1538+
};
1539+
1540+
PullResult pullResult = await this.context.PullAsync([typeof(ClientMovie)], new PullOptions());
1541+
1542+
eventFiredForStart.Should().BeTrue();
1543+
eventFiredForEnd.Should().BeTrue();
1544+
}
1545+
1546+
[Fact]
1547+
public async Task SynchronizationProgress_Event_Works_For_Push()
1548+
{
1549+
// Add movies for testing
1550+
(MovieBase movie, string id)[] newMovies =
1551+
[
1552+
(TestData.Movies.BlackPanther,Guid.NewGuid().ToString("N")),
1553+
(TestData.Movies.Dune,Guid.NewGuid().ToString("N")),
1554+
(TestData.Movies.DrNo ,Guid.NewGuid().ToString("N")),
1555+
];
1556+
1557+
foreach ((MovieBase movie, string id) in newMovies)
1558+
{
1559+
this.context.Movies.Add(new(movie) { Id = id });
1560+
ClientMovie responseMovie = new(movie) { Id = id, UpdatedAt = DateTimeOffset.UtcNow, Version = Guid.NewGuid().ToString() };
1561+
this.context.Handler.AddResponseContent(DatasyncSerializer.Serialize(responseMovie), HttpStatusCode.Created);
1562+
this.context.SaveChanges();
1563+
}
1564+
1565+
bool eventFiredForItem = false;
1566+
bool eventFiredForStart = false;
1567+
bool eventFiredForEnd = false;
1568+
int[] itemsProcessedReported = new int[newMovies.Length]; // Due to multithreading, we can't guarantee the order of items processed. So register arrival of each separately.
1569+
1570+
this.context.SynchronizationProgress += (sender, args) =>
1571+
{
1572+
sender.Should().Be(this.context);
1573+
args.Exception.Should().BeNull();
1574+
args.ServiceResponse.Should().BeNull();
1575+
args.TotalNrItems.Should().Be(newMovies.Length);
1576+
switch (args.EventType)
1577+
{
1578+
case SynchronizationEventType.PushItem:
1579+
args.TotalNrItems.Should().Be(newMovies.Length);
1580+
args.ItemsProcessed.Should().BeInRange(1,newMovies.Length);
1581+
int prevProcessed = Interlocked.Exchange(ref itemsProcessedReported[args.ItemsProcessed-1], 1);
1582+
prevProcessed.Should().Be(0, "Each item should only be reported once");
1583+
args.PushOperation.Should().NotBeNull();
1584+
args.PushOperation.ItemId.Should().Be(newMovies[args.ItemsProcessed - 1].id);
1585+
eventFiredForItem = true;
1586+
break;
1587+
case SynchronizationEventType.PushStarted:
1588+
eventFiredForStart.Should().BeFalse("PushStarted event should only fire once");
1589+
eventFiredForStart = true;
1590+
break;
1591+
case SynchronizationEventType.PushEnded:
1592+
eventFiredForEnd.Should().BeFalse("PushEnded event should only fire once");
1593+
eventFiredForEnd = true;
1594+
args.ItemsProcessed.Should().Be(newMovies.Length);
1595+
itemsProcessedReported.Should().OnlyContain(x => x == 1, "All items should be reported as processed");
1596+
args.PushOperation.Should().BeNull();
1597+
break;
1598+
default:
1599+
Assert.Fail($"Invalid event type: {args.EventType}");
1600+
break;
1601+
}
1602+
};
1603+
1604+
PushResult results = await this.context.Movies.PushAsync();
1605+
1606+
eventFiredForStart.Should().BeTrue();
1607+
eventFiredForItem.Should().BeTrue();
1608+
eventFiredForEnd.Should().BeTrue();
1609+
}
1610+
15041611
#endregion
15051612

15061613
public class NotOfflineDbContext : DbContext

‎tests/CommunityToolkit.Datasync.TestCommon/TestData/Movies.cs‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,26 @@ public static class Movies
2020
Year = 2018
2121
};
2222

23+
public static readonly MovieBase Dune = new()
24+
{
25+
BestPictureWinner = false,
26+
Duration = 155,
27+
Rating = MovieRating.PG13,
28+
ReleaseDate = new DateOnly(2021, 10, 22),
29+
Title = "Dune",
30+
Year = 2021
31+
};
32+
33+
public static readonly MovieBase DrNo = new()
34+
{
35+
BestPictureWinner = false,
36+
Duration = 110,
37+
Rating = MovieRating.PG,
38+
ReleaseDate = new DateOnly(1962, 5, 8),
39+
Title = "Dr. No",
40+
Year = 1962
41+
};
42+
2343
/// <summary>
2444
/// Counts the number of items in the list that match the predicate.
2545
/// </summary>

0 commit comments

Comments
 (0)