Skip to content

Commit 88d26d9

Browse files
niloc132niloc132
authored
Buffer framed payload writes, check for KnownLength (#6)
Reduces the number of Transport.write calls to send out 4k buffers when the size isn't known or just a large payload when it is. Fixes #5
1 parent e6bd6e0 commit 88d26d9

6 files changed

Lines changed: 369 additions & 19 deletions

File tree

‎grpc-web-gwt-fetch/src/main/java/com/vertispan/grpc/fetch/AbstractGrpcWebChannel.java‎

Lines changed: 21 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import io.grpc.Drainable;
1717
import io.grpc.InternalMetadata;
1818
import io.grpc.InternalStatus;
19+
import io.grpc.KnownLength;
1920
import io.grpc.Metadata;
2021
import io.grpc.MethodDescriptor;
2122
import io.grpc.SecurityLevel;
@@ -24,11 +25,12 @@
2425
import org.gwtproject.nio.TypedArrayHelper;
2526

2627
import javax.annotation.Nullable;
28+
import java.io.ByteArrayInputStream;
2729
import java.io.IOException;
2830
import java.io.InputStream;
29-
import java.io.OutputStream;
3031
import java.nio.charset.StandardCharsets;
3132
import java.util.ArrayList;
33+
import java.util.Collections;
3234
import java.util.List;
3335
import java.util.concurrent.atomic.AtomicReference;
3436
import java.util.logging.Logger;
@@ -369,25 +371,18 @@ public String authority() {
369371
* @param <RequestT> the request type
370372
*/
371373
private static <RequestT> List<Uint8Array> frame(final RequestT message, final MethodDescriptor.Marshaller<RequestT> requestMarshaller) {
372-
final List<Uint8Array> result = new ArrayList<>();
374+
final List<Uint8Array> result;
373375
try (final InputStream stream = requestMarshaller.stream(message)) {
374-
final Drainable drainable = (Drainable) stream;
375-
drainable.drainTo(new OutputStream() {
376-
@Override
377-
public void write(final int b) {
378-
// TODO buffer this instead, for custom marshallers to write individual bytes
379-
write(new byte[]{(byte) b}, 0, 1);
380-
}
381-
382-
@Override
383-
public void write(final byte[] b, final int off, final int len) {
384-
final Uint8Array e = new Uint8Array(len);
385-
for (int i = 0; i < len; i++) {
386-
e.setAt(i, (double) b[off + i]);
387-
}
388-
result.add(e);
389-
}
390-
});
376+
int length = getKnownLength(stream);
377+
if (length == 0) {
378+
result = new ArrayList<>();
379+
} else {
380+
final ByteBufferOutputStream bufferingStream = new ByteBufferOutputStream(length == -1 ? 4096 : length);
381+
// TODO also support transferTo, when GWT does
382+
final Drainable drainable = (Drainable) stream;
383+
drainable.drainTo(bufferingStream);
384+
result = bufferingStream.getBuffers();
385+
}
391386
} catch (IOException e) {
392387
throw new RuntimeException(e);
393388
}
@@ -401,6 +396,13 @@ public void write(final byte[] b, final int off, final int len) {
401396
return result;
402397
}
403398

399+
private static int getKnownLength(InputStream inputStream) throws IOException {
400+
if (inputStream instanceof KnownLength || inputStream instanceof ByteArrayInputStream) {
401+
return inputStream.available();
402+
}
403+
return -1;
404+
}
405+
404406
private JsPropertyMap<String> makeHeaders(final Metadata metadata) {
405407
final JsPropertyMap<String> result = JsPropertyMap.of();
406408
final byte[][] bytes = InternalMetadata.serialize(metadata);

‎grpc-web-gwt-fetch/src/main/java/com/vertispan/grpc/fetch/ByteBufferOutputStream.java‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package com.vertispan.grpc.fetch;
2+
3+
import com.google.common.annotations.VisibleForTesting;
4+
import elemental2.core.ArrayBufferView;
5+
import elemental2.core.Uint8Array;
6+
import org.gwtproject.nio.TypedArrayHelper;
7+
8+
import java.io.IOException;
9+
import java.io.OutputStream;
10+
import java.nio.ByteBuffer;
11+
import java.util.ArrayList;
12+
import java.util.Collections;
13+
import java.util.List;
14+
15+
/**
16+
* OutputStream implementation that writes to bytebuffers (backed by typed arrays) of a minimum size. Larger buffers
17+
* may be created if the written data requires it.
18+
*/
19+
public class ByteBufferOutputStream extends OutputStream {
20+
@VisibleForTesting
21+
final List<ByteBuffer> buffers = new ArrayList<>();
22+
private final int bufferSize;
23+
24+
public ByteBufferOutputStream(int bufferSize) {
25+
if (bufferSize <= 0) {
26+
throw new IllegalArgumentException("bufferSize must be positive");
27+
}
28+
this.bufferSize = bufferSize;
29+
buffers.add(ByteBuffer.allocate(bufferSize));
30+
}
31+
32+
@Override
33+
public void write(int i) throws IOException {
34+
ensureRemaining(1).put((byte) i);
35+
}
36+
37+
@Override
38+
public void write(byte[] b, int off, int len) throws IOException {
39+
if (b == null) {
40+
throw new NullPointerException();
41+
}
42+
if (off < 0 || len < 0 || len > b.length - off) {
43+
throw new IndexOutOfBoundsException();
44+
}
45+
if (len == 0) {
46+
return;
47+
}
48+
49+
ByteBuffer current = currentBuffer();
50+
if (len > current.remaining() && current.position() != 0) {
51+
int chunk = current.remaining();
52+
current.put(b, off, chunk);
53+
off += chunk;
54+
len -= chunk;
55+
}
56+
57+
ensureRemaining(len).put(b, off, len);
58+
}
59+
60+
private ByteBuffer ensureRemaining(int length) {
61+
ByteBuffer current = currentBuffer();
62+
if (current.remaining() >= length) {
63+
return current;
64+
}
65+
66+
ByteBuffer expanded = ByteBuffer.allocate(Math.max(bufferSize, length));
67+
if (current.position() == 0) {
68+
buffers.set(buffers.size() - 1, expanded);
69+
} else {
70+
buffers.add(expanded);
71+
}
72+
return expanded;
73+
}
74+
75+
private ByteBuffer currentBuffer() {
76+
return buffers.get(buffers.size() - 1);
77+
}
78+
79+
/**
80+
* Returns the written contents as typed arrays. Intended to be called after all writes are complete.
81+
*/
82+
public List<Uint8Array> getBuffers() {
83+
List<Uint8Array> result = new ArrayList<>(buffers.size());
84+
for (ByteBuffer buffer : buffers) {
85+
if (buffer.position() == 0) {
86+
continue;
87+
}
88+
result.add(asUint8Array(buffer));
89+
}
90+
return result;
91+
}
92+
93+
private Uint8Array asUint8Array(ByteBuffer buffer) {
94+
ByteBuffer readable = buffer.duplicate();
95+
readable.flip();
96+
ArrayBufferView view = TypedArrayHelper.unwrap(readable.slice());
97+
return new Uint8Array(view.buffer, view.byteOffset, view.byteLength);
98+
}
99+
}

‎grpc-web-gwt-fetch/src/test/java/com/vertispan/grpc/fetch/ByteBufferOutputStreamGwtTest.java‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
package com.vertispan.grpc.fetch;
2+
3+
import com.google.gwt.junit.client.GWTTestCase;
4+
import elemental2.core.Uint8Array;
5+
6+
import java.io.IOException;
7+
import java.util.List;
8+
9+
/**
10+
* Browser-emulation behavior tests for {@link ByteBufferOutputStream}, including Uint8Array export semantics.
11+
* Also includes delegate stubs for JVM-only ByteBufferOutputStream unit checks.
12+
*/
13+
public class ByteBufferOutputStreamGwtTest extends GWTTestCase {
14+
@Override
15+
public String getModuleName() {
16+
return "com.vertispan.grpc.fetch.Fetch";
17+
}
18+
19+
public void testByteBufferOutputStreamExportsWrittenBytes() throws IOException {
20+
ByteBufferOutputStream stream = new ByteBufferOutputStream(4);
21+
stream.write(new byte[]{1, 2}, 0, 2);
22+
stream.write(new byte[]{3, 4, 5, 6, 7, 8, 9, 10, 11, 12}, 0, 10);
23+
24+
List<Uint8Array> buffers = stream.getBuffers();
25+
assertEquals(2, buffers.size());
26+
27+
Uint8Array first = buffers.get(0);
28+
assertEquals(4, first.length);
29+
assertEquals(1, first.getAt(0).intValue());
30+
assertEquals(2, first.getAt(1).intValue());
31+
assertEquals(3, first.getAt(2).intValue());
32+
assertEquals(4, first.getAt(3).intValue());
33+
34+
Uint8Array second = buffers.get(1);
35+
assertEquals(8, second.length);
36+
assertEquals(5, second.getAt(0).intValue());
37+
assertEquals(6, second.getAt(1).intValue());
38+
assertEquals(7, second.getAt(2).intValue());
39+
assertEquals(8, second.getAt(3).intValue());
40+
assertEquals(9, second.getAt(4).intValue());
41+
assertEquals(10, second.getAt(5).intValue());
42+
assertEquals(11, second.getAt(6).intValue());
43+
assertEquals(12, second.getAt(7).intValue());
44+
}
45+
46+
public void testByteBufferOutputStreamTruncatesFinalExportedBufferLength() throws IOException {
47+
ByteBufferOutputStream stream = new ByteBufferOutputStream(8);
48+
stream.write(new byte[]{1, 2, 3, 4, 5, 6}, 0, 6);
49+
stream.write(new byte[]{7, 8, 9, 10, 11}, 0, 5);
50+
51+
List<Uint8Array> buffers = stream.getBuffers();
52+
assertEquals(2, buffers.size());
53+
54+
Uint8Array first = buffers.get(0);
55+
assertEquals(8, first.length);
56+
assertEquals(1, first.getAt(0).intValue());
57+
assertEquals(8, first.getAt(7).intValue());
58+
59+
Uint8Array second = buffers.get(1);
60+
assertEquals(3, second.length);
61+
assertEquals(9, second.getAt(0).intValue());
62+
assertEquals(10, second.getAt(1).intValue());
63+
assertEquals(11, second.getAt(2).intValue());
64+
}
65+
66+
public void testByteBufferOutputStreamExportsEmptyWhenNothingWritten() {
67+
ByteBufferOutputStream stream = new ByteBufferOutputStream(8);
68+
assertTrue(stream.getBuffers().isEmpty());
69+
}
70+
71+
public void testByteBufferOutputStreamPreservesUnsignedByteValues() throws IOException {
72+
ByteBufferOutputStream stream = new ByteBufferOutputStream(8);
73+
stream.write(new byte[]{-1, -128, 127}, 0, 3);
74+
75+
List<Uint8Array> buffers = stream.getBuffers();
76+
assertEquals(1, buffers.size());
77+
assertEquals(255, buffers.get(0).getAt(0).intValue());
78+
assertEquals(128, buffers.get(0).getAt(1).intValue());
79+
assertEquals(127, buffers.get(0).getAt(2).intValue());
80+
}
81+
82+
public void testByteBufferOutputStreamFillsExactlyWithWriteInt() throws IOException {
83+
ByteBufferOutputStream stream = new ByteBufferOutputStream(4);
84+
stream.write(1);
85+
stream.write(2);
86+
stream.write(3);
87+
stream.write(4);
88+
89+
List<Uint8Array> buffers = stream.getBuffers();
90+
assertEquals(1, buffers.size());
91+
assertEquals(4, buffers.get(0).length);
92+
assertEquals(1, buffers.get(0).getAt(0).intValue());
93+
assertEquals(2, buffers.get(0).getAt(1).intValue());
94+
assertEquals(3, buffers.get(0).getAt(2).intValue());
95+
assertEquals(4, buffers.get(0).getAt(3).intValue());
96+
}
97+
98+
public void testByteBufferOutputStreamFillsExactlyWithWriteArray() throws IOException {
99+
ByteBufferOutputStream stream = new ByteBufferOutputStream(4);
100+
stream.write(new byte[]{1, 2, 3, 4}, 0, 4);
101+
102+
List<Uint8Array> buffers = stream.getBuffers();
103+
assertEquals(1, buffers.size());
104+
assertEquals(4, buffers.get(0).length);
105+
assertEquals(1, buffers.get(0).getAt(0).intValue());
106+
assertEquals(2, buffers.get(0).getAt(1).intValue());
107+
assertEquals(3, buffers.get(0).getAt(2).intValue());
108+
assertEquals(4, buffers.get(0).getAt(3).intValue());
109+
}
110+
111+
public void testByteBufferOutputStreamRolloverAfterExactFill() throws IOException {
112+
ByteBufferOutputStream stream = new ByteBufferOutputStream(4);
113+
stream.write(new byte[]{1, 2, 3, 4}, 0, 4);
114+
stream.write(5);
115+
116+
List<Uint8Array> buffers = stream.getBuffers();
117+
assertEquals(2, buffers.size());
118+
assertEquals(4, buffers.get(0).length);
119+
assertEquals(1, buffers.get(1).length);
120+
assertEquals(5, buffers.get(1).getAt(0).intValue());
121+
}
122+
123+
public void testByteBufferOutputStreamLargeFirstWriteThenAppend() throws IOException {
124+
ByteBufferOutputStream stream = new ByteBufferOutputStream(4);
125+
stream.write(new byte[]{1, 2, 3, 4, 5, 6}, 0, 6);
126+
stream.write(new byte[]{7, 8}, 0, 2);
127+
128+
List<Uint8Array> buffers = stream.getBuffers();
129+
assertEquals(2, buffers.size());
130+
assertEquals(6, buffers.get(0).length);
131+
assertEquals(2, buffers.get(1).length);
132+
assertEquals(1, buffers.get(0).getAt(0).intValue());
133+
assertEquals(6, buffers.get(0).getAt(5).intValue());
134+
assertEquals(7, buffers.get(1).getAt(0).intValue());
135+
assertEquals(8, buffers.get(1).getAt(1).intValue());
136+
}
137+
138+
public void testDelegatesJvmSingleByteAndArrayCase() throws Exception {
139+
new ByteBufferOutputStreamTest().writesSingleByteAndArrayIntoCurrentBuffer();
140+
}
141+
142+
public void testDelegatesJvmOversizedRemainingCase() throws Exception {
143+
new ByteBufferOutputStreamTest().allocatesLargerBufferForRemainingBytes();
144+
}
145+
146+
public void testDelegatesJvmOversizedInitialWriteCase() throws Exception {
147+
new ByteBufferOutputStreamTest().replacesEmptyBufferForOversizedWrite();
148+
}
149+
150+
public void testDelegatesJvmBoundsValidationCase() {
151+
new ByteBufferOutputStreamTest().rejectsInvalidWriteBounds();
152+
}
153+
}

0 commit comments

Comments
 (0)