488
}
489
490
>
func (p *taskProcessorImpl) paginationFn(_ []byte) ([]any, []byte, error) {
task_processor.go
491
>
respChan := make(chan *replicationspb.ReplicationMessages, 1)
492
>
var lastProcessedVisTime *timestamppb.Timestamp
493
>
if !p.maxRxProcessedTimestamp.IsZero() {
494
lastProcessedVisTime = timestamppb.New(p.maxRxProcessedTimestamp)
495
}
497
>
token: &replicationspb.ReplicationToken{
498
>
ShardId: p.sourceShardID,
499
>
LastProcessedMessageId: p.maxRxProcessedTaskID,
500
>
LastProcessedVisibilityTime: lastProcessedVisTime,
501
>
LastRetrievedMessageId: p.maxRxReceivedTaskID,
502
>
},
503
>
respChan: respChan,
504
>
}
505
>
506
>
select {
507
>
case resp, ok := <-respChan:
508
>
if !ok {
509
return nil, nil, nil
510
}