118
}
119
120
>
response, err := m.queue.ReadMessages(ctx, &InternalReadMessagesRequest{
history_task_queue_manager.go
121
>
QueueType: request.QueueKey.QueueType,
122
>
QueueName: request.QueueKey.GetQueueName(),
123
>
PageSize: request.PageSize,
124
>
NextPageToken: request.NextPageToken,
125
>
})
126
>
if err != nil {
127
return nil, err
128
}
129
131
>
for i, message := range response.Messages {
132
>
var task persistencespb.HistoryTask
133
>
err := serialization.Decode(message.Data, &task)
134
>
if err != nil {
135
return nil, fmt.Errorf("%v: %w", ErrMsgDeserializeRawHistoryTask, err)
136
}