Fixed skipping logic to process messages

When skipping over rows (e.g. early close of data reader), we now properly
handle messages that were previously dropped (e.g. notification).

Also refactored the code for less duplication

Thanks @Emill