Skip to content

Commit

Permalink
Add RSocketGraphQlClient and GraphQlRSocketHandler
Browse files Browse the repository at this point in the history
See gh-339
  • Loading branch information
rstoyanchev committed Mar 28, 2022
1 parent e71ea05 commit 1462050
Show file tree
Hide file tree
Showing 12 changed files with 702 additions and 18 deletions.
1 change: 1 addition & 0 deletions build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ configure(moduleProjects) {
mavenBom "org.springframework.data:spring-data-bom:2021.2.0-M4"
mavenBom "org.springframework.security:spring-security-bom:5.7.0-M3"
mavenBom "com.querydsl:querydsl-bom:5.0.0"
mavenBom "io.rsocket:rsocket-bom:1.1.1"
mavenBom "org.jetbrains.kotlin:kotlin-bom:1.5.32"
mavenBom "org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.5.2"
mavenBom "org.junit:junit-bom:5.8.1"
Expand Down
8 changes: 7 additions & 1 deletion spring-graphql/build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ dependencies {
compileOnly 'org.springframework:spring-webflux'
compileOnly 'org.springframework:spring-webmvc'
compileOnly 'org.springframework:spring-websocket'
compileOnly 'org.springframework:spring-messaging'
compileOnly 'javax.servlet:javax.servlet-api'
compileOnly 'javax.validation:validation-api'

Expand All @@ -19,6 +20,9 @@ dependencies {
compileOnly 'com.querydsl:querydsl-core'
compileOnly 'org.springframework.data:spring-data-commons'

compileOnly 'io.rsocket:rsocket-core'
compileOnly 'io.rsocket:rsocket-transport-netty'

compileOnly 'com.google.code.findbugs:jsr305'
compileOnly 'org.jetbrains.kotlin:kotlin-stdlib'
compileOnly 'org.jetbrains.kotlinx:kotlinx-coroutines-core'
Expand All @@ -29,10 +33,11 @@ dependencies {
testImplementation 'org.assertj:assertj-core'
testImplementation 'org.mockito:mockito-core'
testImplementation 'io.projectreactor:reactor-test'
testImplementation 'org.springframework:spring-messaging'
testImplementation 'org.springframework:spring-test'
testImplementation 'org.springframework:spring-webflux'
testImplementation 'org.springframework:spring-webmvc'
testImplementation 'org.springframework:spring-websocket'
testImplementation 'org.springframework:spring-test'
testImplementation 'org.springframework.data:spring-data-commons'
testImplementation 'org.springframework.data:spring-data-keyvalue'
testImplementation 'org.springframework.data:spring-data-jpa'
Expand All @@ -49,6 +54,7 @@ dependencies {
testImplementation 'com.querydsl:querydsl-collections'
testImplementation 'javax.servlet:javax.servlet-api'
testImplementation 'com.squareup.okhttp3:mockwebserver:3.14.9'
testImplementation 'io.rsocket:rsocket-transport-local'
testImplementation 'javax.validation:validation-api'
testImplementation 'com.jayway.jsonpath:json-path'
testImplementation 'com.fasterxml.jackson.core:jackson-databind'
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
*/
public abstract class AbstractGraphQlClientBuilder<B extends AbstractGraphQlClientBuilder<B>> implements GraphQlClient.Builder<B> {

private static final boolean jackson2Present = ClassUtils.isPresent(
protected static final boolean jackson2Present = ClassUtils.isPresent(
"com.fasterxml.jackson.databind.ObjectMapper", AbstractGraphQlClientBuilder.class.getClassLoader());


Expand Down Expand Up @@ -111,6 +111,20 @@ protected void setJsonCodecs(Encoder<?> encoder, Decoder<?> decoder) {
this.jsonDecoder = decoder;
}

/**
* Variant of {@link #setJsonCodecs} for setting each codec individually.
*/
protected void setJsonEncoder(Encoder<?> encoder) {
this.jsonEncoder = encoder;
}

/**
* Variant of {@link #setJsonCodecs} for setting each codec individually.
*/
protected void setJsonDecoder(Decoder<?> decoder) {
this.jsonDecoder = decoder;
}

/**
* Return the configured interceptors. For subclasses that look for a
* transport specific interceptor extensions.
Expand All @@ -126,8 +140,8 @@ protected List<GraphQlClientInterceptor> getInterceptors() {
protected GraphQlClient buildGraphQlClient(GraphQlTransport transport) {

if (jackson2Present) {
this.jsonEncoder = (this.jsonEncoder == null ? Jackson2Configurer.encoder() : this.jsonEncoder);
this.jsonDecoder = (this.jsonDecoder == null ? Jackson2Configurer.decoder() : this.jsonDecoder);
this.jsonEncoder = (this.jsonEncoder == null ? DefaultJackson2Codecs.encoder() : this.jsonEncoder);
this.jsonDecoder = (this.jsonDecoder == null ? DefaultJackson2Codecs.decoder() : this.jsonDecoder);
}

return new DefaultGraphQlClient(
Expand Down Expand Up @@ -178,7 +192,7 @@ private Decoder<?> getDecoder() {
}


private static class Jackson2Configurer {
protected static class DefaultJackson2Codecs {

static Encoder<?> encoder() {
return new Jackson2JsonEncoder();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,9 @@
*/
package org.springframework.graphql.client;

import java.util.List;
import java.util.stream.Stream;

import org.springframework.core.ResolvableType;
import org.springframework.core.codec.Decoder;
import org.springframework.core.codec.Encoder;
Expand All @@ -32,8 +35,7 @@
import org.springframework.web.reactive.socket.WebSocketSession;

/**
* Delegate that can be embedded in a class to help with encoding and decoding
* GraphQL over WebSocket messages.
* Helper class for encoding and decoding GraphQL messages.
*
* @author Rossen Stoyanchev
* @since 1.0.0
Expand Down Expand Up @@ -61,22 +63,40 @@ final class CodecDelegate {
this.encoder = findJsonEncoder(configurer);
}

static Encoder<?> findJsonEncoder(CodecConfigurer configurer) {
return findJsonEncoder(configurer.getWriters().stream()
.filter(writer -> writer instanceof EncoderHttpMessageWriter)
.map(writer -> ((EncoderHttpMessageWriter<?>) writer).getEncoder()));
}

static Decoder<?> findJsonDecoder(CodecConfigurer configurer) {
return configurer.getReaders().stream()
.filter((reader) -> reader.canRead(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
.map((reader) -> ((DecoderHttpMessageReader<?>) reader).getDecoder())
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
return findJsonDecoder(configurer.getReaders().stream()
.filter(reader -> reader instanceof DecoderHttpMessageReader)
.map(reader -> ((DecoderHttpMessageReader<?>) reader).getDecoder()));
}

static Encoder<?> findJsonEncoder(CodecConfigurer configurer) {
return configurer.getWriters().stream()
.filter((writer) -> writer.canWrite(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
.map((writer) -> ((EncoderHttpMessageWriter<?>) writer).getEncoder())
static Encoder<?> findJsonEncoder(List<Encoder<?>> encoders) {
return findJsonEncoder(encoders.stream());
}

static Decoder<?> findJsonDecoder(List<Decoder<?>> decoders) {
return findJsonDecoder(decoders.stream());
}

private static Encoder<?> findJsonEncoder(Stream<Encoder<?>> stream) {
return stream
.filter(encoder -> encoder.canEncode(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Encoder"));
}

private static Decoder<?> findJsonDecoder(Stream<Decoder<?>> decoderStream) {
return decoderStream
.filter(decoder -> decoder.canDecode(MESSAGE_TYPE, MediaType.APPLICATION_JSON))
.findFirst()
.orElseThrow(() -> new IllegalArgumentException("No JSON Decoder"));
}


public CodecConfigurer getCodecConfigurer() {
return this.codecConfigurer;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,7 @@ public Builder webClient(Consumer<WebClient.Builder> configurer) {
@Override
public HttpGraphQlClient build() {

// Pass the codecs to the parent for response decoding
this.webClientBuilder.codecs(configurer ->
setJsonCodecs(
CodecDelegate.findJsonEncoder(configurer),
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
/*
* Copyright 2002-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.springframework.graphql.client;

import java.net.URI;
import java.util.function.Consumer;

import io.rsocket.transport.ClientTransport;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.client.WebsocketClientTransport;

import org.springframework.lang.Nullable;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;


/**
* Default {@link RSocketGraphQlClient} implementation that builds the underlying
* {@code RSocketGraphQlTransport} to use.
*
* @author Rossen Stoyanchev
* @since 1.0.0
*/
final class DefaultRSocketGraphQlClient extends AbstractDelegatingGraphQlClient implements RSocketGraphQlClient {

private final RSocketRequester.Builder requesterBuilder;

private final ClientTransport clientTransport;

private final String route;

private final Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer;


DefaultRSocketGraphQlClient(
GraphQlClient graphQlClient, RSocketRequester.Builder requesterBuilder,
ClientTransport clientTransport, String route,
Consumer<AbstractGraphQlClientBuilder<?>> builderInitializer) {

super(graphQlClient);

this.requesterBuilder = requesterBuilder;
this.clientTransport = clientTransport;
this.route = route;
this.builderInitializer = builderInitializer;
}


@Override
public RSocketGraphQlClient.Builder<?> mutate() {
Builder builder = new Builder(this.requesterBuilder);
builder.clientTransport(this.clientTransport);
builder.route(this.route);
this.builderInitializer.accept(builder);
return builder;
}


/**
* Default {@link RSocketGraphQlClient.Builder} implementation.
*/
static final class Builder extends AbstractGraphQlClientBuilder<Builder> implements RSocketGraphQlClient.Builder<Builder> {

private final RSocketRequester.Builder requesterBuilder;

@Nullable
private ClientTransport clientTransport;

private String route;

Builder() {
this(initRSocketRequestBuilder());
}

Builder(RSocketRequester.Builder requesterBuilder) {
Assert.notNull(requesterBuilder, "RSocketRequester.Builder is required");
this.requesterBuilder = requesterBuilder;
this.route = "graphql";
}

private static RSocketRequester.Builder initRSocketRequestBuilder() {
MimeType mimeType = MimeType.valueOf("application/graphql+json");
RSocketRequester.Builder requesterBuilder = RSocketRequester.builder().dataMimeType(mimeType);
if (jackson2Present) {
requesterBuilder.rsocketStrategies(
RSocketStrategies.builder()
.encoder(DefaultJackson2Codecs.encoder())
.decoder(DefaultJackson2Codecs.decoder())
.build());
}
return requesterBuilder;
}

@Override
public Builder tcp(String host, int port) {
this.clientTransport = TcpClientTransport.create(host, port);
return this;
}

@Override
public Builder webSocket(URI uri) {
this.clientTransport = WebsocketClientTransport.create(uri);
return this;
}

@Override
public Builder clientTransport(ClientTransport clientTransport) {
this.clientTransport = clientTransport;
return this;
}

@Override
public Builder dataMimeType(MimeType dataMimeType) {
this.requesterBuilder.dataMimeType(dataMimeType);
return this;
}

@Override
public Builder route(String route) {
Assert.notNull(route, "'route' is required");
this.route = route;
return this;
}

@Override
public Builder rsocketRequester(Consumer<RSocketRequester.Builder> requesterConsumer) {
requesterConsumer.accept(this.requesterBuilder);
return this;
}

@Override
public RSocketGraphQlClient build() {

Assert.state(this.clientTransport != null, "Neither WebSocket nor TCP networking configured");
RSocketRequester requester = this.requesterBuilder.transport(this.clientTransport);
RSocketGraphQlTransport graphQlTransport = new RSocketGraphQlTransport(this.route, requester);

// Pass the codecs to the parent for response decoding
this.requesterBuilder.rsocketStrategies(builder -> {
builder.decoders(decoders -> setJsonDecoder(CodecDelegate.findJsonDecoder(decoders)));
builder.encoders(encoders -> setJsonEncoder(CodecDelegate.findJsonEncoder(encoders)));
});

return new DefaultRSocketGraphQlClient(
super.buildGraphQlClient(graphQlTransport),
this.requesterBuilder, this.clientTransport, this.route, getBuilderInitializer());
}
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,9 @@

/**
* Transport to execute GraphQL requests over HTTP via {@link WebClient}.
* Supports only single-response requests over HTTP POST. For subscription
* requests, see {@link WebSocketGraphQlTransport}.
*
* <p>Supports only single-response requests over HTTP POST. For subscriptions,
* see {@link WebSocketGraphQlTransport} and {@link RSocketGraphQlTransport}.
*
* @author Rossen Stoyanchev
* @since 1.0.0
Expand Down
Loading

0 comments on commit 1462050

Please sign in to comment.