360
branchToken []byte,
361
namespaceName string,
363
>
return func(paginationToken []byte) ([]HistoryBlobsPaginationItem, []byte, error) {
364
>
resp, err := r.executionMgr.ReadHistoryBranchByBatch(ctx, &persistence.ReadHistoryBranchRequest{
365
>
BranchToken: branchToken,
366
>
MinEventID: firstEventID,
367
>
MaxEventID: nextEventID,
368
>
PageSize: defaultPageSize,
369
>
NextPageToken: paginationToken,
370
>
ShardID: r.shard.GetShardID(),
371
>
})
372
>
if err != nil {
373
return nil, nil, err
374
}
375
377
>
paginateItems := make([]HistoryBlobsPaginationItem, 0, len(resp.History))
378
>
for i, history := range resp.History {
379
>
nextBatch := HistoryBlobsPaginationItem{
380
>
History: history,
381
>
TransactionID: resp.TransactionIDs[i],
382
>
}
383
>
paginateItems = append(paginateItems, nextBatch)
384
>
385
>
// Calculate and accumulate external payload size and count for this batch of history events
386
>
if r.shard.GetConfig().ExternalPayloadsEnabled(namespaceName) {
387
>
externalPayloadSize, externalPayloadCount, err := workflow.CalculateExternalPayloadSize(
388
>
history.Events,
389
>
metrics.NoopMetricsHandler, // don't record metrics since those are not new uploads
390
>
)
391
>
if err != nil {
392
return nil, nil, err
393
}
395
>
r.rebuiltExternalPayloadCount += externalPayloadCount
396
}
397
}
399
}
400
}