+ topic.toStringUtf8(), bke);
request.callback.scanFailed(request.ctx, new PubSubException.ServiceDownException(bke));
return;
}
LedgerEntry entry = null;
while (seq.hasMoreElements()) {
entry = seq.nextElement();
Message message;
try {
message = Message.parseFrom(entry.getEntryInputStream());
} catch (IOException e) {
String msg = "Unreadable message found in ledger: " + imlr.range.getLedgerId()
+ " for topic: " + topic.toStringUtf8();
logger.error(msg, e);
request.callback.scanFailed(ctx, new PubSubException.UnexpectedConditionException(msg));
return;
}
logger.debug("Read response from ledger: {} entry-id: {}",
lh.getId(), entry.getEntryId());
assert expectedEntryId == entry.getEntryId() : "expectedEntryId (" + expectedEntryId
+ ") != entry.getEntryId() (" + entry.getEntryId() + ")";
assert (message.getMsgId().getLocalComponent() - imlr.getStartSeqIdIncluded()) == expectedEntryId;
expectedEntryId++;
request.callback.messageScanned(ctx, message);
numMessagesRead++;
totalSizeRead += message.getBody().size();
if (numMessagesRead >= request.messageLimit) {
request.callback.scanFinished(ctx, ReasonForFinish.NUM_MESSAGES_LIMIT_EXCEEDED);
return;
}
if (totalSizeRead >= request.sizeLimit) {
request.callback.scanFinished(ctx, ReasonForFinish.SIZE_LIMIT_EXCEEDED);
return;
}
}
// continue scanning messages
scanMessages(request, imlr.getStartSeqIdIncluded() + entry.getEntryId() + 1, numMessagesRead, totalSizeRead);
}
}, request.ctx);
}