Skip to content

Commit 094733e

Browse files
niloc132niloc132
committed
Buffer framed payload writes, check for KnownLength
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 5c5ef47 commit 094733e

6 files changed

Lines changed: 364 additions & 19 deletions

File tree

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

Lines changed: 15 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,10 @@
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;
31-
import java.util.ArrayList;
3232
import java.util.List;
3333
import java.util.concurrent.atomic.AtomicReference;
3434
import java.util.logging.Logger;
@@ -369,25 +369,14 @@ public String authority() {
369369
* @param <RequestT> the request type
370370
*/
371371
private static <RequestT> List<Uint8Array> frame(final RequestT message, final MethodDescriptor.Marshaller<RequestT> requestMarshaller) {
372-
final List<Uint8Array> result = new ArrayList<>();
372+
final List<Uint8Array> result;
373373
try (final InputStream stream = requestMarshaller.stream(message)) {
374+
int length = getKnownLength(stream);
375+
final ByteBufferOutputStream bufferingStream = new ByteBufferOutputStream(length == -1 ? 4096 : length);
376+
// TODO also support transferTo, when GWT does
374377
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-
});
378+
drainable.drainTo(bufferingStream);
379+
result = bufferingStream.getBuffers();
391380
} catch (IOException e) {
392381
throw new RuntimeException(e);
393382
}
@@ -401,6 +390,13 @@ public void write(final byte[] b, final int off, final int len) {
401390
return result;
402391
}
403392

393+
private static int getKnownLength(InputStream inputStream) throws IOException {
394+
if (inputStream instanceof KnownLength || inputStream instanceof ByteArrayInputStream) {
395+
return inputStream.available();
396+
}
397+
return -1;
398+
}
399+
404400
private JsPropertyMap<String> makeHeaders(final Metadata metadata) {
405401
final JsPropertyMap<String> result = JsPropertyMap.of();
406402
final byte[][] bytes = InternalMetadata.serialize(metadata);

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

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
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.
81+
* Intended to be called after all writes are complete.
82+
*/
83+
public List<Uint8Array> getBuffers() {
84+
List<Uint8Array> result = new ArrayList<>(buffers.size());
85+
for (ByteBuffer buffer : buffers) {
86+
if (buffer.position() == 0) {
87+
continue;
88+
}
89+
result.add(asUint8Array(buffer));
90+
}
91+
return Collections.unmodifiableList(result);
92+
}
93+
94+
private Uint8Array asUint8Array(ByteBuffer buffer) {
95+
ByteBuffer readable = buffer.duplicate();
96+
readable.flip();
97+
ArrayBufferView view = TypedArrayHelper.unwrap(readable.slice());
98+
return new Uint8Array(view.buffer, view.byteOffset, view.byteLength);
99+
}
100+
}

‎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)