Skip to content

update to latest versions of libs #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
OlegDokuka wants to merge 4 commits into
base: master
Choose a base branch
from
Open

update to latest versions of libs #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d5291af
update to latest versions of libs
OlegDokuka Apr 9, 2018
832fa85
integrity test
OlegDokuka Apr 9, 2018
405e918
fix code style
OlegDokuka Apr 9, 2018
ee4e5f4
fix code style
OlegDokuka Apr 9, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,12 +11,12 @@
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>2.0.0.BUILD-SNAPSHOT</version>
<version>2.0.1.BUILD-SNAPSHOT</version>
</parent>
<properties>
<java.version>1.8</java.version>
<reactor.version>Bismuth-RELEASE</reactor.version>
<rsocket.version>0.9-SNAPSHOT</rsocket.version>
<reactor.version>Californium-BUILD-SNAPSHOT</reactor.version>
<rsocket.version>0.10.0.BUILD-SNAPSHOT</rsocket.version>
<jacoco-maven-plugin.version>0.7.9</jacoco-maven-plugin.version>
</properties>
<modules>
Expand All @@ -34,12 +34,12 @@
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<version>0.9-SNAPSHOT</version>
<version>${rsocket.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
<version>0.9-SNAPSHOT</version>
<version>${rsocket.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
Expand Down
14 changes: 14 additions & 0 deletions spring-cloud-sockets/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,10 @@
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
<dependency>
<groupId>io.projectreactor.ipc</groupId>
<artifactId>reactor-netty</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
Expand All @@ -31,6 +35,16 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
Expand Down
25 changes: 13 additions & 12 deletions ...ud-sockets/src/main/java/org/springframework/cloud/reactive/socket/DispatcherHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.exceptions.ApplicationException;
import io.rsocket.util.PayloadImpl;
import io.rsocket.exceptions.ApplicationErrorException;
import io.rsocket.util.DefaultPayload;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -82,6 +82,7 @@ public void afterPropertiesSet() throws Exception {
String[] beanNames = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(this.applicationContext, Object.class);
for(String beanName : beanNames){
Class<?> beanType = this.applicationContext.getType(beanName);

if(beanType != null){
final Class<?> userType = ClassUtils.getUserClass(beanType);
ReflectionUtils.doWithMethods(userType, method -> {
Expand Down Expand Up @@ -131,10 +132,10 @@ public Mono<Payload> requestResponse(Payload payload) {
Converter converter = converterFor(MimeType.valueOf(metadata.get("MIME_TYPE").textValue()));
Object converted = converter.read(ServiceUtils.toByteArray(payload.getData()), getActualType(handler.getInfo().getParameterType()));
Object result = handler.invoke(handler.getInfo().buildInvocationArguments(converted, null));
Mono monoResult = monoOF(result);
Mono<?> monoResult = monoOF(result);
return monoResult.map(o -> {
byte[] data = converter.write(o);
return new PayloadImpl(data);
return DefaultPayload.create(data);
});

}catch (Exception e){
Expand All @@ -149,20 +150,20 @@ public Flux<Payload> requestStream(Payload payload) {
MethodHandler handler = handlerFor(metadata);
Converter converter = converterFor(MimeType.valueOf(metadata.get("MIME_TYPE").textValue()));
Object converted = converter.read(ServiceUtils.toByteArray(payload.getData()), getActualType(handler.getInfo().getParameterType()));
Flux result = (Flux)handler.invoke(handler.getInfo().buildInvocationArguments(converted, null));
Flux<?> result = (Flux<?>)handler.invoke(handler.getInfo().buildInvocationArguments(converted, null));
return result.map(o ->
new PayloadImpl(converter.write(o))
DefaultPayload.create(converter.write(o))
);

} catch (Exception e){
return Flux.error(new ApplicationException("No path found for " + metadata.get("PATH").asText()));
return Flux.error(new ApplicationErrorException("No path found for " + metadata.get("PATH").asText()));
}
}

private Mono monoOF(Object argument){
private Mono<?> monoOF(Object argument){

if(argument.getClass().isAssignableFrom(Mono.class)){
return (Mono)argument;
return (Mono<?>)argument;
}else{
return Mono.just(argument);
}
Expand All @@ -180,9 +181,9 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
Flux converted = flux.repeat().map(payload -> {
return converter.read(ServiceUtils.toByteArray(payload.getData()), getActualType( handler.getInfo().getParameterType()));
});
Flux result = (Flux)handler.invoke(handler.getInfo().buildInvocationArguments(converted, null));
Flux<?> result = (Flux<?>)handler.invoke(handler.getInfo().buildInvocationArguments(converted, null));
return result.map(o ->
new PayloadImpl(converter.write(o))
DefaultPayload.create(converter.write(o))
);
}catch (Exception e){
return Flux.error(e);
Expand All @@ -207,7 +208,7 @@ private MethodHandler handlerFor(JsonNode metadata){
.getMappingInfo()
.getPath().equals(metadata.get("PATH").asText()); })
.findFirst()
.orElseThrow(() -> { return new ApplicationException("No handler found");} );
.orElseThrow(() -> { return new ApplicationErrorException("No handler found");} );
}

}
7 changes: 5 additions & 2 deletions ...s/src/main/java/org/springframework/cloud/reactive/socket/client/OneWayRemoteHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import java.nio.ByteBuffer;

import io.rsocket.RSocket;
import io.rsocket.util.PayloadImpl;
import io.rsocket.util.DefaultPayload;

import org.springframework.cloud.reactive.socket.ServiceMethodInfo;

Expand All @@ -37,6 +37,9 @@ public OneWayRemoteHandler(RSocket socket, ServiceMethodInfo info) {
@Override
public Object doInvoke(Object argument) {
byte[] payload = payloadConverter.write(argument);
return socket.fireAndForget(new PayloadImpl(ByteBuffer.wrap(payload), getMetadata()));
return socket.fireAndForget(DefaultPayload.create(
ByteBuffer.wrap(payload),
getMetadata()
));
}
}
5 changes: 2 additions & 3 deletions .../main/java/org/springframework/cloud/reactive/socket/client/RequestManyRemoteHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,10 @@

package org.springframework.cloud.reactive.socket.client;


import java.nio.ByteBuffer;

import io.rsocket.RSocket;
import io.rsocket.util.PayloadImpl;
import io.rsocket.util.DefaultPayload;

import org.springframework.cloud.reactive.socket.ServiceMethodInfo;
import org.springframework.cloud.reactive.socket.util.ServiceUtils;
Expand All @@ -37,7 +36,7 @@ public RequestManyRemoteHandler(RSocket socket, ServiceMethodInfo info) {
@Override
public Object doInvoke(Object argument) {
byte[] data = payloadConverter.write(argument);
return socket.requestStream(new PayloadImpl(ByteBuffer.wrap(data), getMetadata()))
return socket.requestStream(DefaultPayload.create(ByteBuffer.wrap(data), getMetadata()))
.map(payload -> payloadConverter.read(ServiceUtils.toByteArray(payload.getData()), ServiceUtils.getActualType(info.getParameterType())));
}
}
7 changes: 5 additions & 2 deletions ...c/main/java/org/springframework/cloud/reactive/socket/client/RequestOneRemoteHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import java.nio.ByteBuffer;

import io.rsocket.RSocket;
import io.rsocket.util.PayloadImpl;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Mono;

import org.springframework.cloud.reactive.socket.ServiceMethodInfo;
Expand All @@ -38,7 +38,10 @@ public RequestOneRemoteHandler(RSocket socket, ServiceMethodInfo info) {
@Override
public Object doInvoke(Object argument) {
byte[] data = payloadConverter.write(argument);
Mono monoResult = socket.requestResponse(new PayloadImpl(ByteBuffer.wrap(data), getMetadata()))
Mono monoResult = socket.requestResponse(DefaultPayload.create(
ByteBuffer.wrap(data),
getMetadata()
))
.map(payload -> payloadConverter.read(ServiceUtils.toByteArray(payload.getData()), ServiceUtils.getActualType(info.getReturnType())));
if(Mono.class.isAssignableFrom(info.getReturnType().resolve())){
return monoResult;
Expand Down
16 changes: 8 additions & 8 deletions ...sockets/src/test/java/org/springframework/cloud/reactive/socket/DispatchHandlerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;

import io.rsocket.util.PayloadImpl;
import io.rsocket.util.DefaultPayload;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
Expand Down Expand Up @@ -65,7 +65,7 @@ public void setup() throws Exception{
@Test
public void oneWayHandler() throws Exception{
User user = new User("Mary", "blue");
Mono<Void> result = this.handler.fireAndForget(new PayloadImpl(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/oneway")));
Mono<Void> result = this.handler.fireAndForget(DefaultPayload.create(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/oneway")));
User output = (User) resultsQueue.poll();
assertThat(output).isEqualTo(user);
}
Expand All @@ -74,23 +74,23 @@ public void oneWayHandler() throws Exception{
public void oneWaySerializable() throws Exception {
User user = new User("Mary", "blue");
SerializableConverter serializableConverter = new SerializableConverter();
Mono<Void> result = this.handler.fireAndForget(new PayloadImpl(serializableConverter.write(user), getMetadataBytes(MimeType.valueOf("application/java-serialized-object") ,"/onewaybinary")));
Mono<Void> result = this.handler.fireAndForget(DefaultPayload.create(serializableConverter.write(user), getMetadataBytes(MimeType.valueOf("application/java-serialized-object") ,"/onewaybinary")));
User output = (User) resultsQueue.poll();
assertThat(output).isEqualTo(user);
}

@Test
public void oneWayWrongMimeType() throws Exception {
User user = new User("Mary", "blue");
Mono<Void> result = this.handler.fireAndForget(new PayloadImpl(converter.write(user), getMetadataBytes(MimeType.valueOf("application/binary") ,"/oneway")));
Mono<Void> result = this.handler.fireAndForget(DefaultPayload.create(converter.write(user), getMetadataBytes(MimeType.valueOf("application/binary") ,"/oneway")));
result.doOnError(throwable -> resultsQueue.offer(throwable)).subscribe();
assertThat(resultsQueue.poll()).isInstanceOf(Throwable.class);
}

@Test
public void requestOneHandler() throws Exception {
User user = new User("Mary", "red");
Mono<io.rsocket.Payload> invocationResult = this.handler.requestResponse(new PayloadImpl(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/redblue")));
Mono<io.rsocket.Payload> invocationResult = this.handler.requestResponse(DefaultPayload.create(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/redblue")));
User result = invocationResult.map(payload -> {
return (User)converter.read(payload.getDataUtf8().getBytes(), User.class);
}).block();
Expand All @@ -102,7 +102,7 @@ public void requestOneHandler() throws Exception {
@Test
public void requestOneWrongPath() throws Exception {
User user = new User("Mary", "red");
Mono<io.rsocket.Payload> invocationResult = this.handler.requestResponse(new PayloadImpl(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/notfound")));
Mono<io.rsocket.Payload> invocationResult = this.handler.requestResponse(DefaultPayload.create(converter.write(user), getMetadataBytes(MimeType.valueOf("application/json") ,"/notfound")));
invocationResult.doOnError(throwable -> { resultsQueue.offer(throwable);}).subscribe();
assertThat(resultsQueue.poll()).isInstanceOf(Throwable.class);
}
Expand All @@ -111,7 +111,7 @@ public void requestOneWrongPath() throws Exception {
@Test
public void requestMany() throws Exception {
Integer count = 10;
Flux<io.rsocket.Payload> invocationResult = this.handler.requestStream(new PayloadImpl(converter.write(count), getMetadataBytes(MimeType.valueOf("application/json") ,"/requestMany")));
Flux<io.rsocket.Payload> invocationResult = this.handler.requestStream(DefaultPayload.create(converter.write(count), getMetadataBytes(MimeType.valueOf("application/json") ,"/requestMany")));
List<Integer> results = invocationResult.map(payload -> (Integer)converter.read(payload.getDataUtf8().getBytes(), Integer.class)
).collectList().block();

Expand All @@ -122,7 +122,7 @@ public void requestMany() throws Exception {
public void requestStream() throws Exception {
Flux<Integer> from = Flux.range(0,10);
Flux<io.rsocket.Payload> payloadFlux = from.map(integer -> {
return new PayloadImpl(converter.write(integer), getMetadataBytes(MimeType.valueOf("application/json") ,"/requestStream"));
return DefaultPayload.create(converter.write(integer), getMetadataBytes(MimeType.valueOf("application/json") ,"/requestStream"));
});

Flux<io.rsocket.Payload> invocationResult = this.handler.requestChannel(payloadFlux);
Expand Down
4 changes: 2 additions & 2 deletions ...ts/src/test/java/org/springframework/cloud/reactive/socket/ReactiveSocketClientTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import com.fasterxml.jackson.databind.JsonNode;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.util.PayloadImpl;
import io.rsocket.util.DefaultPayload;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
Expand Down Expand Up @@ -81,7 +81,7 @@ public void requestOneClientTests() throws Exception {
ArgumentCaptor<Payload> captor = ArgumentCaptor.forClass(Payload.class);
User user = new User("Alice","blue");
byte[] converted = converter.write(user);
when(mockSocket.requestResponse(Mockito.any(Payload.class))).thenReturn(Mono.just(new PayloadImpl(converted)));
when(mockSocket.requestResponse(Mockito.any(Payload.class))).thenReturn(Mono.just(DefaultPayload.create(converted)));
client.create(user);
verify(mockSocket, times(1)).requestResponse(captor.capture());
Payload payload = captor.getValue();
Expand Down
64 changes: 64 additions & 0 deletions .../test/java/org/springframework/cloud/reactive/socket/ReactiveSocketsApplicationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,74 @@

package org.springframework.cloud.reactive.socket;

import java.time.Duration;

import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import org.junit.Test;
import org.junit.runner.RunWith;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.reactive.socket.annotation.EnableReactiveSockets;
import org.springframework.cloud.reactive.socket.annotation.Payload;
import org.springframework.cloud.reactive.socket.annotation.RequestManyMapping;
import org.springframework.cloud.reactive.socket.client.ReactiveSocketClient;
import org.springframework.test.context.junit4.SpringRunner;

/**
* @author Vinicius Carvalho
*/
@RunWith(SpringRunner.class)
@SpringBootTest
public class ReactiveSocketsApplicationTests {

@Test
public void integrityTest() {
RSocket rSocket = RSocketFactory.connect()
.transport(TcpClientTransport.create("localhost", 5000))
.start()
.block();

ReactiveSocketClient client = new ReactiveSocketClient(rSocket);
TestClient clientProxy = client.create(TestClient.class);

StepVerifier.create(Flux.merge(clientProxy.receiveStream1("a"),
clientProxy.receiveStream2("b")))
.expectSubscription()
.expectNext("a", "b")
.expectNextCount(5)
.thenCancel()
.verify();
}

@SpringBootApplication
@EnableReactiveSockets
public static class TestApplication {

@RequestManyMapping(value = "/stream1", mimeType = "application/json")
public Flux<String> stream1(@Payload String a) {
return Flux.just(a)
.mergeWith(Flux.interval(Duration.ofMillis(100))
.map(i -> "1. Stream Message : [" + i + "]"));
}

@RequestManyMapping(value = "/stream2", mimeType = "application/json")
public Flux<String> stream2(@Payload String b) {
return Flux.just(b)
.mergeWith(Flux.interval(Duration.ofMillis(500))
.map(i -> "2. Stream Message : [" + i + "]"));
}
}

public interface TestClient {
@RequestManyMapping(value = "/stream1", mimeType = "application/json")
Flux<String> receiveStream1(String a);

@RequestManyMapping(value = "/stream1", mimeType = "application/json")
Flux<String> receiveStream2(String b);
}
}