@Test public void readOneLine() throws Exception { reader = new ChunkReader(mockStream); when( mockStream.read( (byte [])Mockito.anyObject()) ).thenAnswer(new Answer<java.lang.Number>() { @Override public java.lang.Number answer(InvocationOnMock invocation) throws Throwable { byte[] buf = (byte[]) invocation.getArguments()[0]; byte [] source = "OneLine\nTwoLine\n".getBytes(); System.arraycopy(source, 0, buf, 0, source.length); return new java.lang.Byte(buf[0]); } }); ChunkWritable out = new ChunkWritable(); int maxBytesToConsume = 1024*1024; assertEquals("OneLine\n".length() , reader.readLine(out, maxBytesToConsume) ); assertEquals("OneLine\n", new String(out.box) ); assertEquals("TwoLine\n".length(), reader.readLine(out, maxBytesToConsume) ); assertEquals("TwoLine\n", new String(out.box) ); }
public int readLine(Writable str, int maxBytesToConsume) throws IOException { ChunkWritable cw = (ChunkWritable) str; List<Node> list = new LinkedList<Node>(); boolean newLine = false; long bytesConsumed = 0; do { int startPosn = bufferPosn; if (bufferPosn >= bufferLength) { startPosn = bufferPosn = 0; bufferLength = in.read(buffer); if (bufferLength <= 0) { break; } } for (; bufferPosn < bufferLength; ++bufferPosn) { if (buffer[bufferPosn] == LF) { newLine = true; ++bufferPosn; break; } } int readLength = bufferPosn - startPosn; bytesConsumed += readLength; if (readLength > 0) { Node nd = new Node(); nd.slice = new byte[readLength]; nd.len = readLength; System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len); list.add(nd); } } while (!newLine && bytesConsumed < maxBytesToConsume); if (list.size() > 0) { cw.box = new byte[(int) bytesConsumed]; int pos = 0; for (int i = 0; i < list.size(); i++) { Node n = list.get(i); System.arraycopy(n.slice, 0, cw.box, pos, n.len); pos += n.len; } } return (int) bytesConsumed; }
ChunkReader implements Closeable { public int readLine(Writable str, int maxBytesToConsume) throws IOException { ChunkWritable cw = (ChunkWritable) str; List<Node> list = new LinkedList<Node>(); boolean newLine = false; long bytesConsumed = 0; do { int startPosn = bufferPosn; if (bufferPosn >= bufferLength) { startPosn = bufferPosn = 0; bufferLength = in.read(buffer); if (bufferLength <= 0) { break; } } for (; bufferPosn < bufferLength; ++bufferPosn) { if (buffer[bufferPosn] == LF) { newLine = true; ++bufferPosn; break; } } int readLength = bufferPosn - startPosn; bytesConsumed += readLength; if (readLength > 0) { Node nd = new Node(); nd.slice = new byte[readLength]; nd.len = readLength; System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len); list.add(nd); } } while (!newLine && bytesConsumed < maxBytesToConsume); if (list.size() > 0) { cw.box = new byte[(int) bytesConsumed]; int pos = 0; for (int i = 0; i < list.size(); i++) { Node n = list.get(i); System.arraycopy(n.slice, 0, cw.box, pos, n.len); pos += n.len; } } return (int) bytesConsumed; } }
ChunkReader implements Closeable { public int readLine(Writable str, int maxBytesToConsume) throws IOException { ChunkWritable cw = (ChunkWritable) str; List<Node> list = new LinkedList<Node>(); boolean newLine = false; long bytesConsumed = 0; do { int startPosn = bufferPosn; if (bufferPosn >= bufferLength) { startPosn = bufferPosn = 0; bufferLength = in.read(buffer); if (bufferLength <= 0) { break; } } for (; bufferPosn < bufferLength; ++bufferPosn) { if (buffer[bufferPosn] == LF) { newLine = true; ++bufferPosn; break; } } int readLength = bufferPosn - startPosn; bytesConsumed += readLength; if (readLength > 0) { Node nd = new Node(); nd.slice = new byte[readLength]; nd.len = readLength; System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len); list.add(nd); } } while (!newLine && bytesConsumed < maxBytesToConsume); if (list.size() > 0) { cw.box = new byte[(int) bytesConsumed]; int pos = 0; for (int i = 0; i < list.size(); i++) { Node n = list.get(i); System.arraycopy(n.slice, 0, cw.box, pos, n.len); pos += n.len; } } return (int) bytesConsumed; } ChunkReader(InputStream in); }
ChunkReader implements Closeable { public int readLine(Writable str, int maxBytesToConsume) throws IOException { ChunkWritable cw = (ChunkWritable) str; List<Node> list = new LinkedList<Node>(); boolean newLine = false; long bytesConsumed = 0; do { int startPosn = bufferPosn; if (bufferPosn >= bufferLength) { startPosn = bufferPosn = 0; bufferLength = in.read(buffer); if (bufferLength <= 0) { break; } } for (; bufferPosn < bufferLength; ++bufferPosn) { if (buffer[bufferPosn] == LF) { newLine = true; ++bufferPosn; break; } } int readLength = bufferPosn - startPosn; bytesConsumed += readLength; if (readLength > 0) { Node nd = new Node(); nd.slice = new byte[readLength]; nd.len = readLength; System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len); list.add(nd); } } while (!newLine && bytesConsumed < maxBytesToConsume); if (list.size() > 0) { cw.box = new byte[(int) bytesConsumed]; int pos = 0; for (int i = 0; i < list.size(); i++) { Node n = list.get(i); System.arraycopy(n.slice, 0, cw.box, pos, n.len); pos += n.len; } } return (int) bytesConsumed; } ChunkReader(InputStream in); @Override void close(); int readChunk(Writable str, int maxBytesToConsume); int readLine(Writable str, int maxBytesToConsume); }
ChunkReader implements Closeable { public int readLine(Writable str, int maxBytesToConsume) throws IOException { ChunkWritable cw = (ChunkWritable) str; List<Node> list = new LinkedList<Node>(); boolean newLine = false; long bytesConsumed = 0; do { int startPosn = bufferPosn; if (bufferPosn >= bufferLength) { startPosn = bufferPosn = 0; bufferLength = in.read(buffer); if (bufferLength <= 0) { break; } } for (; bufferPosn < bufferLength; ++bufferPosn) { if (buffer[bufferPosn] == LF) { newLine = true; ++bufferPosn; break; } } int readLength = bufferPosn - startPosn; bytesConsumed += readLength; if (readLength > 0) { Node nd = new Node(); nd.slice = new byte[readLength]; nd.len = readLength; System.arraycopy(buffer, startPosn, nd.slice, 0, nd.len); list.add(nd); } } while (!newLine && bytesConsumed < maxBytesToConsume); if (list.size() > 0) { cw.box = new byte[(int) bytesConsumed]; int pos = 0; for (int i = 0; i < list.size(); i++) { Node n = list.get(i); System.arraycopy(n.slice, 0, cw.box, pos, n.len); pos += n.len; } } return (int) bytesConsumed; } ChunkReader(InputStream in); @Override void close(); int readChunk(Writable str, int maxBytesToConsume); int readLine(Writable str, int maxBytesToConsume); static final int DEFAULT_BUFFER_SIZE; }