SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Migrates to using rsocket-routing-broker

Browse Source
This commit is contained in:
spencergibb
2020-08-24 15:49:31 -04:00
parent 311d67c38b
commit db28b7efdc
1 changed files with 27 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RoutingBrokerTests.java
View File

@@ -16,28 +16,10 @@
package org.springframework.cloud.function.rsocket; package org.springframework.cloud.function.rsocket;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
import java.util.function.Consumer; import java.util.function.Consumer;
import java.util.function.Function; import java.util.function.Function;
import io.netty.buffer.ByteBuf; import io.rsocket.routing.client.spring.SpringRouting;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.frame.decoder.PayloadDecoder;
import io.rsocket.metadata.CompositeMetadataCodec;
import io.rsocket.routing.broker.spring.MimeTypes;
import io.rsocket.routing.common.Id;
import io.rsocket.routing.common.Tags;
import io.rsocket.routing.common.WellKnownKey;
import io.rsocket.routing.frames.AddressFlyweight;
import io.rsocket.routing.frames.RouteSetupFlyweight;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
@@ -49,12 +31,9 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.util.SocketUtils; import org.springframework.util.SocketUtils;
import static io.netty.buffer.Unpooled.EMPTY_BUFFER;
import static io.rsocket.routing.broker.spring.MimeTypes.COMPOSITE_MIME_TYPE;
/** /**
* *
@@ -68,12 +47,13 @@ public class RoutingBrokerTests {
ConfigurableApplicationContext functionContext = null; ConfigurableApplicationContext functionContext = null;
ConfigurableApplicationContext brokerContext = null; ConfigurableApplicationContext brokerContext = null;
ConfigurableApplicationContext clientContext = null;
try { try {
int routingBrokerProxyPort = SocketUtils.findAvailableTcpPort(); int routingBrokerProxyPort = SocketUtils.findAvailableTcpPort();
int routingBrokerClusterPort = SocketUtils.findAvailableTcpPort(); int routingBrokerClusterPort = SocketUtils.findAvailableTcpPort();
// start broker // start broker
brokerContext = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( brokerContext = new SpringApplicationBuilder(SimpleConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.io.rsocket.routing.broker=TRACE", "--logging.level.io.rsocket.routing.broker=TRACE",
"--spring.cloud.function.rsocket.enabled=false", "--spring.cloud.function.rsocket.enabled=false",
"--io.rsocket.routing.client.enabled=false", "--io.rsocket.routing.client.enabled=false",
@@ -81,37 +61,35 @@ public class RoutingBrokerTests {
"--io.rsocket.routing.broker.tcp.port=" + routingBrokerProxyPort, "--io.rsocket.routing.broker.tcp.port=" + routingBrokerProxyPort,
"--io.rsocket.routing.broker.cluster.port=" + routingBrokerClusterPort); "--io.rsocket.routing.broker.cluster.port=" + routingBrokerClusterPort);
// start function connecting to broker, service-name=toupper // start function connecting to broker, service-name=samplefn
functionContext = new SpringApplicationBuilder(SampleFunctionConfiguration.class) functionContext = new SpringApplicationBuilder(SampleFunctionConfiguration.class)
.web(WebApplicationType.NONE).run( .web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG", "--logging.level.org.springframework.cloud.function=DEBUG",
"--io.rsocket.routing.client.enabled=true", "--io.rsocket.routing.client.enabled=true",
"--io.rsocket.routing.client.service-name=toupper", "--io.rsocket.routing.client.service-name=samplefn",
"--io.rsocket.routing.client.brokers[0].host=localhost", "--io.rsocket.routing.client.brokers[0].host=localhost",
"--io.rsocket.routing.client.brokers[0].port=" + routingBrokerProxyPort, "--io.rsocket.routing.client.brokers[0].port=" + routingBrokerProxyPort,
"--io.rsocket.routing.broker.enabled=false", "--io.rsocket.routing.broker.enabled=false",
"--spring.cloud.function.definition=uppercase"); "--spring.cloud.function.definition=uppercase");
// setup metadata to identify the this test connecting to broker. // start testclient connecting to broker, for RSocketRequester
RSocketStrategies strategies = functionContext.getBean(RSocketStrategies.class); clientContext = new SpringApplicationBuilder(SimpleConfiguration.class)
Id testerId = Id.random(); .web(WebApplicationType.NONE).run(
ByteBuf routeSetup = encodeRouteSetup(strategies, testerId, "tester"); "--logging.level.io.rsocket.routing.client=TRACE",
Payload setupPayload = DefaultPayload.create(EMPTY_BUFFER, routeSetup); "--spring.cloud.function.rsocket.enabled=false",
"--io.rsocket.routing.client.enabled=true",
"--io.rsocket.routing.client.service-name=testclient",
"--io.rsocket.routing.client.brokers[0].host=localhost",
"--io.rsocket.routing.client.brokers[0].port=" + routingBrokerProxyPort,
"--io.rsocket.routing.broker.enabled=false");
// connect to broker RSocketRequester requester = clientContext.getBean(RSocketRequester.class);
RSocket socket = RSocketConnector.create().payloadDecoder(PayloadDecoder.ZERO_COPY) SpringRouting routing = clientContext.getBean(SpringRouting.class);
.metadataMimeType(COMPOSITE_MIME_TYPE.toString()) Mono<String> result = requester.route("toupper") // used to find a messagemapping, so unused here
.setupPayload(setupPayload) // auto creates metadata
.connect(TcpClientTransport.create(routingBrokerProxyPort)) .metadata(routing.address("samplefn"))
.block(); .data("\"hello\"")
.retrieveMono(String.class);
// setup data for request to toupper service
ByteBuffer data = StandardCharsets.UTF_8.encode(CharBuffer.wrap("\"hello\""));
// setup metadata for request to toupper service
ByteBuf routingMetadata = encodeAddress(strategies, testerId, "toupper");
Payload payload = DefaultPayload.create(data, routingMetadata.nioBuffer());
// call toupper service
Mono<String> result = socket.requestResponse(payload).map(Payload::getDataUtf8);
StepVerifier StepVerifier
.create(result) .create(result)
@@ -125,44 +103,15 @@ public class RoutingBrokerTests {
if (brokerContext != null) { if (brokerContext != null) {
brokerContext.close(); brokerContext.close();
} }
if (clientContext != null) {
clientContext.close();
}
} }
} }
static ByteBuf encodeAddress(RSocketStrategies strategies, Id originRouteId, String serviceName) {
Tags tags = Tags.builder().with(WellKnownKey.SERVICE_NAME, serviceName)
.buildTags();
ByteBuf address = AddressFlyweight
.encode(ByteBufAllocator.DEFAULT, originRouteId, Tags.empty(), tags);
CompositeByteBuf composite = encodeComposite(address, MimeTypes.ROUTING_FRAME_MIME_TYPE
.toString());
return composite;
}
private static ByteBuf encodeRouteSetup(RSocketStrategies strategies, Id routeId, String serviceName) {
Tags tags = Tags.builder()
.with("current-time", String.valueOf(System.currentTimeMillis()))
.with(WellKnownKey.TIME_ZONE, System.currentTimeMillis() + "")
.buildTags();
ByteBuf routeSetup = RouteSetupFlyweight
.encode(ByteBufAllocator.DEFAULT, routeId, serviceName, tags);
CompositeByteBuf composite = encodeComposite(routeSetup, MimeTypes.ROUTING_FRAME_MIME_TYPE
.toString());
return composite;
}
private static CompositeByteBuf encodeComposite(ByteBuf byteBuf, String mimeType) {
CompositeByteBuf composite = ByteBufAllocator.DEFAULT.compositeBuffer();
CompositeMetadataCodec
.encodeAndAddMetadata(composite, ByteBufAllocator.DEFAULT,
mimeType, byteBuf);
return composite;
}
@EnableAutoConfiguration @EnableAutoConfiguration
@Configuration @Configuration
public static class RoutingBrokerConfiguration { public static class SimpleConfiguration {
} }