Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replace Reactor Stream by Flux in tests when possible

Browse Source
This commit is contained in:
Sebastien Deleuze
2016-01-12 15:27:02 +01:00
parent d9b67f5e72
commit 99e51a9101
12 changed files with 97 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/ByteBufferDecoderTests.java
View File

@@ -18,13 +18,14 @@ package org.springframework.reactive.codec.decoder;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.ByteBufferDecoder;
@@ -48,9 +49,9 @@ public class ByteBufferDecoderTests {
public void decode() throws InterruptedException {
ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer();
ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer();
Stream<ByteBuffer> source = Streams.just(fooBuffer, barBuffer);
List<ByteBuffer> results = Streams.from(decoder.decode(source,
ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null)).toList().get();
Flux<ByteBuffer> source = Flux.just(fooBuffer, barBuffer);
Flux<ByteBuffer> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
List<ByteBuffer> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals(fooBuffer, results.get(0));
assertEquals(barBuffer, results.get(1));

11
spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JacksonJsonDecoderTests.java
View File

@@ -18,14 +18,15 @@ package org.springframework.reactive.codec.decoder;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.JacksonJsonDecoder;
@@ -47,9 +48,9 @@ public class JacksonJsonDecoderTests {
@Test
public void decode() throws InterruptedException {
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
List<Object> results = Streams.from(decoder.decode(source, ResolvableType.forClass(Pojo.class), null))
.toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
}

11
spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/Jaxb2DecoderTests.java
View File

@@ -18,14 +18,15 @@ package org.springframework.reactive.codec.decoder;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.Jaxb2Decoder;
@@ -48,9 +49,9 @@ public class Jaxb2DecoderTests {
@Test
public void decode() throws InterruptedException {
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>").byteBuffer());
List<Object> results = Streams.from(decoder.decode(source, ResolvableType.forClass(Pojo.class), null))
.toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("<?xml version=\"1.0\" encoding=\"UTF-8\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>").byteBuffer());
Flux<Object> output = decoder.decode(source, ResolvableType.forClass(Pojo.class), null);
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("foofoo", ((Pojo) results.get(0)).getFoo());
}

57
spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/JsonObjectDecoderTests.java
View File

@@ -19,12 +19,13 @@ package org.springframework.reactive.codec.decoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.codec.support.JsonObjectDecoder;
@@ -36,12 +37,13 @@ public class JsonObjectDecoderTests {
@Test
public void decodeSingleChunkToJsonObject() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
List<String> results = Streams.from(decoder.decode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<Object> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
}
@@ -49,12 +51,13 @@ public class JsonObjectDecoderTests {
@Test
public void decodeMultipleChunksToJsonObject() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("{\"foo\": \"foofoo\"").byteBuffer(), Buffer.wrap(", \"bar\": \"barbar\"}").byteBuffer());
List<String> results = Streams.from(decoder.decode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("{\"foo\": \"foofoo\"").byteBuffer(), Buffer.wrap(", \"bar\": \"barbar\"}").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
}
@@ -62,12 +65,13 @@ public class JsonObjectDecoderTests {
@Test
public void decodeSingleChunkToArray() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
List<String> results = Streams.from(decoder.decode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));
@@ -76,12 +80,13 @@ public class JsonObjectDecoderTests {
@Test
public void decodeMultipleChunksToArray() throws InterruptedException {
JsonObjectDecoder decoder = new JsonObjectDecoder();
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
List<String> results = Streams.from(decoder.decode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("[{\"foo\": \"foofoo\", \"bar\"").byteBuffer(), Buffer.wrap(": \"barbar\"},{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}]").byteBuffer());
Flux<String> output = decoder.decode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("{\"foo\": \"foofoo\", \"bar\": \"barbar\"}", results.get(0));
assertEquals("{\"foo\": \"foofoofoo\", \"bar\": \"barbarbar\"}", results.get(1));

11
spring-web-reactive/src/test/java/org/springframework/reactive/codec/decoder/StringDecoderTests.java
View File

@@ -18,15 +18,16 @@ package org.springframework.reactive.codec.decoder;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.StringDecoder;
@@ -48,9 +49,9 @@ public class StringDecoderTests {
@Test
public void decode() throws InterruptedException {
Stream<ByteBuffer> source = Streams.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
List<String> results = Streams.from(decoder.decode(source,
ResolvableType.forClassWithGenerics(Publisher.class, String.class), null)).toList().get();
Flux<ByteBuffer> source = Flux.just(Buffer.wrap("foo").byteBuffer(), Buffer.wrap("bar").byteBuffer());
Flux<String> output = decoder.decode(source, ResolvableType.forClassWithGenerics(Publisher.class, String.class), null);
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("foo", results.get(0));
assertEquals("bar", results.get(1));

11
spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/ByteBufferEncoderTests.java
View File

@@ -18,13 +18,14 @@ package org.springframework.reactive.codec.encoder;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.*;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.Flux;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import reactor.rx.Streams;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.ByteBufferEncoder;
@@ -48,9 +49,9 @@ public class ByteBufferEncoderTests {
public void decode() throws InterruptedException {
ByteBuffer fooBuffer = Buffer.wrap("foo").byteBuffer();
ByteBuffer barBuffer = Buffer.wrap("bar").byteBuffer();
Stream<ByteBuffer> source = Streams.just(fooBuffer, barBuffer);
List<ByteBuffer> results = Streams.from(encoder.encode(source,
ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null)).toList().get();
Flux<ByteBuffer> source = Flux.just(fooBuffer, barBuffer);
Flux<ByteBuffer> output = encoder.encode(source, ResolvableType.forClassWithGenerics(Publisher.class, ByteBuffer.class), null);
List<ByteBuffer> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals(fooBuffer, results.get(0));
assertEquals(barBuffer, results.get(1));

13
spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/JacksonJsonEncoderTests.java
View File

@@ -16,15 +16,17 @@
package org.springframework.reactive.codec.encoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import reactor.rx.Stream;
import reactor.rx.Streams;
import reactor.Flux;
import org.springframework.core.codec.support.JacksonJsonEncoder;
import org.springframework.http.MediaType;
@@ -45,12 +47,13 @@ public class JacksonJsonEncoderTests {
@Test
public void write() throws InterruptedException {
Stream<Pojo> source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
List<String> results = Streams.from(encoder.encode(source, null, null)).map(chunk -> {
Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("{\"foo\":\"foofoo\",\"bar\":\"barbar\"}", results.get(0));
assertEquals("{\"foo\":\"foofoofoo\",\"bar\":\"barbarbar\"}", results.get(1));

18
spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/Jaxb2EncoderTests.java
View File

@@ -18,13 +18,14 @@ package org.springframework.reactive.codec.encoder;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import reactor.rx.Stream;
import reactor.rx.Streams;
import reactor.Flux;
import org.springframework.core.codec.support.Jaxb2Encoder;
import org.springframework.http.MediaType;
@@ -46,12 +47,13 @@ public class Jaxb2EncoderTests {
@Test
public void encode() throws InterruptedException {
Stream<Pojo> source = Streams.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
List<String> results = Streams.from(encoder.encode(source, null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
Flux<String> output = encoder.encode(source, null, null).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(2, results.size());
assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbar</bar><foo>foofoo</foo></pojo>", results.get(0));
assertEquals("<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?><pojo><bar>barbarbar</bar><foo>foofoofoo</foo></pojo>", results.get(1));

17
spring-web-reactive/src/test/java/org/springframework/reactive/codec/encoder/StringEncoderTests.java
View File

@@ -16,14 +16,17 @@
package org.springframework.reactive.codec.encoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import reactor.rx.Streams;
import reactor.Flux;
import org.springframework.core.ResolvableType;
import org.springframework.core.codec.support.StringEncoder;
@@ -45,12 +48,12 @@ public class StringEncoderTests {
@Test
public void write() throws InterruptedException {
List<String> results = Streams.from(encoder.encode(Streams.just("foo"), null, null))
.map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
}).toList().get();
Flux<String> output = Flux.from(encoder.encode(Flux.just("foo"), null, null)).map(chunk -> {
byte[] b = new byte[chunk.remaining()];
chunk.get(b);
return new String(b, StandardCharsets.UTF_8);
});
List<String> results = StreamSupport.stream(output.toIterable().spliterator(), false).collect(toList());
assertEquals(1, results.size());
assertEquals("foo", results.get(0));
}

6
spring-web-reactive/src/test/java/org/springframework/web/reactive/handler/SimpleUrlHandlerMappingIntegrationTests.java
View File

@@ -22,9 +22,9 @@ import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import reactor.Flux;
import reactor.Mono;
import reactor.io.buffer.Buffer;
import reactor.rx.Stream;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.http.HttpStatus;
@@ -139,7 +139,7 @@ public class SimpleUrlHandlerMappingIntegrationTests extends AbstractHttpHandler
@Override
public Mono<Void> handle(WebServerExchange exchange) {
return exchange.getResponse().setBody(Stream.just(Buffer.wrap("foo").byteBuffer()));
return exchange.getResponse().setBody(Flux.just(Buffer.wrap("foo").byteBuffer()));
}
}
@@ -147,7 +147,7 @@ public class SimpleUrlHandlerMappingIntegrationTests extends AbstractHttpHandler
@Override
public Mono<Void> handle(WebServerExchange exchange) {
return exchange.getResponse().setBody(Stream.just(Buffer.wrap("bar").byteBuffer()));
return exchange.getResponse().setBody(Flux.just(Buffer.wrap("bar").byteBuffer()));
}
}

6
spring-web-reactive/src/test/java/org/springframework/web/reactive/method/annotation/RequestMappingHandlerMappingTests.java
View File

@@ -18,11 +18,12 @@ package org.springframework.web.reactive.method.annotation;
import java.net.URI;
import java.util.List;
import java.util.stream.StreamSupport;
import static java.util.stream.Collectors.toList;
import org.junit.Before;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.rx.Stream;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.http.HttpMethod;
@@ -38,6 +39,7 @@ import org.springframework.web.server.WebServerExchange;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import reactor.Flux;
/**
* @author Sebastien Deleuze
@@ -83,7 +85,7 @@ public class RequestMappingHandlerMappingTests {
private HandlerMethod toHandlerMethod(Publisher<?> handlerPublisher) throws InterruptedException {
assertNotNull(handlerPublisher);
List<?> handlerList = Stream.from(handlerPublisher).toList().get();
List<?> handlerList = StreamSupport.stream(Flux.from(handlerPublisher).toIterable().spliterator(), false).collect(toList());
assertEquals(1, handlerList.size());
return (HandlerMethod) handlerList.get(0);
}