Add interception for RSocket handler

See gh-339

spring-graphql/src/main/java/org/springframework/graphql/support/DefaultExecutionGraphQlResponse.java

@@ -23,6 +23,7 @@ import java.util.stream.Collectors;
 import graphql.ErrorClassification;
 import graphql.ExecutionInput;
 import graphql.ExecutionResult;
+import graphql.ExecutionResultImpl;
 import graphql.GraphQLError;
 import graphql.language.SourceLocation;
 
@@ -153,4 +154,65 @@ public class DefaultExecutionGraphQlResponse extends AbstractGraphQlResponse imp
 
 	}
 
+
+	/**
+	 * Builder to transform the response's {@link ExecutionResult}.
+	 */
+	public static abstract class Builder<B extends Builder<B, R>, R extends ExecutionGraphQlResponse> {
+
+		private final R original;
+
+		private final ExecutionResultImpl.Builder executionResultBuilder;
+
+		protected Builder(R original) {
+			this.original = original;
+			this.executionResultBuilder = ExecutionResultImpl.newExecutionResult().from(original.getExecutionResult());
+		}
+
+		/**
+		 * Set the {@link ExecutionResult#getData() data} of the GraphQL execution result.
+		 * @param data the execution result data
+		 * @return the current builder
+		 */
+		public Builder<B, R> data(Object data) {
+			this.executionResultBuilder.data(data);
+			return this;
+		}
+
+		/**
+		 * Set the {@link ExecutionResult#getErrors() errors} of the GraphQL execution
+		 * result.
+		 * @param errors the execution result errors
+		 * @return the current builder
+		 */
+		public Builder<B, R> errors(@Nullable List<GraphQLError> errors) {
+			this.executionResultBuilder.errors(errors);
+			return this;
+		}
+
+		/**
+		 * Set the {@link ExecutionResult#getExtensions() extensions} of the GraphQL
+		 * execution result.
+		 * @param extensions the execution result extensions
+		 * @return the current builder
+		 */
+		public Builder<B, R> extensions(@Nullable Map<Object, Object> extensions) {
+			this.executionResultBuilder.extensions(extensions);
+			return this;
+		}
+
+		/**
+		 * Build the response with the transformed {@code ExecutionResult}.
+		 */
+		public R build() {
+			return build(this.original, this.executionResultBuilder.build());
+		}
+
+		/**
+		 * Subclasses to create the specific response instance.
+		 */
+		protected abstract R build(R original, ExecutionResult newResult);
+
+	}
+
 }

spring-graphql/src/main/java/org/springframework/graphql/web/DefaultWebGraphQlHandlerBuilder.java

@@ -25,6 +25,7 @@ import reactor.core.publisher.Mono;
 import org.springframework.graphql.ExecutionGraphQlService;
 import org.springframework.graphql.execution.ReactorContextManager;
 import org.springframework.graphql.execution.ThreadLocalAccessor;
+import org.springframework.graphql.web.WebGraphQlHandlerInterceptor.Chain;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -88,12 +89,11 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder {
 	@Override
 	public WebGraphQlHandler build() {
 
-		WebGraphQlHandlerInterceptor.Chain endOfChain =
-				request -> this.service.execute(request).map(WebGraphQlResponse::new);
+		Chain endOfChain = request -> this.service.execute(request).map(WebGraphQlResponse::new);
 
-		WebGraphQlHandlerInterceptor.Chain chain = this.interceptors.stream()
+		Chain chain = this.interceptors.stream()
 				.reduce(WebGraphQlHandlerInterceptor::andThen)
-				.map(interceptor -> (WebGraphQlHandlerInterceptor.Chain) (request) -> interceptor.intercept(request, endOfChain))
+				.map(interceptor -> (Chain) (request) -> interceptor.intercept(request, endOfChain))
 				.orElse(endOfChain);
 
 		return new WebGraphQlHandler() {
@@ -112,7 +112,8 @@ class DefaultWebGraphQlHandlerBuilder implements WebGraphQlHandler.Builder {
 
 			@Override
 			public WebSocketGraphQlHandlerInterceptor webSocketInterceptor() {
-				return (webSocketInterceptor != null ? webSocketInterceptor : new WebSocketGraphQlHandlerInterceptor() {});
+				return (webSocketInterceptor != null ?
+						webSocketInterceptor : new WebSocketGraphQlHandlerInterceptor() {});
 			}
 
 		};

spring-graphql/src/main/java/org/springframework/graphql/web/GraphQlRSocketHandler.java

@@ -17,6 +17,7 @@
 package org.springframework.graphql.web;
 
 
+import java.util.List;
 import java.util.Map;
 
 import graphql.ExecutionResult;
@@ -25,38 +26,39 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.graphql.ExecutionGraphQlRequest;
 import org.springframework.graphql.ExecutionGraphQlResponse;
 import org.springframework.graphql.ExecutionGraphQlService;
-import org.springframework.graphql.support.DefaultExecutionGraphQlRequest;
+import org.springframework.graphql.web.RSocketGraphQlHandlerInterceptor.Chain;
+import org.springframework.util.AlternativeJdkIdGenerator;
+import org.springframework.util.IdGenerator;
 
 
 /**
  * Handler for GraphQL over RSocket requests.
  *
- * <p>This class can be extended from an {@code @Controller} that overrides
- * {@link #handle(Map)} and {@link #handleSubscription(Map)} in order to add
+ * <p>This class can be extended or wrapped from an {@code @Controller} in order
+ * to re-declare {@link #handle(Map)} and {@link #handleSubscription(Map)} with
  * {@link org.springframework.messaging.handler.annotation.MessageMapping @MessageMapping}
- * annotations with the route.
+ * annotations including the GraphQL endpoint route.
  *
  * <pre style="class">
  * &#064;Controller
- * private static class GraphQlRSocketController extends GraphQlRSocketHandler {
+ * private static class GraphQlRSocketController {
  *
- *    GraphQlRSocketController(ExecutionGraphQlService graphQlService) {
- *        super(graphQlService);
+ *    private final GraphQlRSocketHandler handler;
+ *
+ *    GraphQlRSocketController(GraphQlRSocketHandler handler) {
+ *        this.handler = handler;
  *    }
  *
- *    &#064;Override
  *    &#064;MessageMapping("graphql")
  *    public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
- *        return super.handle(payload);
+ *        return this.handler.handle(payload);
  *    }
  *
- *    &#064;Override
  *    &#064;MessageMapping("graphql")
  *    public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
- *        return super.handleSubscription(payload);
+ *        return this.handler.handleSubscription(payload);
  *    }
  * }
  * </pre>
@@ -66,11 +68,25 @@ import org.springframework.graphql.support.DefaultExecutionGraphQlRequest;
  */
 public class GraphQlRSocketHandler {
 
-	private final ExecutionGraphQlService service;
+	private final Chain executionChain;
+
+	private final IdGenerator idGenerator = new AlternativeJdkIdGenerator();
 
 
-	public GraphQlRSocketHandler(ExecutionGraphQlService service) {
-		this.service = service;
+	/**
+	 * Create a new instance that handles requests through a chain of interceptors
+	 * followed by the given {@link ExecutionGraphQlService}.
+	 */
+	public GraphQlRSocketHandler(
+			ExecutionGraphQlService service, List<RSocketGraphQlHandlerInterceptor> interceptors) {
+
+		Chain endOfChain = request -> service.execute(request).map(RSocketGraphQlResponse::new);
+
+		this.executionChain = (interceptors.isEmpty() ? endOfChain :
+				interceptors.stream()
+						.reduce(RSocketGraphQlHandlerInterceptor::andThen)
+						.map(interceptor -> (Chain) request -> interceptor.intercept(request, endOfChain))
+						.orElse(endOfChain));
 	}
 
 
@@ -78,14 +94,14 @@ public class GraphQlRSocketHandler {
 	 * Handle a {@code Request-Response} interaction. For queries and mutations.
 	 */
 	public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
-		return this.service.execute(initRequest(payload)).map(ExecutionGraphQlResponse::toMap);
+		return handleInternal(payload).map(ExecutionGraphQlResponse::toMap);
 	}
 
 	/**
 	 * Handle a {@code Request-Stream} interaction. For subscriptions.
 	 */
 	public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
-		return this.service.execute(initRequest(payload))
+		return handleInternal(payload)
 				.flatMapMany(response -> {
 					if (response.getData() instanceof Publisher) {
 						Publisher<ExecutionResult> publisher = response.getData();
@@ -100,12 +116,9 @@ public class GraphQlRSocketHandler {
 				});
 	}
 
-	@SuppressWarnings("unchecked")
-	private ExecutionGraphQlRequest initRequest(Map<String, Object> payload) {
-		String query = (String) payload.get("query");
-		String operationName = (String) payload.get("operationName");
-		Map<String, Object> variables = (Map<String, Object>) payload.get("variables");
-		return new DefaultExecutionGraphQlRequest(query, operationName, variables, "1", null);
+	private Mono<RSocketGraphQlResponse> handleInternal(Map<String, Object> payload) {
+		String requestId = this.idGenerator.generateId().toString();
+		return this.executionChain.next(new RSocketGraphQlRequest(payload, requestId, null));
 	}
 
 }

spring-graphql/src/main/java/org/springframework/graphql/web/RSocketGraphQlHandlerInterceptor.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2020-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.web;
+
+import graphql.ExecutionInput;
+import graphql.ExecutionResult;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.graphql.ExecutionGraphQlService;
+
+
+/**
+ * Interceptor for server handling of GraphQL over RSocket requests,
+ * allowing customization of the {@link ExecutionInput} and
+ * the {@link ExecutionResult}.
+ *
+ * <p>Interceptors are typically declared as beans in Spring configuration and
+ * ordered as defined in {@link ObjectProvider#orderedStream()}.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+public interface RSocketGraphQlHandlerInterceptor {
+
+	/**
+	 * Intercept a request and delegate to the rest of the chain including other
+	 * interceptors and a {@link ExecutionGraphQlService}.
+	 * @param request the request to execute
+	 * @param chain the rest of the chain to execute the request
+	 * @return a {@link Mono} with the response
+	 */
+	Mono<RSocketGraphQlResponse> intercept(RSocketGraphQlRequest request, Chain chain);
+
+	/**
+	 * Return a new {@link RSocketGraphQlHandlerInterceptor} that invokes the current
+	 * interceptor first and then the one that is passed in.
+	 * @param nextInterceptor the interceptor to delegate to after the current
+	 * @return a new interceptor that chains the two
+	 */
+	default RSocketGraphQlHandlerInterceptor andThen(RSocketGraphQlHandlerInterceptor nextInterceptor) {
+		return (request, chain) -> intercept(request, nextRequest -> nextInterceptor.intercept(nextRequest, chain));
+	}
+
+
+	/**
+	 * Contract for delegation to the rest of the chain.
+	 */
+	interface Chain {
+
+		/**
+		 * Delegate to the rest of the chain to execute the request.
+		 * @param request the request to execute
+		 * the {@link ExecutionInput} for {@link graphql.GraphQL}.
+		 * @return {@code Mono} with the response
+		 */
+		Mono<RSocketGraphQlResponse> next(RSocketGraphQlRequest request);
+
+	}
+
+}

spring-graphql/src/main/java/org/springframework/graphql/web/RSocketGraphQlRequest.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2020-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.web;
+
+import java.util.Locale;
+import java.util.Map;
+
+import io.rsocket.exceptions.RejectedException;
+
+import org.springframework.graphql.ExecutionGraphQlRequest;
+import org.springframework.graphql.support.DefaultExecutionGraphQlRequest;
+import org.springframework.lang.Nullable;
+import org.springframework.util.StringUtils;
+
+/**
+ * {@link org.springframework.graphql.GraphQlRequest} implementation for server
+ * handling over RSocket.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+public class RSocketGraphQlRequest extends DefaultExecutionGraphQlRequest implements ExecutionGraphQlRequest {
+
+
+	/**
+	 * Create an instance.
+	 * @param body the deserialized content of the GraphQL request
+	 * @param id an identifier for the GraphQL request
+	 * @param locale the locale from the HTTP request, if any
+	 */
+	public RSocketGraphQlRequest(Map<String, Object> body, String id, @Nullable Locale locale) {
+		super(getKey("query", body), getKey("operationName", body), getKey("variables", body), id, locale);
+	}
+
+	@SuppressWarnings("unchecked")
+	private static <T> T getKey(String key, Map<String, Object> body) {
+		if (key.equals("query") && !StringUtils.hasText((String) body.get(key))) {
+			throw new RejectedException("No \"query\" in the request document");
+		}
+		return (T) body.get(key);
+	}
+
+}

spring-graphql/src/main/java/org/springframework/graphql/web/RSocketGraphQlResponse.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2020-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.web;
+
+import java.util.function.Consumer;
+
+import graphql.ExecutionResult;
+
+import org.springframework.graphql.ExecutionGraphQlResponse;
+import org.springframework.graphql.support.DefaultExecutionGraphQlResponse;
+
+
+/**
+ * {@link org.springframework.graphql.GraphQlResponse} implementation for server
+ * handling over RSocket.
+ *
+ * @author Rossen Stoyanchev
+ * @since 1.0.0
+ */
+public class RSocketGraphQlResponse extends DefaultExecutionGraphQlResponse {
+
+	/**
+	 * Create an instance that wraps the given {@link ExecutionGraphQlResponse}.
+	 * @param response the response to wrap
+	 */
+	public RSocketGraphQlResponse(ExecutionGraphQlResponse response) {
+		super(response);
+	}
+
+	private RSocketGraphQlResponse(RSocketGraphQlResponse original, ExecutionResult executionResult) {
+		super(original.getExecutionInput(), executionResult);
+	}
+
+
+	/**
+	 * Transform the underlying {@link ExecutionResult} through a {@link Builder}
+	 * and return a new instance with the modified values.
+	 * @param consumer callback to transform the result
+	 * @return the new response instance with the mutated {@code ExecutionResult}
+	 */
+	public RSocketGraphQlResponse transform(Consumer<Builder> consumer) {
+		Builder builder = new Builder(this);
+		consumer.accept(builder);
+		return builder.build();
+	}
+
+
+	/**
+	 * Builder to transform a {@link RSocketGraphQlResponse}.
+	 */
+	public static final class Builder
+			extends DefaultExecutionGraphQlResponse.Builder<Builder, RSocketGraphQlResponse> {
+
+		private Builder(RSocketGraphQlResponse original) {
+			super(original);
+		}
+
+		@Override
+		protected RSocketGraphQlResponse build(RSocketGraphQlResponse original, ExecutionResult newResult) {
+			return new RSocketGraphQlResponse(original, newResult);
+		}
+
+	}
+
+}

spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlHandlerInterceptor.java

@@ -22,11 +22,11 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.graphql.ExecutionGraphQlService;
-import org.springframework.util.Assert;
+
 
 /**
  * Interceptor for server handling of GraphQL over HTTP or WebSocket requests,
- * providing access info about the underlying HTTP request or WebSocket
+ * providing access to info about the underlying HTTP request or WebSocket
  * handshake, and allowing customization of the {@link ExecutionInput} and
  * the {@link ExecutionResult}.
  *
@@ -53,15 +53,11 @@ public interface WebGraphQlHandlerInterceptor {
 	/**
 	 * Return a new {@link WebGraphQlHandlerInterceptor} that invokes the current
 	 * interceptor first and then the one that is passed in.
-	 * @param interceptor the interceptor to delegate to after "this"
-	 * @return the new {@code WebGraphQlHandlerInterceptor}
+	 * @param nextInterceptor the interceptor to delegate to after the current
+	 * @return a new interceptor that chains the two
 	 */
-	default WebGraphQlHandlerInterceptor andThen(WebGraphQlHandlerInterceptor interceptor) {
-		Assert.notNull(interceptor, "WebGraphQlHandlerInterceptor is required");
-		return (request, chain) -> {
-			Chain nextChain = nextRequest -> interceptor.intercept(nextRequest, chain);
-			return intercept(request, nextChain);
-		};
+	default WebGraphQlHandlerInterceptor andThen(WebGraphQlHandlerInterceptor nextInterceptor) {
+		return (request, chain) -> intercept(request, nextRequest -> nextInterceptor.intercept(nextRequest, chain));
 	}
 
 
@@ -79,4 +75,5 @@ public interface WebGraphQlHandlerInterceptor {
 		Mono<WebGraphQlResponse> next(WebGraphQlRequest request);
 
 	}
+
 }

spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlRequest.java

@@ -30,6 +30,7 @@ import org.springframework.web.server.ServerWebInputException;
 import org.springframework.web.util.UriComponents;
 import org.springframework.web.util.UriComponentsBuilder;
 
+
 /**
  * {@link org.springframework.graphql.GraphQlRequest} implementation for server
  * handling over HTTP or WebSocket. Provides access to the URL and headers of
@@ -51,8 +52,7 @@ public class WebGraphQlRequest extends DefaultExecutionGraphQlRequest implements
 	 * @param uri the URL for the HTTP request or WebSocket handshake
 	 * @param headers the HTTP request headers
 	 * @param body the deserialized content of the GraphQL request
-	 * @param id an identifier for the GraphQL request, e.g. a subscription id for
-	 * correlating request and response messages, or it could be an id associated with the
+	 * @param id an identifier for the GraphQL request
 	 * @param locale the locale from the HTTP request, if any
 	 */
 	public WebGraphQlRequest(

spring-graphql/src/main/java/org/springframework/graphql/web/WebGraphQlResponse.java

@@ -16,18 +16,14 @@
 
 package org.springframework.graphql.web;
 
-import java.util.List;
-import java.util.Map;
 import java.util.function.Consumer;
 
 import graphql.ExecutionResult;
-import graphql.ExecutionResultImpl;
-import graphql.GraphQLError;
 
 import org.springframework.graphql.ExecutionGraphQlResponse;
 import org.springframework.graphql.support.DefaultExecutionGraphQlResponse;
 import org.springframework.http.HttpHeaders;
-import org.springframework.lang.Nullable;
+
 
 /**
  * {@link org.springframework.graphql.GraphQlResponse} implementation for server
@@ -83,51 +79,15 @@ public class WebGraphQlResponse extends DefaultExecutionGraphQlResponse {
 	/**
 	 * Builder to transform a {@link WebGraphQlResponse}.
 	 */
-	public static final class Builder {
-
-		private final WebGraphQlResponse original;
-
-		private final ExecutionResultImpl.Builder executionResultBuilder;
+	public static final class Builder extends DefaultExecutionGraphQlResponse.Builder<Builder, WebGraphQlResponse> {
 
 		private Builder(WebGraphQlResponse original) {
-			this.original = original;
-			this.executionResultBuilder = ExecutionResultImpl.newExecutionResult().from(original.getExecutionResult());
+			super(original);
 		}
 
-		/**
-		 * Set the {@link ExecutionResult#getData() data} of the GraphQL execution result.
-		 * @param data the execution result data
-		 * @return the current builder
-		 */
-		public Builder data(Object data) {
-			this.executionResultBuilder.data(data);
-			return this;
-		}
-
-		/**
-		 * Set the {@link ExecutionResult#getErrors() errors} of the GraphQL execution
-		 * result.
-		 * @param errors the execution result errors
-		 * @return the current builder
-		 */
-		public Builder errors(@Nullable List<GraphQLError> errors) {
-			this.executionResultBuilder.errors(errors);
-			return this;
-		}
-
-		/**
-		 * Set the {@link ExecutionResult#getExtensions() extensions} of the GraphQL
-		 * execution result.
-		 * @param extensions the execution result extensions
-		 * @return the current builder
-		 */
-		public Builder extensions(@Nullable Map<Object, Object> extensions) {
-			this.executionResultBuilder.extensions(extensions);
-			return this;
-		}
-
-		public WebGraphQlResponse build() {
-			return new WebGraphQlResponse(this.original, this.executionResultBuilder.build());
+		@Override
+		protected WebGraphQlResponse build(WebGraphQlResponse original, ExecutionResult newResult) {
+			return new WebGraphQlResponse(original, newResult);
 		}
 
 	}

spring-graphql/src/test/java/org/springframework/graphql/client/RSocketGraphQlClientBuilderTests.java

@@ -107,7 +107,8 @@ public class RSocketGraphQlClientBuilderTests {
 				return Mono.just(response);
 			};
 
-			GraphQlRSocketController controller = new GraphQlRSocketController(graphQlService);
+			GraphQlRSocketController controller = new GraphQlRSocketController(
+					new GraphQlRSocketHandler(graphQlService, Collections.emptyList()));
 
 			RSocketServer.create()
 					.acceptor(createSocketAcceptor(controller))
@@ -132,6 +133,7 @@ public class RSocketGraphQlClientBuilderTests {
 			return handler.responder();
 		}
 
+		@SuppressWarnings("unused")
 		public void setMockResponse(String document, ExecutionResult result) {
 			ExecutionInput executionInput = ExecutionInput.newExecutionInput().query(document).build();
 			this.responses.put(document, new DefaultExecutionGraphQlResponse(executionInput, result));
@@ -144,23 +146,24 @@ public class RSocketGraphQlClientBuilderTests {
 
 
 	@Controller
-	private static class GraphQlRSocketController extends GraphQlRSocketHandler {
+	private static class GraphQlRSocketController {
 
-		GraphQlRSocketController(ExecutionGraphQlService service) {
-			super(service);
+		private final GraphQlRSocketHandler handler;
+
+		GraphQlRSocketController(GraphQlRSocketHandler handler) {
+			this.handler = handler;
 		}
 
-		@Override
 		@MessageMapping("graphql")
 		public Mono<Map<String, Object>> handle(Map<String, Object> payload) {
-			return super.handle(payload);
+			return this.handler.handle(payload);
 		}
 
-		@Override
 		@MessageMapping("graphql")
 		public Flux<Map<String, Object>> handleSubscription(Map<String, Object> payload) {
-			return super.handleSubscription(payload);
+			return this.handler.handleSubscription(payload);
 		}
+
 	}
 
 }