@Override
    public Iterator<E> iterator() {
        return new AbstractIterator<E>() {
            private boolean endOfChannel = false;

            @Override
            protected E computeNext() {
                if (!buffer.hasRemaining()) {
                    if (endOfChannel) {
                        return endOfData();
                    }
                    buffer.rewind();
                    setMaxLimit();
                    int totalReadBytes = 0;
                    int readBytes;
                    do {
                        try {
                            readBytes = channel.read(buffer);
                            if (readBytes < 0) {
                                endOfChannel = true;
                            } else {
                                totalReadBytes += readBytes;
                            }
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    } while (readBytes > 0);
                    buffer.rewind();
                    buffer.limit(totalReadBytes);
                    Preconditions.checkArgument(totalReadBytes % serializer.elementSize() == 0,
                            totalReadBytes + " " + serializer.elementSize());
                }
                if (!buffer.hasRemaining()) {
                    if (endOfChannel) {
                        return endOfData();
                    }
                }
                return Utils.readElement(serializer, buffer);

            }
        };
    }