micrometerTags) {
+ this.micrometerTags = micrometerTags;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ public TransportType getTransport() {
+ return transport;
+ }
+
+ public void setTransport(TransportType transport) {
+ this.transport = transport;
+ }
+
+ @Override
+ public String toString() {
+ return new ToStringCreator(this).append("micrometerTags", micrometerTags)
+ .append("port", port).append("transport", transport).toString();
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java
new file mode 100644
index 00000000..21e44b1d
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java
@@ -0,0 +1,117 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.filter;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.ListIterator;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
+import org.springframework.lang.Nullable;
+
+/**
+ * Default implementation of {@link FilterChain}.
+ *
+ *
+ * Each instance of this class represents one link in the chain. The public constructor
+ * {@link #AbstractFilterChain(List)} initializes the full chain and represents its first
+ * link.
+ *
+ *
+ * This class is immutable and thread-safe. It can be created once and re-used to handle
+ * request concurrently.
+ *
+ * Copied from org.springframework.web.server.handler.AbstractFilterChain
+ *
+ * @since 5.0
+ */
+public abstract class AbstractFilterChain
+ implements FilterChain {
+
+ private final Log log = LogFactory.getLog(getClass());
+
+ protected final List allFilters;
+
+ @Nullable
+ protected final F currentFilter;
+
+ @Nullable
+ protected final FC next;
+
+ /**
+ * Public constructor with the list of filters and the target handler to use.
+ * @param filters the filters ahead of the handler
+ */
+ @SuppressWarnings("unchecked")
+ protected AbstractFilterChain(List filters) {
+ this.allFilters = Collections.unmodifiableList(filters);
+ FC chain = initChain(filters);
+ this.currentFilter = (F) chain.currentFilter;
+ this.next = (FC) chain.next;
+ }
+
+ private FC initChain(List filters) {
+ FC chain = create(filters, null, null);
+ ListIterator iterator = filters.listIterator(filters.size());
+ while (iterator.hasPrevious()) {
+ chain = create(filters, iterator.previous(), chain);
+ }
+ return chain;
+ }
+
+ /**
+ * Private constructor to represent one link in the chain.
+ */
+ protected AbstractFilterChain(List allFilters, @Nullable F currentFilter,
+ @Nullable FC next) {
+
+ this.allFilters = allFilters;
+ this.currentFilter = currentFilter;
+ this.next = next;
+ }
+
+ /**
+ * Private constructor to represent one link in the chain.
+ */
+ protected abstract FC create(List allFilters, @Nullable F currentFilter,
+ @Nullable FC next);
+
+ public List getFilters() {
+ return this.allFilters;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Mono filter(E exchange) {
+ return Mono.defer(() -> this.currentFilter != null && this.next != null
+ ? this.currentFilter.filter(exchange, this.next) : getMonoSuccess());
+ }
+
+ private Mono getMonoSuccess() {
+ if (log.isDebugEnabled()) {
+ log.debug("filter chain completed with success");
+ }
+ return MONO_SUCCESS;
+ }
+
+ private static final Mono MONO_SUCCESS = Mono.just(Success.INSTANCE);
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java
new file mode 100644
index 00000000..1ebcc74b
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.filter;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public abstract class AbstractRSocketExchange implements RSocketExchange {
+
+ private final Map attributes = new ConcurrentHashMap<>();
+
+ @Override
+ public Map getAttributes() {
+ return this.attributes;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java
new file mode 100644
index 00000000..041e2e8a
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.filter;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
+
+/**
+ * Contract to allow a {@link RSocketFilter} to delegate to the next in the chain.
+ *
+ * @author Spencer Gibb
+ */
+public interface FilterChain {
+
+ /**
+ * Delegate to the next {@code WebFilter} in the chain.
+ * @param exchange the current server exchange
+ * @return {@code Mono} to indicate when request handling is complete
+ */
+ Mono filter(E exchange);
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java
new file mode 100644
index 00000000..52f4806e
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.filter;
+
+import java.util.Map;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+public interface RSocketExchange {
+
+ /**
+ * Return a mutable map of request attributes for the current exchange.
+ * @return current attributes.
+ */
+ Map getAttributes();
+
+ /**
+ * Return the request attribute value if present.
+ * @param name the attribute name
+ * @param the attribute type
+ * @return the attribute value
+ */
+ @SuppressWarnings("unchecked")
+ @Nullable
+ default T getAttribute(String name) {
+ return (T) getAttributes().get(name);
+ }
+
+ /**
+ * Return the request attribute value or if not present raise an
+ * {@link IllegalArgumentException}.
+ * @param name the attribute name
+ * @param the attribute type
+ * @return the attribute value
+ */
+ @SuppressWarnings("unchecked")
+ default T getRequiredAttribute(String name) {
+ T value = getAttribute(name);
+ Assert.notNull(value, () -> "Required attribute '" + name + "' is missing");
+ return value;
+ }
+
+ /**
+ * Return the request attribute value, or a default, fallback value.
+ * @param name the attribute name
+ * @param defaultValue a default value to return instead
+ * @param the attribute type
+ * @return the attribute value
+ */
+ @SuppressWarnings("unchecked")
+ default T getAttributeOrDefault(String name, T defaultValue) {
+ return (T) getAttributes().getOrDefault(name, defaultValue);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java
new file mode 100644
index 00000000..6b19f231
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.filter;
+
+import reactor.core.publisher.Mono;
+
+/**
+ * Contract for interception-style, chained processing of Web requests that may be used to
+ * implement cross-cutting, application-agnostic requirements such as security, timeouts,
+ * and others.
+ *
+ * Copied from WebFilter
+ *
+ * @author Spencer Gibb
+ */
+public interface RSocketFilter> {
+
+ /**
+ * Enum to signal successful end of chain reached without the end being empty, i.e.
+ * Mono<Void> via Mono.empty(). This is because at the end of the chain an
+ * actual value needs to be returned. We can map success, but not empty.
+ */
+ enum Success {
+
+ INSTANCE
+
+ } // should never have more than one value
+
+ /**
+ * Process the Web request and (optionally) delegate to the next {@code RSocketFilter}
+ * through the given {@link FilterChain}.
+ * @param exchange the current RSocket exchange
+ * @param chain provides a way to delegate to the next filter
+ * @return {@code Mono} to indicate when request processing is complete.
+ */
+ Mono filter(E exchange, FC chain);
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java
new file mode 100644
index 00000000..f698d719
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java
@@ -0,0 +1,249 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.metrics;
+
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.Timer;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.ResponderRSocket;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+
+import org.springframework.util.Assert;
+
+import static reactor.core.publisher.SignalType.CANCEL;
+import static reactor.core.publisher.SignalType.ON_COMPLETE;
+import static reactor.core.publisher.SignalType.ON_ERROR;
+
+public class MicrometerResponderRSocket implements ResponderRSocket {
+
+ private static final Log log = LogFactory.getLog(MicrometerResponderRSocket.class);
+
+ private final RSocket delegate;
+
+ private final InteractionCounters metadataPush;
+
+ private final InteractionCounters requestChannel;
+
+ private final InteractionCounters requestFireAndForget;
+
+ private final InteractionTimers requestResponse;
+
+ private final InteractionCounters requestStream;
+
+ /**
+ * Creates a new {@link RSocket}.
+ * @param delegate the {@link RSocket} to delegate to
+ * @param meterRegistry the {@link MeterRegistry} to use
+ * @param tags additional tags to attach to {@link Meter}s
+ * @throws IllegalArgumentException if {@code delegate} or {@code meterRegistry} is
+ * {@code null}
+ */
+ public MicrometerResponderRSocket(RSocket delegate, MeterRegistry meterRegistry,
+ Tag... tags) {
+ Assert.notNull(delegate, "delegate must not be null");
+ Assert.notNull(meterRegistry, "meterRegistry must not be null");
+
+ this.delegate = delegate;
+ this.metadataPush = new InteractionCounters(meterRegistry, "metadata.push", tags);
+ this.requestChannel = new InteractionCounters(meterRegistry, "request.channel",
+ tags);
+ this.requestFireAndForget = new InteractionCounters(meterRegistry, "request.fnf",
+ tags);
+ this.requestResponse = new InteractionTimers(meterRegistry, "request.response",
+ tags);
+ this.requestStream = new InteractionCounters(meterRegistry, "request.stream",
+ tags);
+ }
+
+ @Override
+ public void dispose() {
+ delegate.dispose();
+ }
+
+ @Override
+ public Mono fireAndForget(Payload payload) {
+ return delegate.fireAndForget(payload).doFinally(requestFireAndForget);
+ }
+
+ @Override
+ public Mono metadataPush(Payload payload) {
+ return delegate.metadataPush(payload).doFinally(metadataPush);
+ }
+
+ @Override
+ public Mono onClose() {
+ return delegate.onClose();
+ }
+
+ @Override
+ public Flux requestChannel(Publisher payloads) {
+ return delegate.requestChannel(payloads).doFinally(requestChannel);
+ }
+
+ @Override
+ public Mono requestResponse(Payload payload) {
+ return Mono.defer(() -> {
+ Timer.Sample sample = requestResponse.start();
+
+ return delegate.requestResponse(payload)
+ .doFinally(signalType -> requestResponse.accept(sample, signalType));
+ });
+ }
+
+ @Override
+ public Flux requestStream(Payload payload) {
+ return delegate.requestStream(payload).doFinally(requestStream);
+ }
+
+ @Override
+ public Flux requestChannel(Payload payload, Publisher payloads) {
+ if (delegate instanceof ResponderRSocket) {
+ ResponderRSocket rSocket = (ResponderRSocket) delegate;
+ return rSocket.requestChannel(payload, payloads).doFinally(requestChannel);
+ }
+ return delegate.requestChannel(payloads).doFinally(requestChannel);
+ }
+
+ private static final class InteractionCounters implements Consumer {
+
+ private final Counter cancel;
+
+ private final Counter onComplete;
+
+ private final Counter onError;
+
+ private InteractionCounters(MeterRegistry meterRegistry, String interactionModel,
+ Tag... tags) {
+ this.cancel = counter(meterRegistry, interactionModel, CANCEL, tags);
+ this.onComplete = counter(meterRegistry, interactionModel, ON_COMPLETE, tags);
+ this.onError = counter(meterRegistry, interactionModel, ON_ERROR, tags);
+ }
+
+ @Override
+ public void accept(SignalType signalType) {
+ switch (signalType) {
+ case CANCEL:
+ if (this.cancel != null) {
+ this.cancel.increment();
+ }
+ break;
+ case ON_COMPLETE:
+ if (this.onComplete != null) {
+ this.onComplete.increment();
+ }
+ break;
+ case ON_ERROR:
+ if (this.onError != null) {
+ this.onError.increment();
+ }
+ break;
+ }
+ }
+
+ private Counter counter(MeterRegistry meterRegistry, String interactionModel,
+ SignalType signalType, Tag... tags) {
+
+ Tags withType = Tags.of(tags).and("signal.type", signalType.name());
+ try {
+ return meterRegistry.counter("rsocket." + interactionModel, withType);
+ }
+ catch (Exception e) {
+ if (log.isTraceEnabled()) {
+ log.trace("Error creating counter with tags: " + withType, e);
+ }
+ return null;
+ }
+ }
+
+ }
+
+ private static final class InteractionTimers
+ implements BiConsumer {
+
+ private final Timer cancel;
+
+ private final MeterRegistry meterRegistry;
+
+ private final Timer onComplete;
+
+ private final Timer onError;
+
+ private InteractionTimers(MeterRegistry meterRegistry, String interactionModel,
+ Tag... tags) {
+ this.meterRegistry = meterRegistry;
+
+ this.cancel = timer(meterRegistry, interactionModel, CANCEL, tags);
+ this.onComplete = timer(meterRegistry, interactionModel, ON_COMPLETE, tags);
+ this.onError = timer(meterRegistry, interactionModel, ON_ERROR, tags);
+ }
+
+ @Override
+ public void accept(Timer.Sample sample, SignalType signalType) {
+ switch (signalType) {
+ case CANCEL:
+ if (this.cancel != null) {
+ sample.stop(this.cancel);
+ }
+ break;
+ case ON_COMPLETE:
+ if (this.onComplete != null) {
+ sample.stop(this.onComplete);
+ }
+ break;
+ case ON_ERROR:
+ if (this.onError != null) {
+ sample.stop(this.onError);
+ }
+ break;
+ }
+ }
+
+ Timer.Sample start() {
+ return Timer.start(meterRegistry);
+ }
+
+ private static Timer timer(MeterRegistry meterRegistry, String interactionModel,
+ SignalType signalType, Tag... tags) {
+
+ Tags withType = Tags.of(tags).and("signal.type", signalType.name());
+ try {
+ return meterRegistry.timer("rsocket." + interactionModel, withType);
+ }
+ catch (Exception e) {
+ if (log.isTraceEnabled()) {
+ log.trace("Error creating timer with tags: " + withType, e);
+ }
+ return null;
+ }
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java
new file mode 100644
index 00000000..f8480e42
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.metrics;
+
+import java.util.Objects;
+
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.rsocket.RSocket;
+import io.rsocket.plugins.RSocketInterceptor;
+
+public class MicrometerResponderRSocketInterceptor implements RSocketInterceptor {
+
+ private final MeterRegistry meterRegistry;
+
+ private final Tag[] tags;
+
+ /**
+ * Creates a new {@link RSocketInterceptor}.
+ * @param meterRegistry the {@link MeterRegistry} to use to create {@link Meter}s.
+ * @param tags the additional tags to attach to each {@link Meter}
+ * @throws NullPointerException if {@code meterRegistry} is {@code null}
+ */
+ public MicrometerResponderRSocketInterceptor(MeterRegistry meterRegistry,
+ Tag... tags) {
+ this.meterRegistry = Objects.requireNonNull(meterRegistry,
+ "meterRegistry must not be null");
+ this.tags = tags;
+ }
+
+ @Override
+ public MicrometerResponderRSocket apply(RSocket delegate) {
+ Objects.requireNonNull(delegate, "delegate must not be null");
+
+ return new MicrometerResponderRSocket(delegate, meterRegistry, tags);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/LoadBalancedRSocket.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/LoadBalancedRSocket.java
new file mode 100644
index 00000000..e02bfdd9
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/LoadBalancedRSocket.java
@@ -0,0 +1,129 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.registry;
+
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+import io.rsocket.RSocket;
+import io.rsocket.util.RSocketProxy;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+public class LoadBalancedRSocket {
+
+ private static final Log log = LogFactory.getLog(LoadBalancedRSocket.class);
+
+ private final List delegates = new CopyOnWriteArrayList<>();
+
+ private final String serviceName;
+
+ private final LoadBalancer loadBalancer;
+
+ public LoadBalancedRSocket(String serviceName) {
+ this(serviceName, new RoundRobinLoadBalancer(serviceName));
+ }
+
+ public LoadBalancedRSocket(String serviceName, LoadBalancer loadBalancer) {
+ this.serviceName = serviceName;
+ this.loadBalancer = loadBalancer;
+ }
+
+ public Mono choose() {
+ return this.loadBalancer.apply(this.delegates);
+ }
+
+ public void addRSocket(RSocket rsocket, Metadata metadata) {
+ this.delegates.add(new EnrichedRSocket(rsocket, metadata));
+ }
+
+ public void remove(Metadata metadata) {
+ // TODO: move delegates to a map for easy removal
+ this.delegates.stream()
+ .filter(enriched -> metadata.matches(enriched.getMetadata())).findFirst()
+ .ifPresent(this.delegates::remove);
+ }
+
+ public List getDelegates() {
+ return this.delegates;
+ }
+
+ public static class EnrichedRSocket extends RSocketProxy {
+
+ private final Metadata metadata;
+
+ public EnrichedRSocket(RSocket source, Metadata metadata) {
+ super(source);
+ this.metadata = metadata;
+ }
+
+ public Metadata getMetadata() {
+ return this.metadata;
+ }
+
+ public RSocket getSource() {
+ return this.source;
+ }
+
+ }
+
+ // TODO: Flux as input?
+ // TODO: reuse commons load balancer?
+ public interface LoadBalancer
+ extends Function, Mono> {
+
+ }
+
+ public static class RoundRobinLoadBalancer implements LoadBalancer {
+
+ private final AtomicInteger position;
+
+ private final String serviceName;
+
+ public RoundRobinLoadBalancer(String serviceName) {
+ this(serviceName, new Random().nextInt(1000));
+ }
+
+ public RoundRobinLoadBalancer(String serviceName, int seedPosition) {
+ this.serviceName = serviceName;
+ this.position = new AtomicInteger(seedPosition);
+ }
+
+ @Override
+ public Mono apply(List rSockets) {
+ if (rSockets.isEmpty()) {
+ if (log.isWarnEnabled()) {
+ log.warn("No servers available for: " + this.serviceName);
+ }
+ return Mono.empty();
+ }
+ // TODO: enforce order?
+ int pos = Math.abs(this.position.incrementAndGet());
+
+ EnrichedRSocket rSocket = rSockets.get(pos % rSockets.size());
+ return Mono.just(rSocket);
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/Registry.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/Registry.java
new file mode 100644
index 00000000..2185a114
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/Registry.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.registry;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import io.rsocket.RSocket;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.DirectProcessor;
+import reactor.core.publisher.FluxSink;
+
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+import org.springframework.util.Assert;
+
+/**
+ * The Registry handles all RSocket connections that have been made that have associated
+ * announcement metadata. RSocket connections can then be found based on routing metadata.
+ * When a new RSocket is registered, a RegisteredEvent is pushed onto a DirectProcessor
+ * that is acting as an event bus for registered Consumers.
+ */
+// TODO: name?
+public class Registry {
+
+ private static final Log log = LogFactory.getLog(Registry.class);
+
+ private final Map rsockets = new ConcurrentHashMap<>();
+
+ private final DirectProcessor registeredEvents = DirectProcessor
+ .create();
+
+ private final FluxSink registeredEventsSink = registeredEvents
+ .sink(FluxSink.OverflowStrategy.DROP);
+
+ public Registry() {
+ }
+
+ // TODO: Mono?
+ public void register(Metadata metadata, RSocket rsocket) {
+ Assert.notNull(metadata, "metadata may not be null");
+ Assert.notNull(rsocket, "RSocket may not be null");
+ if (log.isDebugEnabled()) {
+ log.debug("Registering RSocket: " + metadata);
+ }
+ LoadBalancedRSocket composite = rsockets.computeIfAbsent(metadata.getName(),
+ s -> new LoadBalancedRSocket(metadata.getName()));
+ composite.addRSocket(rsocket, metadata);
+ registeredEventsSink.next(new RegisteredEvent(metadata, rsocket));
+ }
+
+ public void deregister(Metadata metadata) {
+ Assert.notNull(metadata, "metadata may not be null");
+ if (log.isDebugEnabled()) {
+ log.debug("Deregistering RSocket: " + metadata);
+ }
+ LoadBalancedRSocket loadBalanced = this.rsockets.get(metadata.getName());
+ if (loadBalanced != null) {
+ loadBalanced.remove(metadata);
+ }
+ }
+
+ public LoadBalancedRSocket getRegistered(Metadata metadata) {
+ return rsockets.get(metadata.getName());
+ }
+
+ public Disposable addListener(Consumer consumer) {
+ return this.registeredEvents.subscribe(consumer);
+ }
+
+ public static class RegisteredEvent {
+
+ private final Metadata routingMetadata;
+
+ private final RSocket rSocket;
+
+ public RegisteredEvent(Metadata routingMetadata, RSocket rSocket) {
+ Assert.notNull(routingMetadata, "routingMetadata may not be null");
+ Assert.notNull(rSocket, "RSocket may not be null");
+ this.routingMetadata = routingMetadata;
+ this.rSocket = rSocket;
+ }
+
+ public Metadata getRoutingMetadata() {
+ return routingMetadata;
+ }
+
+ public RSocket getRSocket() {
+ return rSocket;
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistryRoutes.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistryRoutes.java
new file mode 100644
index 00000000..60d136d0
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistryRoutes.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.registry;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Consumer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.route.Route;
+import org.springframework.cloud.gateway.rsocket.route.Routes;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+/**
+ * Creates routes from RegisteredEvents.
+ */
+public class RegistryRoutes implements Routes, Consumer {
+
+ private static final Log log = LogFactory.getLog(RegistryRoutes.class);
+
+ private Map routes = new ConcurrentHashMap<>();
+
+ @Override
+ public Flux getRoutes() {
+ // TODO: sorting
+ // TODO: caching
+ Collection routeCollection = routes.values();
+ if (log.isDebugEnabled()) {
+ log.debug("Found routes: " + routeCollection);
+ }
+ return Flux.fromIterable(routeCollection);
+ }
+
+ @Override
+ public void accept(Registry.RegisteredEvent registeredEvent) {
+ Metadata routingMetadata = registeredEvent.getRoutingMetadata();
+ String id = getId(routingMetadata);
+
+ routes.computeIfAbsent(id, key -> createRoute(id, routingMetadata));
+ }
+
+ private String getId(Metadata routingMetadata) {
+ String id = routingMetadata.getName();
+ if (id == null) {
+ id = UUID.randomUUID().toString();
+ }
+ return id;
+ }
+
+ private Route createRoute(String id, Metadata routingMetadata) {
+ Route route = Route.builder().id(id).routingMetadata(routingMetadata)
+ .predicate(exchange -> {
+ // TODO: standard predicates
+ // TODO: allow customized predicates
+ Metadata incomingRouting = exchange.getRoutingMetadata();
+ boolean matches = incomingRouting.getName()
+ .equalsIgnoreCase(routingMetadata.getName());
+ return Mono.just(matches);
+ })
+ // TODO: allow customized filters
+ .build();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Created Route for registered service " + route);
+ }
+
+ return route;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistrySocketAcceptorFilter.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistrySocketAcceptorFilter.java
new file mode 100644
index 00000000..f790ced2
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistrySocketAcceptorFilter.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.registry;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorExchange;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilterChain;
+import org.springframework.core.Ordered;
+import org.springframework.util.StringUtils;
+
+/**
+ * Filter that registers the SendingSocket.
+ */
+public class RegistrySocketAcceptorFilter implements SocketAcceptorFilter, Ordered {
+
+ private final Registry registry;
+
+ public RegistrySocketAcceptorFilter(Registry registry) {
+ this.registry = registry;
+ }
+
+ @Override
+ public Mono filter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ if (exchange.getMetadata() != null
+ && StringUtils.hasLength(exchange.getMetadata().getName())) {
+ this.registry.register(exchange.getMetadata(), exchange.getSendingSocket());
+ }
+
+ return chain.filter(exchange);
+ }
+
+ @Override
+ public int getOrder() {
+ return HIGHEST_PRECEDENCE + 1000;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java
new file mode 100644
index 00000000..e645b7e4
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java
@@ -0,0 +1,203 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.route;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
+import org.springframework.cloud.gateway.rsocket.server.GatewayFilter;
+import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+import org.springframework.core.Ordered;
+import org.springframework.core.style.ToStringCreator;
+import org.springframework.util.Assert;
+
+/**
+ * @author Spencer Gibb
+ */
+public class Route implements Ordered {
+
+ private final String id;
+
+ private final Metadata targetMetadata;
+
+ private final int order;
+
+ private final AsyncPredicate predicate;
+
+ private final List gatewayFilters;
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ private Route(String id, Metadata targetMetadata, int order,
+ AsyncPredicate predicate,
+ List gatewayFilters) {
+ this.id = id;
+ this.targetMetadata = targetMetadata;
+ this.order = order;
+ this.predicate = predicate;
+ this.gatewayFilters = gatewayFilters;
+ }
+
+ public String getId() {
+ return this.id;
+ }
+
+ public Metadata getTargetMetadata() {
+ return this.targetMetadata;
+ }
+
+ public int getOrder() {
+ return order;
+ }
+
+ public AsyncPredicate getPredicate() {
+ return this.predicate;
+ }
+
+ public List getFilters() {
+ return Collections.unmodifiableList(this.gatewayFilters);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Route route = (Route) o;
+ return Objects.equals(id, route.id)
+ && Objects.equals(targetMetadata, route.targetMetadata)
+ && Objects.equals(order, route.order)
+ && Objects.equals(predicate, route.predicate)
+ && Objects.equals(gatewayFilters, route.gatewayFilters);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, targetMetadata, predicate, gatewayFilters);
+ }
+
+ @Override
+ public String toString() {
+ return new ToStringCreator(this).append("id", id)
+ .append("targetMetadata", targetMetadata).append("order", order)
+ .append("predicate", predicate).append("gatewayFilters", gatewayFilters)
+ .toString();
+ }
+
+ public static class Builder {
+
+ protected String id;
+
+ protected Metadata routingMetadata;
+
+ protected int order = 0;
+
+ protected AsyncPredicate predicate;
+
+ protected List gatewayFilters = new ArrayList<>();
+
+ protected Builder() {
+ }
+
+ public Builder id(String id) {
+ this.id = id;
+ return this;
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public Builder order(int order) {
+ this.order = order;
+ return this;
+ }
+
+ public AsyncPredicate getPredicate() {
+ return this.predicate;
+ }
+
+ public Builder routingMetadata(Metadata routingMetadata) {
+ this.routingMetadata = routingMetadata;
+ return this;
+ }
+
+ public Builder setFilters(List gatewayFilters) {
+ this.gatewayFilters = gatewayFilters;
+ return this;
+ }
+
+ public Builder filter(GatewayFilter gatewayFilter) {
+ this.gatewayFilters.add(gatewayFilter);
+ return this;
+ }
+
+ public Builder filters(Collection gatewayFilters) {
+ this.gatewayFilters.addAll(gatewayFilters);
+ return this;
+ }
+
+ public Builder filters(GatewayFilter... gatewayFilters) {
+ return filters(Arrays.asList(gatewayFilters));
+ }
+
+ public Builder predicate(AsyncPredicate predicate) {
+ this.predicate = predicate;
+ return this;
+ }
+
+ public Builder and(AsyncPredicate predicate) {
+ Assert.notNull(this.predicate, "can not call and() on null predicate");
+ this.predicate = this.predicate.and(predicate);
+ return this;
+ }
+
+ public Builder or(AsyncPredicate predicate) {
+ Assert.notNull(this.predicate, "can not call or() on null predicate");
+ this.predicate = this.predicate.or(predicate);
+ return this;
+ }
+
+ public Builder negate() {
+ Assert.notNull(this.predicate, "can not call negate() on null predicate");
+ this.predicate = this.predicate.negate();
+ return this;
+ }
+
+ public Route build() {
+ Assert.notNull(this.id, "id can not be null");
+ Assert.notNull(this.routingMetadata, "targetMetadata can not be null");
+ Assert.notNull(this.predicate, "predicate can not be null");
+
+ return new Route(this.id, this.routingMetadata, this.order, predicate,
+ this.gatewayFilters);
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java
new file mode 100644
index 00000000..4db13e31
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.route;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
+
+/**
+ * @author Spencer Gibb
+ */
+public interface Routes {
+
+ /** log. */
+ Log log = LogFactory.getLog(Routes.class);
+
+ Flux getRoutes();
+
+ default Mono findRoute(GatewayExchange exchange) {
+ return getRoutes()
+ // individually filter routes so that filterWhen error delaying is not a
+ // problem
+ .concatMap(route -> Mono.just(route).filterWhen(r -> {
+ // add the current route we are testing
+ // TODO: exchange attributes
+ // exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR,
+ // r.getId());
+ return r.getPredicate().apply(exchange);
+ })
+ // instead of immediately stopping main flux due to error, log and
+ // swallow it
+ .doOnError(e -> log.error(
+ "Error applying predicate for route: " + route.getId(),
+ e))
+ .onErrorResume(e -> Mono.empty()))
+ .next().map(route -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Route matched: " + route.getId());
+ }
+ return route;
+ });
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayExchange.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayExchange.java
new file mode 100644
index 00000000..fa9d896b
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayExchange.java
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import io.micrometer.core.instrument.Tags;
+import io.rsocket.Payload;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.gateway.rsocket.filter.AbstractRSocketExchange;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+/**
+ * Exchange object used in GatewayFilterChain started by GatewayRSocket.
+ */
+public class GatewayExchange extends AbstractRSocketExchange {
+
+ private static final Log log = LogFactory.getLog(GatewayExchange.class);
+
+ /**
+ * Key for the route object in attributes.
+ */
+ public static final String ROUTE_ATTR = "__route_attr_";
+
+ enum Type {
+
+ FIRE_AND_FORGET("request.fnf"), REQUEST_CHANNEL(
+ "request.channel"), REQUEST_RESPONSE(
+ "request.response"), REQUEST_STREAM("request.stream");
+
+ private String key;
+
+ Type(String key) {
+ this.key = key;
+ }
+
+ String getKey() {
+ return this.key;
+ }
+
+ }
+
+ private final Type type;
+
+ private final Metadata routingMetadata;
+
+ private Tags tags = Tags.empty();
+
+ public static GatewayExchange fromPayload(Type type, Payload payload) {
+ return new GatewayExchange(type, getRoutingMetadata(payload));
+ }
+
+ private static Metadata getRoutingMetadata(Payload payload) {
+ if (payload == null || !payload.hasMetadata()) { // and metadata is routing
+ return null;
+ }
+
+ // TODO: deal with composite metadata
+
+ Metadata metadata = Metadata.decodeMetadata(payload.sliceMetadata());
+
+ if (log.isDebugEnabled()) {
+ log.debug("found routing metadata " + metadata);
+ }
+ return metadata;
+ }
+
+ public GatewayExchange(Type type, Metadata routingMetadata) {
+ this.type = type;
+ this.routingMetadata = routingMetadata;
+ }
+
+ public Type getType() {
+ return type;
+ }
+
+ public Metadata getRoutingMetadata() {
+ return routingMetadata;
+ }
+
+ public Tags getTags() {
+ return this.tags;
+ }
+
+ public void setTags(Tags tags) {
+ this.tags = tags;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilter.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilter.java
new file mode 100644
index 00000000..f5a88089
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilter.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter;
+
+public interface GatewayFilter
+ extends RSocketFilter {
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilterChain.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilterChain.java
new file mode 100644
index 00000000..00c3d9da
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilterChain.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import java.util.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.filter.AbstractFilterChain;
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
+
+public class GatewayFilterChain
+ extends AbstractFilterChain {
+
+ /**
+ * Public constructor with the list of filters and the target handler to use.
+ * @param filters the filters ahead of the handler
+ */
+ private GatewayFilterChain(List filters) {
+ super(filters);
+ }
+
+ protected GatewayFilterChain(List allFilters,
+ GatewayFilter currentFilter, GatewayFilterChain next) {
+ super(allFilters, currentFilter, next);
+ }
+
+ @Override
+ protected GatewayFilterChain create(List allFilters,
+ GatewayFilter currentFilter, GatewayFilterChain next) {
+ return new GatewayFilterChain(allFilters, currentFilter, next);
+ }
+
+ public static Mono executeFilterChain(List filters,
+ GatewayExchange exchange) {
+ return new GatewayFilterChain(filters).filter(exchange);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayPredicate.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayPredicate.java
new file mode 100644
index 00000000..07d6d406
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayPredicate.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
+
+public interface GatewayPredicate extends AsyncPredicate {
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocket.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocket.java
new file mode 100644
index 00000000..6f89297a
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocket.java
@@ -0,0 +1,329 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.logging.Level;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.Timer;
+import io.rsocket.AbstractRSocket;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.ResponderRSocket;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket;
+import org.springframework.cloud.gateway.rsocket.registry.Registry;
+import org.springframework.cloud.gateway.rsocket.route.Route;
+import org.springframework.cloud.gateway.rsocket.route.Routes;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.ROUTE_ATTR;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.FIRE_AND_FORGET;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_CHANNEL;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_RESPONSE;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_STREAM;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain.executeFilterChain;
+
+/**
+ * Acts as a proxy to other registered sockets. Creates a GatewayExchange and attempts to
+ * locate a Route. If a Route is found, it is added to the exchange and the filter chains
+ * is executed againts the Route's filters. If the filter chain is successful, an attempt
+ * to locate a target RSocket via the Registry is executed. If not found a pending RSocket
+ * * is returned.
+ */
+public class GatewayRSocket extends AbstractRSocket implements ResponderRSocket {
+
+ private static final Log log = LogFactory.getLog(GatewayRSocket.class);
+
+ private final Registry registry;
+
+ private final Routes routes;
+
+ private final MeterRegistry meterRegistry;
+
+ private final GatewayRSocketProperties properties;
+
+ private final Metadata metadata;
+
+ GatewayRSocket(Registry registry, Routes routes, MeterRegistry meterRegistry,
+ GatewayRSocketProperties properties, Metadata metadata) {
+ this.registry = registry;
+ this.routes = routes;
+ this.meterRegistry = meterRegistry;
+ this.properties = properties;
+ this.metadata = metadata;
+ this.onClose().doOnSuccess(v -> registry.deregister(metadata))
+ // .doOnNext(v -> log.error("OnClose doOnNext"))
+ .doOnError(t -> {
+ if (log.isErrorEnabled()) {
+ log.error("Error received, deregistering " + metadata, t);
+ }
+ registry.deregister(metadata);
+ })
+ // .doOnTerminate(() -> log.error("OnClose doOnTerminate"))
+ // .doFinally(st -> log.error("OnClose doFinally"))
+ .subscribe();
+ }
+
+ protected Registry getRegistry() {
+ return registry;
+ }
+
+ protected Routes getRoutes() {
+ return routes;
+ }
+
+ @Override
+ public Mono fireAndForget(Payload payload) {
+ GatewayExchange exchange = createExchange(FIRE_AND_FORGET, payload);
+ return findRSocketOrCreatePending(exchange)
+ .flatMap(rSocket -> rSocket.fireAndForget(payload))
+ .doOnError(t -> count(exchange, "error"))
+ .doFinally(s -> count(exchange, ""));
+ }
+
+ private GatewayExchange createExchange(GatewayExchange.Type type, Payload payload) {
+ GatewayExchange exchange = GatewayExchange.fromPayload(type, payload);
+ Tags tags = getTags(exchange);
+ exchange.setTags(tags);
+ return exchange;
+ }
+
+ private Tags getTags(GatewayExchange exchange) {
+ // TODO: add tags to exchange
+ String requesterName = this.metadata.getName();
+ String requesterId = this.metadata.get("id");
+ String responderName = exchange.getRoutingMetadata().getName();
+ Assert.hasText(responderName, "responderName must not be empty");
+ Assert.hasText(requesterId, "requesterId must not be empty");
+ Assert.hasText(requesterName, "requesterName must not be empty");
+ // responder.id happens in a callback, later
+ return Tags.of("requester.name", requesterName, "responder.name", responderName,
+ "requester.id", requesterId, "gateway.id", this.properties.getId());
+ }
+
+ @Override
+ public Flux requestChannel(Payload payload, Publisher payloads) {
+ GatewayExchange exchange = createExchange(REQUEST_CHANNEL, payload);
+ Tags responderTags = Tags.of("source", "responder");
+ return findRSocketOrCreatePending(exchange).flatMapMany(rSocket -> {
+ Tags requesterTags = Tags.of("source", "requester");
+ Flux flux = Flux.from(payloads)
+ .doOnNext(s -> count(exchange, "payload", requesterTags))
+ .doOnError(t -> count(exchange, "error", requesterTags))
+ .doFinally(s -> count(exchange, requesterTags));
+
+ if (rSocket instanceof ResponderRSocket) {
+ ResponderRSocket socket = (ResponderRSocket) rSocket;
+ return socket.requestChannel(payload, flux).log(
+ GatewayRSocket.class.getName() + ".request-channel",
+ Level.FINEST);
+ }
+ return rSocket.requestChannel(flux);
+ }).doOnNext(s -> count(exchange, "payload", responderTags))
+ .doOnError(t -> count(exchange, "error", responderTags))
+ .doFinally(s -> count(exchange, responderTags));
+ }
+
+ private void count(GatewayExchange exchange, String suffix) {
+ count(exchange, suffix, Tags.empty());
+ }
+
+ private void count(GatewayExchange exchange, Tags additionalTags) {
+ count(exchange, null, additionalTags);
+ }
+
+ private void count(GatewayExchange exchange, String suffix, Tags additionalTags) {
+ Tags tags = exchange.getTags().and(additionalTags);
+ String name = getMetricName(exchange, suffix);
+ this.meterRegistry.counter(name, tags).increment();
+ }
+
+ private String getMetricName(GatewayExchange exchange) {
+ return getMetricName(exchange, null);
+ }
+
+ private String getMetricName(GatewayExchange exchange, String suffix) {
+ StringBuilder name = new StringBuilder("forward.");
+ name.append(exchange.getType().getKey());
+ if (StringUtils.hasLength(suffix)) {
+ name.append(".");
+ name.append(suffix);
+ }
+ return name.toString();
+ }
+
+ @Override
+ public Mono requestResponse(Payload payload) {
+ AtomicReference timer = new AtomicReference<>();
+ GatewayExchange exchange = createExchange(REQUEST_RESPONSE, payload);
+ return findRSocketOrCreatePending(exchange)
+ .flatMap(rSocket -> rSocket.requestResponse(payload))
+ .doOnSubscribe(s -> timer.set(Timer.start(meterRegistry)))
+ .doOnError(t -> count(exchange, "error"))
+ .doFinally(s -> timer.get().stop(meterRegistry
+ .timer(getMetricName(exchange), exchange.getTags())));
+ }
+
+ @Override
+ public Flux requestStream(Payload payload) {
+ GatewayExchange exchange = createExchange(REQUEST_STREAM, payload);
+ return findRSocketOrCreatePending(exchange)
+ .flatMapMany(rSocket -> rSocket.requestStream(payload))
+ // S N E F
+ .doOnNext(s -> count(exchange, "payload"))
+ .doOnError(t -> count(exchange, "error"))
+ .doFinally(s -> count(exchange, Tags.empty()));
+ }
+
+ /**
+ * Attempt to locate target RSocket via filter chain. If not found, create a pending
+ * RSocket.
+ * @param exchange GatewayExchange
+ * @return
+ */
+ private Mono findRSocketOrCreatePending(GatewayExchange exchange) {
+ return findRSocket(exchange)
+ // if a route can't be found or registered RSocket, create pending
+ .switchIfEmpty(createPendingRSocket(exchange));
+ }
+
+ private Mono createPendingRSocket(GatewayExchange exchange) {
+ if (log.isDebugEnabled()) {
+ log.debug("creating pending RSocket for " + exchange.getRoutingMetadata());
+ }
+ PendingRequestRSocket pending = constructPendingRSocket(exchange);
+ Disposable disposable = this.registry.addListener(pending);
+ pending.setSubscriptionDisposable(disposable);
+ return Mono.just(pending);
+ }
+
+ /* for testing */ PendingRequestRSocket constructPendingRSocket(
+ GatewayExchange exchange) {
+ Function> routeFinder = registeredEvent -> getRouteMono(
+ registeredEvent, exchange);
+ return new PendingRequestRSocket(routeFinder, map -> {
+ Tags tags = exchange.getTags().and("responder.id", map.get("id"));
+ exchange.setTags(tags);
+ });
+ }
+
+ protected Mono getRouteMono(Registry.RegisteredEvent registeredEvent,
+ GatewayExchange exchange) {
+ return findRoute(exchange)
+ .log(PendingRequestRSocket.class.getName() + ".find route pending",
+ Level.FINEST)
+ // can this be replaced with filter?
+ .flatMap(
+ route -> matchRoute(route, registeredEvent.getRoutingMetadata()));
+ }
+
+ private Mono findRoute(GatewayExchange exchange) {
+ Mono routeMono;
+ /*
+ * if (this.route != null) { //TODO: cache Route? routeMono = Mono.just(route); }
+ * else {
+ */
+ routeMono = this.routes.findRoute(exchange);
+ // }
+ return routeMono;
+ }
+
+ private Mono matchRoute(Route route, Metadata annoucementMetadata) {
+ Metadata targetMetadata = route.getTargetMetadata();
+ if (targetMetadata.matches(annoucementMetadata)) {
+ return Mono.just(route);
+ }
+ return Mono.empty();
+ }
+
+ /**
+ * First locate Route. If found, put route in exchange and execute filter chain. If
+ * successful, locate target RSocket.
+ * @param exchange GatewayExchange.
+ * @return Target RSocket or empty.
+ */
+ private Mono findRSocket(GatewayExchange exchange) {
+ return this.routes.findRoute(exchange)
+ .log(GatewayRSocket.class.getName() + ".find route", Level.FINEST)
+ .flatMap(route -> {
+ // put route in exchange for later use
+ exchange.getAttributes().put(ROUTE_ATTR, route);
+ return executeFilterChain(route.getFilters(), exchange)
+ .flatMap(success -> {
+ LoadBalancedRSocket loadBalancedRSocket = registry
+ .getRegistered(exchange.getRoutingMetadata());
+
+ return loadBalancedRSocket.choose();
+ }).map(enrichedRSocket -> {
+ Metadata metadata = enrichedRSocket.getMetadata();
+ Tags tags = exchange.getTags().and("responder.id",
+ metadata.get("id"));
+ exchange.setTags(tags);
+ return enrichedRSocket;
+ }).cast(RSocket.class).switchIfEmpty(doOnEmpty(exchange));
+ });
+
+ // TODO: deal with connecting to cluster?
+ }
+
+ private Mono doOnEmpty(GatewayExchange exchange) {
+ if (log.isDebugEnabled()) {
+ log.debug("Unable to find destination RSocket for "
+ + exchange.getRoutingMetadata());
+ }
+ return Mono.empty();
+ }
+
+ public static class Factory {
+
+ private final Registry registry;
+
+ private final Routes routes;
+
+ private final MeterRegistry meterRegistry;
+
+ private final GatewayRSocketProperties properties;
+
+ public Factory(Registry registry, Routes routes, MeterRegistry meterRegistry,
+ GatewayRSocketProperties properties) {
+ this.registry = registry;
+ this.routes = routes;
+ this.meterRegistry = meterRegistry;
+ this.properties = properties;
+ }
+
+ public GatewayRSocket create(Metadata metadata) {
+ return new GatewayRSocket(this.registry, this.routes, this.meterRegistry,
+ this.properties, metadata);
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java
new file mode 100644
index 00000000..c59aae21
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import io.rsocket.RSocketFactory;
+import io.rsocket.SocketAcceptor;
+import io.rsocket.micrometer.MicrometerDuplexConnectionInterceptor;
+import io.rsocket.plugins.RSocketInterceptor;
+import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties.Server.TransportType;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.Ordered;
+import org.springframework.util.Assert;
+
+public class GatewayRSocketServer implements Ordered, SmartLifecycle {
+
+ private static final Log log = LogFactory.getLog(GatewayRSocketServer.class);
+
+ private static final RSocketInterceptor[] EMPTY_INTERCEPTORS = new RSocketInterceptor[0];
+
+ private final GatewayRSocketProperties properties;
+
+ private final SocketAcceptor socketAcceptor;
+
+ private final List serverInterceptors;
+
+ private final AtomicBoolean running = new AtomicBoolean();
+
+ private CloseableChannel closeableChannel;
+
+ private final MeterRegistry meterRegistry;
+
+ public GatewayRSocketServer(GatewayRSocketProperties properties,
+ SocketAcceptor socketAcceptor, MeterRegistry meterRegistry) {
+ this(properties, socketAcceptor, meterRegistry, EMPTY_INTERCEPTORS);
+ }
+
+ public GatewayRSocketServer(GatewayRSocketProperties properties,
+ SocketAcceptor socketAcceptor, MeterRegistry meterRegistry,
+ RSocketInterceptor... interceptors) {
+ Assert.notNull(properties, "properties may not be null");
+ Assert.notNull(socketAcceptor, "socketAcceptor may not be null");
+ Assert.notNull(meterRegistry, "meterRegistry may not be null");
+ Assert.notNull(interceptors, "interceptors may not be null");
+ this.properties = properties;
+ this.socketAcceptor = socketAcceptor;
+ this.meterRegistry = meterRegistry;
+ this.serverInterceptors = Arrays.asList(interceptors);
+ }
+
+ @Override
+ public int getOrder() {
+ // return 0;
+ return HIGHEST_PRECEDENCE;
+ }
+
+ @Override
+ public void start() {
+ if (running.compareAndSet(false, true)) {
+ startServer();
+ }
+ }
+
+ @Override
+ public void stop() {
+ if (running.compareAndSet(true, false)) {
+ if (log.isInfoEnabled()) {
+ log.info("Stopping Gateway RSocket Server");
+ }
+ if (closeableChannel != null) {
+ closeableChannel.dispose();
+ }
+ }
+ }
+
+ @Override
+ public boolean isRunning() {
+ return running.get();
+ }
+
+ protected void startServer() {
+ GatewayRSocketProperties.Server server = properties.getServer();
+ int port = server.getPort();
+
+ TransportType transportType = server.getTransport();
+ TcpServerTransport transport;
+ switch (transportType) {
+ case TCP:
+ transport = TcpServerTransport.create(port);
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "No support for server transport " + transportType);
+ }
+
+ if (log.isInfoEnabled()) {
+ log.info("Starting Gateway RSocket Server on port: " + port + ", transport: "
+ + transportType);
+ }
+
+ RSocketFactory.ServerRSocketFactory factory = RSocketFactory.receive();
+
+ serverInterceptors.forEach(factory::addServerPlugin);
+
+ List micrometerTags = server.getMicrometerTags();
+ Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {}))
+ .and("gateway.id", properties.getId()).stream()
+ .collect(Collectors.toList()).toArray(new Tag[] {});
+
+ factory
+ // TODO: add as bean like serverInterceptors above
+ .addConnectionPlugin(
+ new MicrometerDuplexConnectionInterceptor(meterRegistry, tags))
+ .errorConsumer(throwable -> {
+ if (log.isDebugEnabled()) {
+ log.debug("Error with connection", throwable);
+ }
+ }) // TODO: add configurable errorConsumer
+ .acceptor(this.socketAcceptor).transport(transport).start()
+ .map(closeableChannel -> {
+ this.closeableChannel = closeableChannel;
+ return closeableChannel;
+ }).subscribe();
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/PendingRequestRSocket.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/PendingRequestRSocket.java
new file mode 100644
index 00000000..2a4f4de4
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/PendingRequestRSocket.java
@@ -0,0 +1,152 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.logging.Level;
+
+import io.rsocket.AbstractRSocket;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.ResponderRSocket;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.reactivestreams.Publisher;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+import reactor.util.function.Tuple2;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
+import org.springframework.cloud.gateway.rsocket.registry.Registry.RegisteredEvent;
+import org.springframework.cloud.gateway.rsocket.route.Route;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.ROUTE_ATTR;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_STREAM;
+import static org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain.executeFilterChain;
+
+public class PendingRequestRSocket extends AbstractRSocket
+ implements ResponderRSocket, Consumer {
+
+ private static final Log log = LogFactory.getLog(PendingRequestRSocket.class);
+
+ private final Function> routeFinder;
+
+ private final Consumer metadataCallback;
+
+ private final MonoProcessor rSocketProcessor;
+
+ private Disposable subscriptionDisposable;
+
+ private Route route;
+
+ public PendingRequestRSocket(Function> routeFinder,
+ Consumer metadataCallback) {
+ this(routeFinder, metadataCallback, MonoProcessor.create());
+ }
+
+ /* for testing */ PendingRequestRSocket(
+ Function> routeFinder,
+ Consumer metadataCallback,
+ MonoProcessor rSocketProcessor) {
+ this.routeFinder = routeFinder;
+ this.metadataCallback = metadataCallback;
+ this.rSocketProcessor = rSocketProcessor;
+ }
+
+ /**
+ * Find route (if needed) using pendingExchange. If found, see if the route target
+ * matches the registered service. If it matches, send registered RSocket to
+ * processor. Then execute normal filter chain. If filter chain is successful, execute
+ * request.
+ * @param registeredEvent the RegisteredEvent
+ */
+ @Override
+ public void accept(RegisteredEvent registeredEvent) {
+ this.routeFinder.apply(registeredEvent).subscribe(route -> {
+ this.route = route;
+ this.metadataCallback.accept(registeredEvent.getRoutingMetadata());
+ this.rSocketProcessor.onNext(registeredEvent.getRSocket());
+ this.rSocketProcessor.onComplete();
+ if (this.subscriptionDisposable != null) {
+ this.subscriptionDisposable.dispose();
+ }
+ });
+ }
+
+ @Override
+ public Mono fireAndForget(Payload payload) {
+ return processor("pending-request-faf", payload)
+ .flatMap(tuple -> tuple.getT1().fireAndForget(payload));
+ }
+
+ @Override
+ public Mono requestResponse(Payload payload) {
+ return processor("pending-request-rr", payload)
+ .flatMap(tuple -> tuple.getT1().requestResponse(payload));
+ }
+
+ @Override
+ public Flux requestStream(Payload payload) {
+ return processor("pending-request-rs", payload)
+ .flatMapMany(tuple -> tuple.getT1().requestStream(payload));
+ }
+
+ @Override
+ public Flux requestChannel(Payload payload, Publisher payloads) {
+ return processor("pending-request-rc", payload).flatMapMany(tuple -> {
+ RSocket rSocket = tuple.getT1();
+ if (rSocket instanceof ResponderRSocket) {
+ ResponderRSocket socket = (ResponderRSocket) rSocket;
+ return socket.requestChannel(payload, payloads);
+ }
+ return rSocket.requestChannel(payloads);
+ });
+ }
+
+ /**
+ * After processor receives onNext signal, get route from exchange attrs, create a new
+ * exchange from payload. Copy exchange attrs. Execute filter chain, if successful,
+ * execute request.
+ * @param logCategory log category
+ * @param payload payload.
+ * @return
+ */
+ protected Mono> processor(String logCategory,
+ Payload payload) {
+ return rSocketProcessor
+ .log(PendingRequestRSocket.class.getName() + "." + logCategory,
+ Level.FINEST)
+ .flatMap(rSocket -> {
+ GatewayExchange exchange = GatewayExchange.fromPayload(REQUEST_STREAM,
+ payload);
+ exchange.getAttributes().put(ROUTE_ATTR, route);
+ // exchange.getAttributes().putAll(pendingExchange.getAttributes());
+ return Mono.just(rSocket)
+ .zipWith(executeFilterChain(route.getFilters(), exchange));
+ });
+
+ }
+
+ public void setSubscriptionDisposable(Disposable subscriptionDisposable) {
+ this.subscriptionDisposable = subscriptionDisposable;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
new file mode 100644
index 00000000..eee73510
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.util.List;
+import java.util.logging.Level;
+import java.util.stream.Collectors;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import io.rsocket.ConnectionSetupPayload;
+import io.rsocket.RSocket;
+import io.rsocket.SocketAcceptor;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.metrics.MicrometerResponderRSocket;
+import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+public class GatewaySocketAcceptor implements SocketAcceptor {
+
+ private final SocketAcceptorFilterChain filterChain;
+
+ private final GatewayRSocket.Factory rSocketFactory;
+
+ private final MeterRegistry meterRegistry;
+
+ private final GatewayRSocketProperties properties;
+
+ public GatewaySocketAcceptor(GatewayRSocket.Factory rSocketFactory,
+ List filters, MeterRegistry meterRegistry,
+ GatewayRSocketProperties properties) {
+ this.rSocketFactory = rSocketFactory;
+ this.filterChain = new SocketAcceptorFilterChain(filters);
+ this.meterRegistry = meterRegistry;
+ this.properties = properties;
+ }
+
+ @Override
+ @SuppressWarnings("Duplicates")
+ public Mono accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
+
+ // decorate GatewayRSocket with metrics
+ // current gateway id, type requester, service name (from metadata), service id
+
+ Tags requesterTags = Tags.of("gateway.id", properties.getId(), "type",
+ "requester");
+
+ Tags metadataTags;
+ SocketAcceptorExchange exchange;
+ if (setup.hasMetadata()) { // TODO: and setup.metadataMimeType() is Announcement
+ // metadata or composite
+ Metadata metadata = Metadata.decodeMetadata(setup.sliceMetadata());
+ metadataTags = Tags.of("service.name", metadata.getName()).and("service.id",
+ metadata.get("id"));
+ // enrich exchange to have metadata
+ exchange = new SocketAcceptorExchange(setup,
+ decorate(sendingSocket, requesterTags.and(metadataTags)), metadata);
+ }
+ else {
+ metadataTags = Tags.of("service.name", "UNKNOWN").and("service.id",
+ "UNKNOWN");
+ exchange = new SocketAcceptorExchange(setup,
+ decorate(sendingSocket, requesterTags));
+ }
+
+ Tags responderTags = Tags
+ .of("gateway.id", properties.getId(), "type", "responder")
+ .and(metadataTags);
+
+ // decorate with metrics gateway id, type responder, service name, service id
+ // (instance id)
+ return this.filterChain.filter(exchange)
+ .log(GatewaySocketAcceptor.class.getName()
+ + ".socket acceptor filter chain", Level.FINEST)
+ .map(success -> decorate(
+ this.rSocketFactory.create(exchange.getMetadata()),
+ responderTags));
+ }
+
+ private RSocket decorate(RSocket rSocket, Tags tags) {
+ Tag[] tagArray = tags.stream().collect(Collectors.toList()).toArray(new Tag[] {});
+ return new MicrometerResponderRSocket(rSocket, meterRegistry, tagArray);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java
new file mode 100644
index 00000000..1108e20c
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.util.Collections;
+
+import io.rsocket.ConnectionSetupPayload;
+import io.rsocket.RSocket;
+
+import org.springframework.cloud.gateway.rsocket.filter.AbstractRSocketExchange;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+public class SocketAcceptorExchange extends AbstractRSocketExchange {
+
+ private final ConnectionSetupPayload setup;
+
+ private final RSocket sendingSocket;
+
+ private final Metadata metadata;
+
+ public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket) {
+ this(setup, sendingSocket, new Metadata(null, Collections.emptyMap()));
+ }
+
+ public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket,
+ Metadata metadata) {
+ this.setup = setup;
+ this.sendingSocket = sendingSocket;
+ this.metadata = metadata;
+ }
+
+ public ConnectionSetupPayload getSetup() {
+ return setup;
+ }
+
+ public RSocket getSendingSocket() {
+ return sendingSocket;
+ }
+
+ public Metadata getMetadata() {
+ return metadata;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java
new file mode 100644
index 00000000..c27c2791
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter;
+
+public interface SocketAcceptorFilter
+ extends RSocketFilter {
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java
new file mode 100644
index 00000000..e21672d1
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.util.List;
+
+import org.springframework.cloud.gateway.rsocket.filter.AbstractFilterChain;
+
+public class SocketAcceptorFilterChain extends
+ AbstractFilterChain {
+
+ /**
+ * Public constructor with the list of filters and the target handler to use.
+ * @param filters the filters ahead of the handler
+ */
+ public SocketAcceptorFilterChain(List filters) {
+ super(filters);
+ }
+
+ public SocketAcceptorFilterChain(List allFilters,
+ SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
+ super(allFilters, currentFilter, next);
+ }
+
+ @Override
+ protected SocketAcceptorFilterChain create(List allFilters,
+ SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
+ return new SocketAcceptorFilterChain(allFilters, currentFilter, next);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java
new file mode 100644
index 00000000..836ba951
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
+
+public interface SocketAcceptorPredicate extends AsyncPredicate {
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java
new file mode 100644
index 00000000..e659d58c
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.util.List;
+
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
+import org.springframework.core.Ordered;
+import org.springframework.util.Assert;
+
+public class SocketAcceptorPredicateFilter implements SocketAcceptorFilter, Ordered {
+
+ private final AsyncPredicate predicate;
+
+ // TODO: change from List to Flux?
+ public SocketAcceptorPredicateFilter(List predicates) {
+ Assert.notNull(predicates, "predicates may not be null");
+ if (predicates.isEmpty()) {
+ predicate = exchange -> Mono.just(true);
+ }
+ else {
+ AsyncPredicate combined = predicates.get(0);
+ for (SocketAcceptorPredicate p : predicates.subList(1, predicates.size())) {
+ combined = combined.and(p);
+ }
+ predicate = combined;
+ }
+ }
+
+ @Override
+ public int getOrder() {
+ return HIGHEST_PRECEDENCE + 10000;
+ }
+
+ @Override
+ public Mono filter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ return Mono.from(predicate.apply(exchange)).flatMap(test -> {
+ if (test) {
+ return chain.filter(exchange);
+ }
+ return Mono.empty();
+ });
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java
new file mode 100644
index 00000000..5707c86b
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ * http://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.cloud.gateway.rsocket.support;
+
+import java.util.function.Function;
+
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Ben Hale
+ */
+public interface AsyncPredicate extends Function> {
+
+ default AsyncPredicate and(AsyncPredicate other) {
+ Assert.notNull(other, "other must not be null");
+
+ return t -> Flux.zip(apply(t), other.apply(t))
+ .map(tuple -> tuple.getT1() && tuple.getT2());
+ }
+
+ default AsyncPredicate negate() {
+ return t -> Mono.from(apply(t)).map(b -> !b);
+ }
+
+ default AsyncPredicate or(AsyncPredicate other) {
+ Assert.notNull(other, "other must not be null");
+
+ return t -> Flux.zip(apply(t), other.apply(t))
+ .map(tuple -> tuple.getT1() || tuple.getT2());
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/Metadata.java b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/Metadata.java
new file mode 100644
index 00000000..c6f5357d
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/Metadata.java
@@ -0,0 +1,207 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.support;
+
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import io.rsocket.util.NumberUtils;
+
+import org.springframework.core.style.ToStringCreator;
+import org.springframework.util.Assert;
+
+public class Metadata {
+
+ /**
+ * Mime type of routing extension.
+ */
+ public static final String ROUTING_MIME_TYPE = "message/x.rsocket.routing.v0";
+
+ /**
+ * The logical name.
+ */
+ private final String name;
+
+ /**
+ * Keys and values associated with name.
+ */
+ private final Map properties;
+
+ public Metadata(String name, Map properties) {
+ this.name = name;
+ this.properties = properties;
+ }
+
+ public String getName() {
+ return this.name;
+ }
+
+ public Map getProperties() {
+ return this.properties;
+ }
+
+ public String get(String key) {
+ return this.properties.get(key);
+ }
+
+ public String put(String key, String value) {
+ return this.properties.put(key, value);
+ }
+
+ @Override
+ public String toString() {
+ return new ToStringCreator(this).append("name", name)
+ .append("properties", properties).toString();
+ }
+
+ public static Builder from(String name) {
+ return new Builder(name);
+ }
+
+ public static ByteBuf encode(Metadata metadata) {
+ return encode(ByteBufAllocator.DEFAULT, metadata);
+ }
+
+ public static ByteBuf encode(ByteBufAllocator allocator, Metadata metadata) {
+ return encode(allocator, metadata.getName(), metadata.getProperties());
+ }
+
+ public static ByteBuf encode(String name, Map properties) {
+ return encode(ByteBufAllocator.DEFAULT, name, properties);
+ }
+
+ public static ByteBuf encode(ByteBufAllocator allocator, String name,
+ Map properties) {
+ Assert.hasText(name, "name may not be empty");
+ Assert.notNull(properties, "properties may not be null");
+ Assert.notNull(allocator, "allocator may not be null");
+ ByteBuf byteBuf = allocator.buffer();
+
+ encodeString(byteBuf, name);
+
+ properties.entrySet().stream().forEach(entry -> {
+ encodeString(byteBuf, entry.getKey());
+ encodeString(byteBuf, entry.getValue());
+ });
+ return byteBuf;
+ }
+
+ private static void encodeString(ByteBuf byteBuf, String s) {
+ int length = NumberUtils.requireUnsignedByte(ByteBufUtil.utf8Bytes(s));
+ byteBuf.writeByte(length);
+ ByteBufUtil.reserveAndWriteUtf8(byteBuf, s, length);
+ }
+
+ public static Metadata decodeMetadata(ByteBuf byteBuf) {
+ AtomicInteger offset = new AtomicInteger(0);
+
+ String name = decodeString(byteBuf, offset);
+
+ Map properties = new LinkedHashMap<>();
+ while (offset.get() < byteBuf.readableBytes()) { // TODO: What is the best
+ // conditional here?
+ String key = decodeString(byteBuf, offset);
+ String value = null;
+ if (offset.get() < byteBuf.readableBytes()) {
+ value = decodeString(byteBuf, offset);
+ }
+ properties.put(key, value);
+ }
+
+ return new Metadata(name, properties);
+ }
+
+ private static String decodeString(ByteBuf byteBuf, AtomicInteger offset) {
+ int length = byteBuf.getByte(offset.get());
+ int index = offset.addAndGet(Byte.BYTES);
+ String s = byteBuf.toString(index, length, StandardCharsets.UTF_8);
+ offset.addAndGet(length);
+ return s;
+ }
+
+ public boolean matches(Metadata other) {
+ if (other == null) {
+ return false;
+ }
+ if (other.getName() == null) {
+ return false;
+ }
+ if (!getName().equalsIgnoreCase(other.getName())) {
+ return false;
+ }
+ return matches(getProperties(), other.getProperties());
+ }
+
+ /**
+ * Matches leftMetadata to rightMetadata. rightMetadata must contain all key with
+ * equal values (ignoring case) of leftMetadata.
+ * @param leftMetadata first metadata to compare.
+ * @param rightMetadata second metadata to compare.
+ * @return true if all keys and values (case-insensitive) from leftMetadata are in
+ * rightMetadata.
+ */
+ // TODO: find a way to make this more performant
+ public static boolean matches(Map leftMetadata,
+ Map rightMetadata) {
+ if (leftMetadata == null || rightMetadata == null) {
+ return false;
+ }
+
+ for (Map.Entry entry : leftMetadata.entrySet()) {
+ String enrichedValue = rightMetadata.get(entry.getKey());
+ if (enrichedValue == null ||
+ // TODO: regex and possibly SpEL?
+ !enrichedValue.equalsIgnoreCase(entry.getValue())) {
+ return false;
+ }
+ }
+
+ // all entries in metadata exist and match corresponding entries in
+ // enriched.metadata
+ return true;
+ }
+
+ public static class Builder {
+
+ private final Metadata metadata;
+
+ public Builder(String name) {
+ Assert.hasText(name, "Name must not be empty.");
+ this.metadata = new Metadata(name, new LinkedHashMap<>());
+ }
+
+ public Builder with(String key, String value) {
+ this.metadata.put(key, value);
+ return this;
+ }
+
+ public Metadata build() {
+ return this.metadata;
+ }
+
+ public ByteBuf encode() {
+ return Metadata.encode(build());
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/main/resources/META-INF/spring.factories b/spring-cloud-gateway-rsocket/src/main/resources/META-INF/spring.factories
new file mode 100644
index 00000000..4cf65ad1
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,3 @@
+# Auto Configure
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketAutoConfiguration
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
new file mode 100644
index 00000000..bcdefad8
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.autoconfigure;
+
+import org.junit.Test;
+
+import org.springframework.boot.actuate.autoconfigure.metrics.CompositeMeterRegistryAutoConfiguration;
+import org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
+import org.springframework.cloud.gateway.rsocket.registry.Registry;
+import org.springframework.cloud.gateway.rsocket.registry.RegistryRoutes;
+import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter;
+import org.springframework.cloud.gateway.rsocket.server.GatewayRSocketServer;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicate;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GatewayRSocketAutoConfigurationTests {
+
+ @Test
+ public void gatewayRSocketConfigured() {
+ new ReactiveWebApplicationContextRunner()
+ .withConfiguration(
+ AutoConfigurations.of(GatewayRSocketAutoConfiguration.class,
+ CompositeMeterRegistryAutoConfiguration.class,
+ MetricsAutoConfiguration.class))
+ .run(context -> assertThat(context).hasSingleBean(Registry.class)
+ .hasSingleBean(RegistryRoutes.class)
+ .hasSingleBean(RegistrySocketAcceptorFilter.class)
+ .hasSingleBean(GatewayRSocketServer.class)
+ .hasSingleBean(GatewayRSocketProperties.class)
+ .hasSingleBean(GatewaySocketAcceptor.class)
+ .hasSingleBean(SocketAcceptorPredicateFilter.class)
+ .doesNotHaveBean(SocketAcceptorPredicate.class));
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java
new file mode 100644
index 00000000..c4dfb6bf
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import reactor.test.StepVerifier;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.test.PingPongApp;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.SocketUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = PingPongApp.class, properties = {
+ "ping.take=5" }, webEnvironment = WebEnvironment.RANDOM_PORT)
+public class GatewayRSocketIntegrationTests {
+
+ private static int port;
+
+ @Autowired
+ private PingPongApp.Ping ping;
+
+ @Autowired
+ private PingPongApp.Pong pong;
+
+ @Autowired
+ private GatewayRSocketProperties properties;
+
+ @Autowired
+ private PingPongApp.MySocketAcceptorFilter mySocketAcceptorFilter;
+
+ @Autowired
+ private GatewayRSocketServer server;
+
+ @BeforeClass
+ public static void init() {
+ port = SocketUtils.findAvailableTcpPort();
+ System.setProperty("spring.cloud.gateway.rsocket.server.port",
+ String.valueOf(port));
+ }
+
+ @AfterClass
+ public static void after() {
+ System.clearProperty("spring.cloud.gateway.rsocket.server.port");
+ }
+
+ @Test
+ public void contextLoads() {
+ StepVerifier.create(ping.getPongFlux()).expectSubscription()
+ .then(() -> server.stop()).thenConsumeWhile(s -> true).verifyComplete();
+
+ assertThat(ping.getPongsReceived()).isGreaterThan(0);
+ assertThat(pong.getPingsReceived()).isGreaterThan(0);
+ assertThat(properties.getServer().getPort()).isNotEqualTo(7002);
+ assertThat(mySocketAcceptorFilter.invoked()).isTrue();
+ assertThat(server.isRunning()).isFalse();
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketTests.java
new file mode 100644
index 00000000..6e5e790f
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketTests.java
@@ -0,0 +1,278 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.server;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.netty.buffer.Unpooled;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.util.DefaultPayload;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
+import reactor.test.StepVerifier;
+
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket;
+import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket.EnrichedRSocket;
+import org.springframework.cloud.gateway.rsocket.registry.Registry;
+import org.springframework.cloud.gateway.rsocket.route.Route;
+import org.springframework.cloud.gateway.rsocket.route.Routes;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public class GatewayRSocketTests {
+
+ private static Log logger = LogFactory.getLog(GatewayRSocketTests.class);
+
+ private Registry registry;
+
+ private Payload incomingPayload;
+
+ // TODO: add tests for metrics and other request types
+
+ @Before
+ public void init() {
+ registry = mock(Registry.class);
+ incomingPayload = DefaultPayload.create(Unpooled.EMPTY_BUFFER,
+ Metadata.from("mock").with("id", "mock1").encode());
+
+ RSocket rSocket = mock(RSocket.class);
+ LoadBalancedRSocket loadBalancedRSocket = mock(LoadBalancedRSocket.class);
+ when(registry.getRegistered(any(Metadata.class))).thenReturn(loadBalancedRSocket);
+
+ Mono mono = Mono
+ .just(new EnrichedRSocket(rSocket, getMetadata()));
+ when(loadBalancedRSocket.choose()).thenReturn(mono);
+
+ when(rSocket.requestResponse(any(Payload.class)))
+ .thenReturn(Mono.just(DefaultPayload.create("response")));
+ }
+
+ @Test
+ public void multipleFilters() {
+ TestFilter filter1 = new TestFilter();
+ TestFilter filter2 = new TestFilter();
+ TestFilter filter3 = new TestFilter();
+
+ Payload payload = new TestGatewayRSocket(registry,
+ new TestRoutes(filter1, filter2, filter3))
+ .requestResponse(incomingPayload).block(Duration.ZERO);
+
+ assertThat(filter1.invoked()).isTrue();
+ assertThat(filter2.invoked()).isTrue();
+ assertThat(filter3.invoked()).isTrue();
+ assertThat(payload).isNotNull();
+ }
+
+ @Test
+ public void zeroFilters() {
+ Payload payload = new TestGatewayRSocket(registry, new TestRoutes())
+ .requestResponse(incomingPayload).block(Duration.ZERO);
+
+ assertThat(payload).isNotNull();
+ }
+
+ @Test
+ public void shortcircuitFilter() {
+
+ TestFilter filter1 = new TestFilter();
+ ShortcircuitingFilter filter2 = new ShortcircuitingFilter();
+ TestFilter filter3 = new TestFilter();
+
+ TestGatewayRSocket gatewayRSocket = new TestGatewayRSocket(registry,
+ new TestRoutes(filter1, filter2, filter3));
+ Mono response = gatewayRSocket.requestResponse(incomingPayload);
+
+ // a false filter will create a pending rsocket that blocks forever
+ // this tweaks the rsocket to compelte.
+ gatewayRSocket.processor.onNext(null);
+
+ StepVerifier.withVirtualTime(() -> response).expectSubscription()
+ .verifyComplete();
+
+ assertThat(filter1.invoked()).isTrue();
+ assertThat(filter2.invoked()).isTrue();
+ assertThat(filter3.invoked()).isFalse();
+ }
+
+ @Test
+ public void asyncFilter() {
+
+ AsyncFilter filter = new AsyncFilter();
+
+ Payload payload = new TestGatewayRSocket(registry, new TestRoutes(filter))
+ .requestResponse(incomingPayload).block(Duration.ofSeconds(5));
+
+ assertThat(filter.invoked()).isTrue();
+ assertThat(payload).isNotNull();
+ }
+
+ // TODO: add exception handlers?
+ @Test(expected = IllegalStateException.class)
+ public void handleErrorFromFilter() {
+
+ ExceptionFilter filter = new ExceptionFilter();
+
+ new TestGatewayRSocket(registry, new TestRoutes(filter))
+ .requestResponse(incomingPayload).block(Duration.ofSeconds(5));
+
+ // assertNull(socket);
+ }
+
+ private static Metadata getMetadata() {
+ return Metadata.from("service").with("id", "service1").build();
+ }
+
+ private static class TestGatewayRSocket extends GatewayRSocket {
+
+ private final MonoProcessor processor = MonoProcessor.create();
+
+ TestGatewayRSocket(Registry registry, Routes routes) {
+ super(registry, routes, new SimpleMeterRegistry(),
+ new GatewayRSocketProperties(), getMetadata());
+ }
+
+ @Override
+ PendingRequestRSocket constructPendingRSocket(GatewayExchange exchange) {
+ Function> routeFinder = registeredEvent -> getRouteMono(
+ registeredEvent, exchange);
+ return new PendingRequestRSocket(routeFinder, map -> {
+ Tags tags = exchange.getTags().and("responder.id", map.get("id"));
+ exchange.setTags(tags);
+ }, processor);
+ }
+
+ public MonoProcessor getProcessor() {
+ return processor;
+ }
+
+ }
+
+ private static class TestRoutes implements Routes {
+
+ private final Route route;
+
+ private List filters;
+
+ TestRoutes() {
+ this(Collections.emptyList());
+ }
+
+ TestRoutes(GatewayFilter... filters) {
+ this(Arrays.asList(filters));
+ }
+
+ TestRoutes(List filters) {
+ this.filters = filters;
+ route = Route.builder().id("route1")
+ .routingMetadata(Metadata.from("mock").build())
+ .predicate(exchange -> Mono.just(true)).filters(filters).build();
+ }
+
+ @Override
+ public Flux getRoutes() {
+ return Flux.just(route);
+ }
+
+ }
+
+ private static class TestFilter implements GatewayFilter {
+
+ private volatile boolean invoked;
+
+ public boolean invoked() {
+ return this.invoked;
+ }
+
+ @Override
+ public Mono filter(GatewayExchange exchange, GatewayFilterChain chain) {
+ this.invoked = true;
+ return doFilter(exchange, chain);
+ }
+
+ public Mono doFilter(GatewayExchange exchange,
+ GatewayFilterChain chain) {
+ return chain.filter(exchange);
+ }
+
+ }
+
+ private static class ShortcircuitingFilter extends TestFilter {
+
+ @Override
+ public Mono doFilter(GatewayExchange exchange,
+ GatewayFilterChain chain) {
+ return Mono.empty();
+ }
+
+ }
+
+ private static class AsyncFilter extends TestFilter {
+
+ @Override
+ public Mono doFilter(GatewayExchange exchange,
+ GatewayFilterChain chain) {
+ return doAsyncWork().flatMap(asyncResult -> {
+ logger.debug("Async result: " + asyncResult);
+ return chain.filter(exchange);
+ });
+ }
+
+ private Mono doAsyncWork() {
+ return Mono.delay(Duration.ofMillis(100L)).map(l -> "123");
+ }
+
+ }
+
+ private static class ExceptionFilter implements GatewayFilter {
+
+ @Override
+ public Mono filter(GatewayExchange exchange, GatewayFilterChain chain) {
+ return Mono.error(new IllegalStateException("boo"));
+ }
+
+ }
+
+ /*
+ * private static class TestExceptionHandler implements WebExceptionHandler {
+ *
+ * private Throwable ex;
+ *
+ * @Override public Mono handle(GatewayExchange exchange, Throwable ex) {
+ * this.ex = ex; return Mono.error(ex); } }
+ */
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java
new file mode 100644
index 00000000..c36fd30a
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java
@@ -0,0 +1,212 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.rsocket.ConnectionSetupPayload;
+import io.rsocket.RSocket;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Before;
+import org.junit.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
+import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rossen Stoyanchev
+ */
+public class GatewaySocketAcceptorTests {
+
+ private static Log logger = LogFactory.getLog(GatewaySocketAcceptorTests.class);
+
+ private GatewayRSocket.Factory factory;
+
+ private ConnectionSetupPayload setupPayload;
+
+ private RSocket sendingSocket;
+
+ private MeterRegistry meterRegistry;
+
+ private GatewayRSocketProperties properties = new GatewayRSocketProperties();
+
+ @Before
+ public void init() {
+ this.factory = mock(GatewayRSocket.Factory.class);
+ this.setupPayload = mock(ConnectionSetupPayload.class);
+ this.sendingSocket = mock(RSocket.class);
+ this.meterRegistry = new SimpleMeterRegistry();
+
+ when(this.factory.create(any(Metadata.class)))
+ .thenReturn(mock(GatewayRSocket.class));
+ }
+
+ // TODO: test metrics
+
+ @Test
+ public void multipleFilters() {
+ TestFilter filter1 = new TestFilter();
+ TestFilter filter2 = new TestFilter();
+ TestFilter filter3 = new TestFilter();
+
+ RSocket socket = new GatewaySocketAcceptor(this.factory,
+ Arrays.asList(filter1, filter2, filter3), this.meterRegistry,
+ this.properties).accept(this.setupPayload, this.sendingSocket)
+ .block(Duration.ZERO);
+
+ assertThat(filter1.invoked()).isTrue();
+ assertThat(filter2.invoked()).isTrue();
+ assertThat(filter3.invoked()).isTrue();
+ assertThat(socket).isNotNull();
+ }
+
+ @Test
+ public void zeroFilters() {
+ RSocket socket = new GatewaySocketAcceptor(this.factory, Collections.emptyList(),
+ this.meterRegistry, this.properties)
+ .accept(this.setupPayload, this.sendingSocket)
+ .block(Duration.ZERO);
+
+ assertThat(socket).isNotNull();
+ }
+
+ @Test
+ public void shortcircuitFilter() {
+
+ TestFilter filter1 = new TestFilter();
+ ShortcircuitingFilter filter2 = new ShortcircuitingFilter();
+ TestFilter filter3 = new TestFilter();
+
+ RSocket socket = new GatewaySocketAcceptor(this.factory,
+ Arrays.asList(filter1, filter2, filter3), this.meterRegistry,
+ this.properties).accept(this.setupPayload, this.sendingSocket)
+ .block(Duration.ZERO);
+
+ assertThat(filter1.invoked()).isTrue();
+ assertThat(filter2.invoked()).isTrue();
+ assertThat(filter3.invoked()).isFalse();
+ assertThat(socket).isNull();
+ }
+
+ @Test
+ public void asyncFilter() {
+
+ AsyncFilter filter = new AsyncFilter();
+
+ RSocket socket = new GatewaySocketAcceptor(this.factory, singletonList(filter),
+ this.meterRegistry, this.properties)
+ .accept(this.setupPayload, this.sendingSocket)
+ .block(Duration.ofSeconds(5));
+
+ assertThat(filter.invoked()).isTrue();
+ assertThat(socket).isNotNull();
+ }
+
+ // TODO: add exception handlers?
+ @Test(expected = IllegalStateException.class)
+ public void handleErrorFromFilter() {
+
+ ExceptionFilter filter = new ExceptionFilter();
+
+ new GatewaySocketAcceptor(this.factory, singletonList(filter), this.meterRegistry,
+ this.properties).accept(this.setupPayload, this.sendingSocket)
+ .block(Duration.ofSeconds(5));
+
+ }
+
+ private static class TestFilter implements SocketAcceptorFilter {
+
+ private volatile boolean invoked;
+
+ public boolean invoked() {
+ return this.invoked;
+ }
+
+ @Override
+ public Mono filter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ this.invoked = true;
+ return doFilter(exchange, chain);
+ }
+
+ public Mono doFilter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ return chain.filter(exchange);
+ }
+
+ }
+
+ private static class ShortcircuitingFilter extends TestFilter {
+
+ @Override
+ public Mono doFilter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ return Mono.empty();
+ }
+
+ }
+
+ private static class AsyncFilter extends TestFilter {
+
+ @Override
+ public Mono doFilter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ return doAsyncWork().flatMap(asyncResult -> {
+ logger.debug("Async result: " + asyncResult);
+ return chain.filter(exchange);
+ });
+ }
+
+ private Mono doAsyncWork() {
+ return Mono.delay(Duration.ofMillis(100L)).map(l -> "123");
+ }
+
+ }
+
+ private static class ExceptionFilter implements SocketAcceptorFilter {
+
+ @Override
+ public Mono filter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ return Mono.error(new IllegalStateException("boo"));
+ }
+
+ }
+
+ /*
+ * private static class TestExceptionHandler implements WebExceptionHandler {
+ *
+ * private Throwable ex;
+ *
+ * @Override public Mono handle(SocketAcceptorExchange exchange, Throwable ex) {
+ * this.ex = ex; return Mono.error(ex); } }
+ */
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java
new file mode 100644
index 00000000..dca6e359
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java
@@ -0,0 +1,131 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.socketacceptor;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import io.rsocket.ConnectionSetupPayload;
+import io.rsocket.RSocket;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+public class SocketAcceptorPredicateFilterTests {
+
+ @Test
+ public void noPredicateWorks() {
+ Mono result = runFilter(Collections.emptyList());
+ StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
+ }
+
+ @Test
+ public void singleTruePredicateWorks() {
+ TestPredicate predicate = new TestPredicate(true);
+ Mono result = runFilter(predicate);
+ StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
+ assertThat(predicate.invoked()).isTrue();
+ }
+
+ @Test
+ public void singleFalsePredicateWorks() {
+ TestPredicate predicate = new TestPredicate(false);
+ Mono result = runFilter(predicate);
+ StepVerifier.create(result).verifyComplete();
+
+ assertThat(predicate.invoked()).isTrue();
+ }
+
+ @Test
+ public void multipleFalsePredicateWorks() {
+ TestPredicate predicate = new TestPredicate(false);
+ TestPredicate predicate2 = new TestPredicate(false);
+ Mono result = runFilter(predicate, predicate2);
+ StepVerifier.create(result).verifyComplete();
+
+ assertThat(predicate.invoked()).isTrue();
+ assertThat(predicate2.invoked()).isTrue(); // Async predicates don't short circuit
+ }
+
+ @Test
+ public void multiplePredicatesNoSuccessWorks() {
+ TestPredicate truePredicate = new TestPredicate(true);
+ TestPredicate falsePredicate = new TestPredicate(false);
+ Mono result = runFilter(truePredicate, falsePredicate);
+ StepVerifier.create(result).verifyComplete();
+ assertThat(truePredicate.invoked()).isTrue();
+ assertThat(falsePredicate.invoked()).isTrue();
+ }
+
+ @Test
+ public void multiplePredicatesSuccessWorks() {
+ TestPredicate truePredicate = new TestPredicate(true);
+ TestPredicate truePredicate2 = new TestPredicate(true);
+ Mono result = runFilter(truePredicate, truePredicate2);
+ StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
+ assertThat(truePredicate.invoked()).isTrue();
+ assertThat(truePredicate2.invoked()).isTrue();
+ }
+
+ private Mono runFilter(SocketAcceptorPredicate predicate) {
+ return runFilter(Collections.singletonList(predicate));
+ }
+
+ private Mono runFilter(SocketAcceptorPredicate... predicates) {
+ return runFilter(Arrays.asList(predicates));
+ }
+
+ private Mono runFilter(List predicates) {
+ SocketAcceptorPredicateFilter filter = new SocketAcceptorPredicateFilter(
+ predicates);
+ SocketAcceptorExchange exchange = new SocketAcceptorExchange(
+ mock(ConnectionSetupPayload.class), mock(RSocket.class));
+ SocketAcceptorFilterChain filterChain = new SocketAcceptorFilterChain(
+ Collections.singletonList(filter));
+ return filter.filter(exchange, filterChain);
+ }
+
+ private class TestPredicate implements SocketAcceptorPredicate {
+
+ private boolean invoked = false;
+
+ private final Mono test;
+
+ TestPredicate(boolean value) {
+ test = Mono.just(value);
+ }
+
+ @Override
+ public Publisher apply(SocketAcceptorExchange exchange) {
+ invoked = true;
+ return test;
+ }
+
+ public boolean invoked() {
+ return invoked;
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/support/MetadataTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/support/MetadataTests.java
new file mode 100644
index 00000000..74138294
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/support/MetadataTests.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.support;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.IntStream;
+
+import io.netty.buffer.ByteBuf;
+import org.junit.Test;
+
+import org.springframework.util.Assert;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.cloud.gateway.rsocket.support.Metadata.matches;
+
+public class MetadataTests {
+
+ @Test
+ public void encodeAndDecodeJustName() {
+ ByteBuf byteBuf = Metadata.from("test").encode();
+ assertMetadata(byteBuf, "test");
+ }
+
+ @Test
+ public void encodeAndDecodeWorks() {
+ ByteBuf byteBuf = Metadata.from("test1").with("key1111", "val111111")
+ .with("key22", "val222").encode();
+ Metadata metadata = assertMetadata(byteBuf, "test1");
+ Map properties = metadata.getProperties();
+ assertThat(properties).hasSize(2).containsOnlyKeys("key1111", "key22")
+ .containsValues("val111111", "val222");
+ }
+
+ private Metadata assertMetadata(ByteBuf byteBuf, String name) {
+ Metadata metadata = Metadata.decodeMetadata(byteBuf);
+ assertThat(metadata).isNotNull();
+ assertThat(metadata.getName()).isEqualTo(name);
+ return metadata;
+ }
+
+ @Test
+ public void nullMetadataDoesNotMatch() {
+ assertThat(matches(null, new HashMap<>())).isFalse();
+
+ assertThat(matches(new HashMap<>(), null)).isFalse();
+ }
+
+ @Test
+ public void metadataSubsetMatches() {
+ assertThat(matches(metadata(2), metadata(3))).isTrue();
+ }
+
+ @Test
+ public void metadataEqualSetMatches() {
+ assertThat(matches(metadata(3), metadata(3))).isTrue();
+ }
+
+ @Test
+ public void metadataSuperSetDoesNotMatch() {
+ assertThat(matches(metadata(3), metadata(2))).isFalse();
+ }
+
+ private Map metadata(int size) {
+ Assert.isTrue(size > 0, "size must be > 0");
+ HashMap metadata = new HashMap<>();
+ IntStream.rangeClosed(1, size).forEach(i -> metadata.put("key" + i, "val" + i));
+ return metadata;
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
new file mode 100644
index 00000000..6abe9a7a
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
@@ -0,0 +1,293 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.test;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.RSocketFactory;
+import io.rsocket.micrometer.MicrometerRSocketInterceptor;
+import io.rsocket.transport.netty.client.TcpClientTransport;
+import io.rsocket.util.DefaultPayload;
+import io.rsocket.util.RSocketProxy;
+import lombok.extern.slf4j.Slf4j;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
+import org.springframework.cloud.gateway.rsocket.server.GatewayFilter;
+import org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorExchange;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilterChain;
+import org.springframework.cloud.gateway.rsocket.support.Metadata;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.Ordered;
+import org.springframework.core.env.ConfigurableEnvironment;
+
+import static io.netty.buffer.Unpooled.EMPTY_BUFFER;
+
+@SpringBootApplication
+public class PingPongApp {
+
+ @Bean
+ public Ping ping1() {
+ return new Ping("1");
+ }
+
+ @Bean
+ @ConditionalOnProperty("ping.two.enabled")
+ public Ping ping2() {
+ return new Ping("2");
+ }
+
+ @Bean
+ public Pong pong() {
+ return new Pong();
+ }
+
+ @Bean
+ public MySocketAcceptorFilter mySocketAcceptorFilter() {
+ return new MySocketAcceptorFilter();
+ }
+
+ public static void main(String[] args) {
+ SpringApplication.run(PingPongApp.class, args);
+ }
+
+ static String reply(String in) {
+ if (in.length() > 4) {
+ in = in.substring(0, 4);
+ }
+ switch (in.toLowerCase()) {
+ case "ping":
+ return "pong";
+ case "pong":
+ return "ping";
+ default:
+ throw new IllegalArgumentException("Value must be ping or pong, not " + in);
+ }
+ }
+
+ @Slf4j
+ public static class Ping
+ implements Ordered, ApplicationListener {
+
+ @Autowired
+ private MeterRegistry meterRegistry;
+
+ private final String id;
+
+ private final AtomicInteger pongsReceived = new AtomicInteger();
+
+ private Flux pongFlux;
+
+ public Ping(String id) {
+ this.id = id;
+ }
+
+ @Override
+ public int getOrder() {
+ return 0;
+ }
+
+ @Override
+ public void onApplicationEvent(ApplicationReadyEvent event) {
+ log.info("Starting Ping" + id);
+ ConfigurableEnvironment env = event.getApplicationContext().getEnvironment();
+ Integer take = env.getProperty("ping.take", Integer.class, null);
+ Integer gatewayPort = env.getProperty(
+ "spring.cloud.gateway.rsocket.server.port", Integer.class, 7002);
+
+ log.debug("ping.take: " + take);
+
+ MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
+ meterRegistry, Tag.of("component", "ping"));
+ ByteBuf announcementMetadata = Metadata.from("ping").with("id", "ping" + id)
+ .encode();
+ pongFlux = RSocketFactory.connect()
+ .metadataMimeType(Metadata.ROUTING_MIME_TYPE)
+ .setupPayload(
+ DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
+ .addClientPlugin(interceptor)
+ .transport(TcpClientTransport.create(gatewayPort)) // proxy
+ .start().flatMapMany(socket -> {
+ Flux pong = socket.requestChannel(
+ Flux.interval(Duration.ofSeconds(1)).map(i -> {
+ ByteBuf data = ByteBufUtil.writeUtf8(
+ ByteBufAllocator.DEFAULT, "ping" + id);
+ ByteBuf routingMetadata = Metadata.from("pong")
+ .encode();
+ return DefaultPayload.create(data, routingMetadata);
+ }).onBackpressureDrop(payload -> log.debug(
+ "Dropped payload " + payload.getDataUtf8())) // this
+ // is
+ // needed
+ // in
+ // case
+ // pong
+ // is
+ // not
+ // available
+ // yet
+ ).map(Payload::getDataUtf8).doOnNext(str -> {
+ int received = pongsReceived.incrementAndGet();
+ log.info("received " + str + "(" + received + ") in Ping"
+ + id);
+ }).doFinally(signal -> socket.dispose());
+ if (take != null) {
+ return pong.take(take);
+ }
+ return pong;
+ });
+
+ pongFlux.subscribe();
+ }
+
+ public Flux getPongFlux() {
+ return pongFlux;
+ }
+
+ public int getPongsReceived() {
+ return pongsReceived.get();
+ }
+
+ }
+
+ @Slf4j
+ public static class Pong
+ implements Ordered, ApplicationListener {
+
+ @Autowired
+ private MeterRegistry meterRegistry;
+
+ private final AtomicInteger pingsReceived = new AtomicInteger();
+
+ @Override
+ public int getOrder() {
+ return 1;
+ }
+
+ @Override
+ public void onApplicationEvent(ApplicationReadyEvent event) {
+ ConfigurableEnvironment env = event.getApplicationContext().getEnvironment();
+ Integer pongDelay = env.getProperty("pong.delay", Integer.class, 5000);
+ try {
+ Thread.sleep(pongDelay);
+ }
+ catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ log.info("Starting Pong");
+ Integer gatewayPort = env.getProperty(
+ "spring.cloud.gateway.rsocket.server.port", Integer.class, 7002);
+ MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
+ meterRegistry, Tag.of("component", "pong"));
+ ByteBuf announcementMetadata = Metadata.from("pong").with("id", "pong1")
+ .encode();
+ RSocketFactory.connect().metadataMimeType(Metadata.ROUTING_MIME_TYPE)
+ .setupPayload(
+ DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
+ .addClientPlugin(interceptor).acceptor(this::accept)
+ .transport(TcpClientTransport.create(gatewayPort)) // proxy
+ .start().block();
+ }
+
+ @SuppressWarnings("Duplicates")
+ RSocket accept(RSocket rSocket) {
+ RSocket pong = new RSocketProxy(rSocket) {
+
+ @Override
+ public Flux requestChannel(Publisher payloads) {
+ return Flux.from(payloads).map(Payload::getDataUtf8).doOnNext(str -> {
+ int received = pingsReceived.incrementAndGet();
+ log.info("received " + str + "(" + received + ") in Pong");
+ }).map(PingPongApp::reply).map(reply -> {
+ ByteBuf data = ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT,
+ reply);
+ ByteBuf routingMetadata = Metadata.from("ping").encode();
+ return DefaultPayload.create(data, routingMetadata);
+ });
+ }
+ };
+ return pong;
+ }
+
+ public int getPingsReceived() {
+ return pingsReceived.get();
+ }
+
+ }
+
+ @Slf4j
+ public static class MyGatewayFilter implements GatewayFilter {
+
+ private AtomicBoolean invoked = new AtomicBoolean(false);
+
+ @Override
+ public Mono filter(GatewayExchange exchange, GatewayFilterChain chain) {
+ log.info("in custom gateway filter");
+ invoked.compareAndSet(false, true);
+ return chain.filter(exchange);
+ }
+
+ public boolean invoked() {
+ return invoked.get();
+ }
+
+ }
+
+ @Slf4j
+ public static class MySocketAcceptorFilter implements SocketAcceptorFilter, Ordered {
+
+ private AtomicBoolean invoked = new AtomicBoolean(false);
+
+ @Override
+ public Mono filter(SocketAcceptorExchange exchange,
+ SocketAcceptorFilterChain chain) {
+ log.info("in custom socket acceptor filter");
+ invoked.compareAndSet(false, true);
+ return chain.filter(exchange);
+ }
+
+ @Override
+ public int getOrder() {
+ return 0;
+ }
+
+ public boolean invoked() {
+ return invoked.get();
+ }
+
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java
new file mode 100644
index 00000000..d30c937b
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2018-2019 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
+ *
+ * http://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.cloud.gateway.rsocket.test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.springframework.cloud.gateway.rsocket.registry.Registry;
+import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
+import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter;
+import org.springframework.core.OrderComparator;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+public class SocketAcceptorFilterOrderTests {
+
+ @Test
+ public void predicateFilterAfterRegistryFilter() {
+ SocketAcceptorFilter predicateFilter = new SocketAcceptorPredicateFilter(
+ Collections.emptyList());
+ SocketAcceptorFilter registryFilter = new RegistrySocketAcceptorFilter(
+ mock(Registry.class));
+ List filters = Arrays.asList(predicateFilter,
+ registryFilter);
+ OrderComparator.sort(filters);
+
+ assertThat(filters).containsExactly(registryFilter, predicateFilter);
+ }
+
+}
diff --git a/spring-cloud-gateway-rsocket/src/test/resources/application.yml b/spring-cloud-gateway-rsocket/src/test/resources/application.yml
new file mode 100644
index 00000000..e12210c9
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/resources/application.yml
@@ -0,0 +1,9 @@
+logging:
+ level:
+ org.springframework.cloud.gateway.rsocket: DEBUG
+
+management:
+ endpoints:
+ web:
+ exposure:
+ include: '*'
\ No newline at end of file
diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml
index 9adff450..00a2ac62 100644
--- a/src/checkstyle/checkstyle-suppressions.xml
+++ b/src/checkstyle/checkstyle-suppressions.xml
@@ -3,6 +3,7 @@
"-//Puppy Crawl//DTD Suppressions 1.1//EN"
"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
+