stream.close();
buffer.flip();
DirectByteBufferInputStream inStream = new DirectByteBufferInputStream();
inStream.setBuffer(new SpilledByteBuffer(buffer, 400));
for (int i = 0; i < 100; ++i) {
intWritable.readFields(inStream);
assertEquals(i, intWritable.get());
}
assertFalse(inStream.hasDataToRead());
assertTrue(inStream.hasUnmarkData());
inStream.prepareForNext();
// push in another buffer and check if the unmarked data could be read.
buffer.clear();
stream = new DirectByteBufferOutputStream();
buffer = ByteBuffer.allocateDirect(2048);
stream.setBuffer(buffer);
for (int i = 0; i < 400; ++i) {
intWritable.set(i);
intWritable.write(stream);
}
stream.close();
buffer.flip();
inStream.setBuffer(new SpilledByteBuffer(buffer, 400));
// Read previous data
intWritable.readFields(inStream);
assertEquals(99, intWritable.get());
for (int i = 0; i < 100; ++i) {
intWritable.readFields(inStream);
assertEquals(i, intWritable.get());
}
assertFalse(inStream.hasDataToRead());
assertTrue(inStream.hasUnmarkData());
inStream.prepareForNext();
buffer.clear();
stream = new DirectByteBufferOutputStream();
stream.setBuffer(buffer);
for (int i = 0; i < 100; ++i) {
intWritable.set(i);
intWritable.write(stream);
}
stream.close();
buffer.flip();
inStream.setBuffer(new SpilledByteBuffer(buffer, 400));
// Read previous data with resized intermediate buffer
for (int i = 100; i < 400; ++i) {
intWritable.readFields(inStream);
assertEquals(i, intWritable.get());
}
for (int i = 0; i < 100; ++i) {
intWritable.readFields(inStream);
assertEquals(i, intWritable.get());
}
assertFalse(inStream.hasDataToRead());
assertFalse(inStream.hasUnmarkData());
}