Skip to content
This repository was archived by the owner on May 28, 2018. It is now read-only.

Commit 74f7608

Browse files
mpotociarmpotociar
committed
Merge pull request #155 from luengnat/JERSEY-2837
 JERSEY-2837: Correctly handle negative byte values
2 parents 5d26e02 + cb34eea commit 74f7608

File tree

2 files changed

+198
-89
lines changed

2 files changed

+198
-89
lines changed

‎core-common/src/main/java/org/glassfish/jersey/internal/util/collection/ByteBufferInputStream.java‎

Lines changed: 80 additions & 81 deletions
Original file line numberDiff line numberDiff line change
@@ -162,53 +162,16 @@ public int available() throws IOException {
162162

163163
@Override
164164
public int read() throws IOException {
165-
if (eof) {
166-
checkThrowable();
167-
checkNotClosed();
168-
return -1;
169-
}
170-
171-
final int c;
172-
if (current != null && current.hasRemaining()) {
173-
c = current.get();
174-
} else {
175-
try {
176-
// let's block until next non-empty chunk or EOF
177-
c = fetchChunk(true) ? current.get() : -1;
178-
} catch (final InterruptedException e) {
179-
Thread.currentThread().interrupt();
180-
throw new IOException(e);
181-
}
182-
}
183-
184-
checkThrowable();
185-
checkNotClosed();
186-
return c;
165+
return tryRead(true);
187166
}
188167

189168
@Override
190-
public int tryRead() throws IOException {
191-
checkThrowable();
192-
checkNotClosed();
193-
194-
if (eof) {
195-
return -1;
196-
}
197-
198-
if (current != null && current.hasRemaining()) {
199-
return current.get();
200-
}
201-
202-
try {
203-
// try to fetch, but don't block && check if something has been fetched
204-
if (fetchChunk(false) && current != null) {
205-
return current.get();
206-
}
207-
} catch (final InterruptedException e) {
208-
Thread.currentThread().interrupt();
209-
}
169+
public int read(byte[] b, int off, int len) throws IOException {
170+
return tryRead(b, off, len, true);
171+
}
210172

211-
return (eof) ? -1 : NOTHING;
173+
public int tryRead() throws IOException {
174+
return tryRead(false);
212175
}
213176

214177
@Override
@@ -218,44 +181,7 @@ public int tryRead(final byte[] b) throws IOException {
218181

219182
@Override
220183
public int tryRead(final byte[] b, final int off, final int len) throws IOException {
221-
checkThrowable();
222-
checkNotClosed();
223-
224-
if (b == null) {
225-
throw new NullPointerException();
226-
} else if (off < 0 || len < 0 || len > b.length - off) {
227-
throw new IndexOutOfBoundsException();
228-
} else if (len == 0) {
229-
return 0;
230-
}
231-
232-
if (eof) {
233-
return -1;
234-
}
235-
236-
int i = 0;
237-
while (i < len) {
238-
if (current != null && current.hasRemaining()) {
239-
final int available = current.remaining();
240-
if (available < len - i) {
241-
current.get(b, off + i, available);
242-
i += available;
243-
} else {
244-
current.get(b, off + i, len - i);
245-
return len;
246-
}
247-
} else {
248-
try {
249-
if (!fetchChunk(false) || current == null) {
250-
break; // eof or no data
251-
}
252-
} catch (final InterruptedException e) {
253-
Thread.currentThread().interrupt();
254-
}
255-
}
256-
}
257-
258-
return i;
184+
return tryRead(b, off, len, false);
259185
}
260186

261187
@Override
@@ -335,4 +261,77 @@ public void closeQueue(final Throwable throwable) {
335261
}
336262
}
337263
}
264+
265+
private int tryRead(final byte[] b, final int off, final int len, boolean block) throws IOException {
266+
checkThrowable();
267+
checkNotClosed();
268+
269+
if (b == null) {
270+
throw new NullPointerException();
271+
} else if (off < 0 || len < 0 || len > b.length - off) {
272+
throw new IndexOutOfBoundsException();
273+
} else if (len == 0) {
274+
return 0;
275+
}
276+
277+
if (eof) {
278+
return -1;
279+
}
280+
281+
int i = 0;
282+
while (i < len) {
283+
if (current != null && current.hasRemaining()) {
284+
final int available = current.remaining();
285+
if (available < len - i) {
286+
current.get(b, off + i, available);
287+
i += available;
288+
} else {
289+
current.get(b, off + i, len - i);
290+
return len;
291+
}
292+
} else {
293+
try {
294+
if (!fetchChunk(block) || current == null) {
295+
break; // eof or no data
296+
}
297+
} catch (final InterruptedException e) {
298+
Thread.currentThread().interrupt();
299+
if (block) {
300+
throw new IOException(e);
301+
}
302+
}
303+
}
304+
}
305+
306+
return i == 0 && eof ? -1 : i;
307+
}
308+
309+
private int tryRead(boolean block) throws IOException {
310+
checkThrowable();
311+
checkNotClosed();
312+
313+
if (eof) {
314+
return -1;
315+
}
316+
317+
if (current != null && current.hasRemaining()) {
318+
return current.get() & 0xFF;
319+
}
320+
321+
try {
322+
// try to fetch, but don't block && check if something has been fetched
323+
if (fetchChunk(block) && current != null) {
324+
return current.get() & 0xFF;
325+
} else if (block) {
326+
return -1;
327+
}
328+
} catch (final InterruptedException e) {
329+
Thread.currentThread().interrupt();
330+
if (block) {
331+
throw new IOException(e);
332+
}
333+
}
334+
335+
return (eof) ? -1 : NOTHING;
336+
}
338337
}

‎core-common/src/test/java/org/glassfish/jersey/internal/util/collection/ByteBufferInputStreamTest.java‎

Lines changed: 118 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -41,9 +41,12 @@
4141

4242
import java.io.IOException;
4343
import java.nio.ByteBuffer;
44+
import java.util.Random;
4445
import java.util.concurrent.ExecutorService;
4546
import java.util.concurrent.Executors;
47+
import java.util.concurrent.Semaphore;
4648
import java.util.concurrent.TimeUnit;
49+
import java.util.concurrent.atomic.AtomicBoolean;
4750

4851
import org.glassfish.jersey.internal.LocalizationMessages;
4952

@@ -52,6 +55,7 @@
5255
import static org.junit.Assert.assertNotEquals;
5356
import static org.junit.Assert.assertNotNull;
5457
import static org.junit.Assert.assertNull;
58+
import static org.junit.Assert.assertTrue;
5559
import static org.junit.Assert.fail;
5660

5761
/**
@@ -61,6 +65,112 @@
6165
*/
6266
public class ByteBufferInputStreamTest {
6367

68+
@Test
69+
public void testBlockingReadAByteEmptyStream() throws Exception {
70+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
71+
bbis.closeQueue();
72+
assertEquals(-1, bbis.read());
73+
}
74+
75+
@Test
76+
public void testNonBlockingReadAByteEmptyStream() throws Exception {
77+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
78+
bbis.closeQueue();
79+
assertEquals(-1, bbis.tryRead());
80+
}
81+
82+
@Test
83+
public void testBlockingReadByteArrayEmptyStream() throws Exception {
84+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
85+
bbis.closeQueue();
86+
byte[] buf = new byte[1024];
87+
assertEquals(-1, bbis.read(buf));
88+
}
89+
90+
@Test
91+
public void testNonBlockingReadByteArrayEmptyStream() throws Exception {
92+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
93+
bbis.closeQueue();
94+
byte[] buf = new byte[1024];
95+
assertEquals(-1, bbis.tryRead(buf));
96+
}
97+
98+
@Test
99+
public void testBlockingReadByteArrayFromFinishedExactLengthStream() throws Exception {
100+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
101+
byte[] sourceData = new byte[1024];
102+
new Random().nextBytes(sourceData);
103+
ByteBuffer byteBuf = ByteBuffer.wrap(sourceData);
104+
bbis.put(byteBuf);
105+
bbis.closeQueue();
106+
byte[] buf = new byte[1024];
107+
assertEquals(1024, bbis.read(buf));
108+
// no more data to read; so it should return -1
109+
assertEquals(-1, bbis.read(buf));
110+
}
111+
112+
@Test
113+
public void testNonBlockingReadByteArrayFromFinishedExactLengthStream() throws Exception {
114+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
115+
byte[] sourceData = new byte[1024];
116+
new Random().nextBytes(sourceData);
117+
ByteBuffer byteBuf = ByteBuffer.wrap(sourceData);
118+
bbis.put(byteBuf);
119+
byte[] buf = new byte[1024];
120+
assertEquals(1024, bbis.tryRead(buf));
121+
// the queue has not been close; so it should return 0
122+
assertEquals(0, bbis.tryRead(buf));
123+
bbis.closeQueue();
124+
}
125+
126+
@Test
127+
public void testBlockingReadByteArrayFromUnfinishedExactLengthStream() throws Exception {
128+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
129+
byte[] sourceData = new byte[1024];
130+
new Random().nextBytes(sourceData);
131+
ByteBuffer byteBuf = ByteBuffer.wrap(sourceData);
132+
bbis.put(byteBuf);
133+
final byte[] buf = new byte[1024];
134+
assertEquals(1024, bbis.read(buf));
135+
final AtomicBoolean closed = new AtomicBoolean(false);
136+
final Semaphore s = new Semaphore(1);
137+
s.acquire();
138+
Thread t = new Thread(new Runnable() {
139+
@Override
140+
public void run() {
141+
try {
142+
// it should return -1 since there is no more data
143+
assertEquals(-1, bbis.read(buf));
144+
// it should only reach here if the stream has been closed
145+
assertTrue(closed.get());
146+
} catch (IOException e) {
147+
e.printStackTrace();
148+
} finally {
149+
s.release();
150+
}
151+
}
152+
});
153+
t.start();
154+
Thread.sleep(500);
155+
closed.set(true);
156+
bbis.closeQueue();
157+
// wait until the job is done
158+
s.acquire();
159+
}
160+
161+
@Test
162+
public void testNonBlockingReadByteArrayFromUnfinishedExactLengthStream() throws Exception {
163+
final ByteBufferInputStream bbis = new ByteBufferInputStream();
164+
byte[] sourceData = new byte[1024];
165+
new Random().nextBytes(sourceData);
166+
ByteBuffer byteBuf = ByteBuffer.wrap(sourceData);
167+
bbis.put(byteBuf);
168+
bbis.closeQueue();
169+
byte[] buf = new byte[1024];
170+
assertEquals(1024, bbis.tryRead(buf));
171+
assertEquals(-1, bbis.tryRead(buf));
172+
}
173+
64174
/**
65175
* Test for non blocking single-byte read of the stream.
66176
*
@@ -85,7 +195,7 @@ public void run() {
85195
}
86196
data.clear();
87197
for (int j = 0; j < data.capacity(); j++) {
88-
data.put((byte) (i % 128));
198+
data.put((byte) (i & 0xFF));
89199
}
90200
data.flip();
91201
if (!bbis.put(data)) {
@@ -113,7 +223,7 @@ public void run() {
113223
Thread.yield(); // Give the other thread a chance to run.
114224
continue;
115225
}
116-
assertEquals("At position: " + j, (byte) (i % 128), c);
226+
assertEquals("At position: " + j, (byte) (i & 0xFF), (byte) (c & 0xFF));
117227
if (++j % BUFFER_SIZE == 0) {
118228
i++;
119229
Thread.yield(); // Give the other thread a chance to run.
@@ -155,7 +265,7 @@ public void run() {
155265
}
156266
data.clear();
157267
for (int j = 0; j < data.capacity(); j++) {
158-
data.put((byte) (i % 128));
268+
data.put((byte) (i & 0xFF));
159269
}
160270
data.flip();
161271
if (!bbis.put(data)) {
@@ -185,7 +295,7 @@ public void run() {
185295
continue;
186296
}
187297
for (int p = 0; p < c; p++) {
188-
assertEquals("At position: " + j, (byte) (i % 128), buffer[p]);
298+
assertEquals("At position: " + j, (byte) (i & 0xFF), (byte) buffer[p]);
189299
if (++j % BUFFER_SIZE == 0) {
190300
i++;
191301
Thread.yield(); // Give the other thread a chance to run.
@@ -228,7 +338,7 @@ public void run() {
228338
}
229339
data.clear();
230340
for (int j = 0; j < data.capacity(); j++) {
231-
data.put((byte) (i % 128));
341+
data.put((byte) (i & 0xFF));
232342
}
233343
data.flip();
234344
if (!bbis.put(data)) {
@@ -253,7 +363,7 @@ public void run() {
253363
while ((c = bbis.read()) != -1) {
254364
assertNotEquals("Should not read 'nothing' in blocking mode.", Integer.MIN_VALUE, c);
255365

256-
assertEquals("At position: " + j, (byte) (i % 128), c);
366+
assertEquals("At position: " + j, (byte) (i & 0xFF), (byte) c);
257367
if (++j % BUFFER_SIZE == 0) {
258368
i++;
259369
Thread.yield(); // Give the other thread a chance to run.
@@ -295,7 +405,7 @@ public void run() {
295405
}
296406
data.clear();
297407
for (int j = 0; j < data.capacity(); j++) {
298-
data.put((byte) (i % 128));
408+
data.put((byte) (i & 0xFF));
299409
}
300410
data.flip();
301411
if (!bbis.put(data)) {
@@ -322,7 +432,7 @@ public void run() {
322432
assertNotEquals("Should not read 0 bytes in blocking mode.", 0, c);
323433

324434
for (int p = 0; p < c; p++) {
325-
assertEquals("At position: " + j, (byte) (i % 128), buffer[p]);
435+
assertEquals("At position: " + j, (byte) (i & 0xFF), buffer[p]);
326436
if (++j % BUFFER_SIZE == 0) {
327437
i++;
328438
Thread.yield(); // Give the other thread a chance to run.

0 commit comments

Comments
 (0)