@Test public void testDecode() { StringDecoder stringDecoder = new StringDecoder(5000L); Flux<DataBuffer> pub = Flux.just("abc\n", "abc", "def\n", "abc", "def\nxyz\n", "abc", "def", "xyz\n") .map(s -> dataBufferFactory.wrap(s.getBytes(UTF_8))); StepVerifier.create(stringDecoder.decode(pub, null, null, null)) .expectNext("abc", "abcdef", "abcdef", "xyz", "abcdefxyz") .verifyComplete(); }
@Override public Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) { DataBuffer incomplete = new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT).allocateBuffer(0); return Flux.from(publisher) .scan(Tuples.of(Flux.<DataBuffer>empty(), retain(incomplete), Instant.now()), (acc, buffer) -> { List<DataBuffer> results = new ArrayList<>(); int startIdx = 0, endIdx = 0, limit = buffer.readableByteCount(); Instant lastEmissionTime = acc.getT3(); while (startIdx < limit && endIdx != -1) { endIdx = buffer.indexOf(NEWLINE_DELIMITER, startIdx); int length = (endIdx == -1 ? limit : endIdx) - startIdx; DataBuffer slice = buffer.slice(startIdx, length); byte[] slice1 = new byte[length]; slice.read(slice1, 0, slice1.length); Instant now = Instant.now(); boolean maxDurationElapsed = Duration.between(lastEmissionTime, now).toMillis() >= maxDurationMillis; if (endIdx != -1 || maxDurationElapsed) { byte[] slice2 = new byte[incomplete.readableByteCount()]; incomplete.read(slice2, 0, slice2.length); results.add(retain( incomplete.factory().allocateBuffer() .write(slice2) .write(slice1) )); lastEmissionTime = now; if (endIdx != -1) { startIdx = endIdx + 1; } } else { incomplete.write(slice1); } } return Tuples.of(Flux.fromIterable(results), incomplete, lastEmissionTime); }) .flatMap(Tuple2::getT1) .map(buffer -> { String s = UTF_8.decode(buffer.asByteBuffer()).toString(); return s; }) .doOnTerminate(() -> release(incomplete)) .log(); }
StringDecoder extends AbstractDecoder<String> { @Override public Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) { DataBuffer incomplete = new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT).allocateBuffer(0); return Flux.from(publisher) .scan(Tuples.of(Flux.<DataBuffer>empty(), retain(incomplete), Instant.now()), (acc, buffer) -> { List<DataBuffer> results = new ArrayList<>(); int startIdx = 0, endIdx = 0, limit = buffer.readableByteCount(); Instant lastEmissionTime = acc.getT3(); while (startIdx < limit && endIdx != -1) { endIdx = buffer.indexOf(NEWLINE_DELIMITER, startIdx); int length = (endIdx == -1 ? limit : endIdx) - startIdx; DataBuffer slice = buffer.slice(startIdx, length); byte[] slice1 = new byte[length]; slice.read(slice1, 0, slice1.length); Instant now = Instant.now(); boolean maxDurationElapsed = Duration.between(lastEmissionTime, now).toMillis() >= maxDurationMillis; if (endIdx != -1 || maxDurationElapsed) { byte[] slice2 = new byte[incomplete.readableByteCount()]; incomplete.read(slice2, 0, slice2.length); results.add(retain( incomplete.factory().allocateBuffer() .write(slice2) .write(slice1) )); lastEmissionTime = now; if (endIdx != -1) { startIdx = endIdx + 1; } } else { incomplete.write(slice1); } } return Tuples.of(Flux.fromIterable(results), incomplete, lastEmissionTime); }) .flatMap(Tuple2::getT1) .map(buffer -> { String s = UTF_8.decode(buffer.asByteBuffer()).toString(); return s; }) .doOnTerminate(() -> release(incomplete)) .log(); } }
StringDecoder extends AbstractDecoder<String> { @Override public Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) { DataBuffer incomplete = new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT).allocateBuffer(0); return Flux.from(publisher) .scan(Tuples.of(Flux.<DataBuffer>empty(), retain(incomplete), Instant.now()), (acc, buffer) -> { List<DataBuffer> results = new ArrayList<>(); int startIdx = 0, endIdx = 0, limit = buffer.readableByteCount(); Instant lastEmissionTime = acc.getT3(); while (startIdx < limit && endIdx != -1) { endIdx = buffer.indexOf(NEWLINE_DELIMITER, startIdx); int length = (endIdx == -1 ? limit : endIdx) - startIdx; DataBuffer slice = buffer.slice(startIdx, length); byte[] slice1 = new byte[length]; slice.read(slice1, 0, slice1.length); Instant now = Instant.now(); boolean maxDurationElapsed = Duration.between(lastEmissionTime, now).toMillis() >= maxDurationMillis; if (endIdx != -1 || maxDurationElapsed) { byte[] slice2 = new byte[incomplete.readableByteCount()]; incomplete.read(slice2, 0, slice2.length); results.add(retain( incomplete.factory().allocateBuffer() .write(slice2) .write(slice1) )); lastEmissionTime = now; if (endIdx != -1) { startIdx = endIdx + 1; } } else { incomplete.write(slice1); } } return Tuples.of(Flux.fromIterable(results), incomplete, lastEmissionTime); }) .flatMap(Tuple2::getT1) .map(buffer -> { String s = UTF_8.decode(buffer.asByteBuffer()).toString(); return s; }) .doOnTerminate(() -> release(incomplete)) .log(); } StringDecoder(long maxDurationMillis); }
StringDecoder extends AbstractDecoder<String> { @Override public Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) { DataBuffer incomplete = new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT).allocateBuffer(0); return Flux.from(publisher) .scan(Tuples.of(Flux.<DataBuffer>empty(), retain(incomplete), Instant.now()), (acc, buffer) -> { List<DataBuffer> results = new ArrayList<>(); int startIdx = 0, endIdx = 0, limit = buffer.readableByteCount(); Instant lastEmissionTime = acc.getT3(); while (startIdx < limit && endIdx != -1) { endIdx = buffer.indexOf(NEWLINE_DELIMITER, startIdx); int length = (endIdx == -1 ? limit : endIdx) - startIdx; DataBuffer slice = buffer.slice(startIdx, length); byte[] slice1 = new byte[length]; slice.read(slice1, 0, slice1.length); Instant now = Instant.now(); boolean maxDurationElapsed = Duration.between(lastEmissionTime, now).toMillis() >= maxDurationMillis; if (endIdx != -1 || maxDurationElapsed) { byte[] slice2 = new byte[incomplete.readableByteCount()]; incomplete.read(slice2, 0, slice2.length); results.add(retain( incomplete.factory().allocateBuffer() .write(slice2) .write(slice1) )); lastEmissionTime = now; if (endIdx != -1) { startIdx = endIdx + 1; } } else { incomplete.write(slice1); } } return Tuples.of(Flux.fromIterable(results), incomplete, lastEmissionTime); }) .flatMap(Tuple2::getT1) .map(buffer -> { String s = UTF_8.decode(buffer.asByteBuffer()).toString(); return s; }) .doOnTerminate(() -> release(incomplete)) .log(); } StringDecoder(long maxDurationMillis); @Override Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints); }
StringDecoder extends AbstractDecoder<String> { @Override public Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints) { DataBuffer incomplete = new NettyDataBufferFactory(UnpooledByteBufAllocator.DEFAULT).allocateBuffer(0); return Flux.from(publisher) .scan(Tuples.of(Flux.<DataBuffer>empty(), retain(incomplete), Instant.now()), (acc, buffer) -> { List<DataBuffer> results = new ArrayList<>(); int startIdx = 0, endIdx = 0, limit = buffer.readableByteCount(); Instant lastEmissionTime = acc.getT3(); while (startIdx < limit && endIdx != -1) { endIdx = buffer.indexOf(NEWLINE_DELIMITER, startIdx); int length = (endIdx == -1 ? limit : endIdx) - startIdx; DataBuffer slice = buffer.slice(startIdx, length); byte[] slice1 = new byte[length]; slice.read(slice1, 0, slice1.length); Instant now = Instant.now(); boolean maxDurationElapsed = Duration.between(lastEmissionTime, now).toMillis() >= maxDurationMillis; if (endIdx != -1 || maxDurationElapsed) { byte[] slice2 = new byte[incomplete.readableByteCount()]; incomplete.read(slice2, 0, slice2.length); results.add(retain( incomplete.factory().allocateBuffer() .write(slice2) .write(slice1) )); lastEmissionTime = now; if (endIdx != -1) { startIdx = endIdx + 1; } } else { incomplete.write(slice1); } } return Tuples.of(Flux.fromIterable(results), incomplete, lastEmissionTime); }) .flatMap(Tuple2::getT1) .map(buffer -> { String s = UTF_8.decode(buffer.asByteBuffer()).toString(); return s; }) .doOnTerminate(() -> release(incomplete)) .log(); } StringDecoder(long maxDurationMillis); @Override Flux<String> decode(Publisher<DataBuffer> publisher, ResolvableType elementType, MimeType mimeType, Map<String, Object> hints);  }