@@ -105,6 +105,15 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
105105 }
106106 }
107107
108+ context . SendSynchronizationEvent ( new SynchronizationEventArgs ( )
109+ {
110+ EventType = SynchronizationEventType . ItemsCommitted ,
111+ EntityType = pullResponse . EntityType ,
112+ ItemsProcessed = pullResponse . TotalItemsProcessed ,
113+ TotalNrItems = pullResponse . TotalRequestItems ,
114+ QueryId = pullResponse . QueryId
115+ } ) ;
116+
108117 if ( pullOptions . SaveAfterEveryServiceRequest )
109118 {
110119 _ = await context . SaveChangesAsync ( true , false , cancellationToken ) . ConfigureAwait ( false ) ;
@@ -120,10 +129,22 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
120129 try
121130 {
122131 bool completed = false ;
132+ long itemsProcessed = 0 ;
123133 do
124134 {
125135 Page < object > page = await GetPageAsync ( pullRequest . HttpClient , requestUri , pageType , cancellationToken ) . ConfigureAwait ( false ) ;
126- databaseUpdateQueue . Enqueue ( new PullResponse ( pullRequest . EntityType , pullRequest . QueryId , page . Items ) ) ;
136+ itemsProcessed += page . Items . Count ( ) ;
137+
138+ context . SendSynchronizationEvent ( new SynchronizationEventArgs ( )
139+ {
140+ EventType = SynchronizationEventType . ItemsFetched ,
141+ EntityType = pullRequest . EntityType ,
142+ ItemsProcessed = itemsProcessed ,
143+ TotalNrItems = page . Count ?? 0 ,
144+ QueryId = pullRequest . QueryId
145+ } ) ;
146+
147+ databaseUpdateQueue . Enqueue ( new PullResponse ( pullRequest . EntityType , pullRequest . QueryId , page . Items , page . Count ?? 0 , itemsProcessed ) ) ;
127148 if ( ! string . IsNullOrEmpty ( page . NextLink ) )
128149 {
129150 requestUri = new UriBuilder ( endpoint ) { Query = page . NextLink } . Uri ;
@@ -173,6 +194,8 @@ public async Task<PullResult> ExecuteAsync(IEnumerable<PullRequest> requests, Pu
173194 /// <exception cref="DatasyncPullException">Thrown on error</exception>
174195 internal async Task < Page < object > > GetPageAsync ( HttpClient client , Uri requestUri , Type pageType , CancellationToken cancellationToken = default )
175196 {
197+ PropertyInfo countPropInfo = pageType . GetProperty ( "Count" )
198+ ?? throw new DatasyncException ( $ "Page type '{ pageType . Name } ' does not have a 'Count' property") ;
176199 PropertyInfo itemsPropInfo = pageType . GetProperty ( "Items" )
177200 ?? throw new DatasyncException ( $ "Page type '{ pageType . Name } ' does not have an 'Items' property") ;
178201 PropertyInfo nextLinkPropInfo = pageType . GetProperty ( "NextLink" )
@@ -193,6 +216,7 @@ internal async Task<Page<object>> GetPageAsync(HttpClient client, Uri requestUri
193216
194217 return new Page < object > ( )
195218 {
219+ Count = ( long ? ) countPropInfo . GetValue ( result ) ,
196220 Items = ( IEnumerable < object > ) itemsPropInfo . GetValue ( result ) ! ,
197221 NextLink = ( string ? ) nextLinkPropInfo . GetValue ( result )
198222 } ;
@@ -237,6 +261,8 @@ internal static QueryDescription PrepareQueryDescription(QueryDescription source
237261 /// <param name="EntityType">The type of entity contained within the items.</param>
238262 /// <param name="QueryId">The query ID for the request.</param>
239263 /// <param name="Items">The list of items to process.</param>
264+ /// <param name="TotalRequestItems">The total number of items in the current pull request.</param>
265+ /// <param name="TotalItemsProcessed">The total number of items processed, <paramref name="Items"/> included.</param>
240266 [ ExcludeFromCodeCoverage ]
241- internal record PullResponse ( Type EntityType , string QueryId , IEnumerable < object > Items ) ;
267+ internal record PullResponse ( Type EntityType , string QueryId , IEnumerable < object > Items , long TotalRequestItems , long TotalItemsProcessed ) ;
242268}
0 commit comments