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..b47c602c
--- /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..4b37ffff
--- /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..0fdd3a30
--- /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..5bbffa77
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java
@@ -0,0 +1,69 @@
+/*
+ * 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.
+ */
+ 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..e6eeb703
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java
@@ -0,0 +1,48 @@
+/*
+ * 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..5c63b945
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java
@@ -0,0 +1,209 @@
+/*
+ * 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.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 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:
+ cancel.increment();
+ break;
+ case ON_COMPLETE:
+ onComplete.increment();
+ break;
+ case ON_ERROR:
+ onError.increment();
+ break;
+ }
+ }
+
+ private static Counter counter(
+ MeterRegistry meterRegistry, String interactionModel, SignalType signalType, Tag... tags) {
+
+ return meterRegistry.counter(
+ "rsocket." + interactionModel, Tags.of(tags).and("signal.type", signalType.name()));
+ }
+ }
+
+ 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:
+ sample.stop(cancel);
+ break;
+ case ON_COMPLETE:
+ sample.stop(onComplete);
+ break;
+ case ON_ERROR:
+ sample.stop(onError);
+ break;
+ }
+ }
+
+ Timer.Sample start() {
+ return Timer.start(meterRegistry);
+ }
+
+ private static Timer timer(
+ MeterRegistry meterRegistry, String interactionModel, SignalType signalType, Tag... tags) {
+
+ return meterRegistry.timer(
+ "rsocket." + interactionModel, Tags.of(tags).and("signal.type", signalType.name()));
+ }
+ }
+}
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..dd0f74aa
--- /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..42c6d3c5
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/LoadBalancedRSocket.java
@@ -0,0 +1,127 @@
+/*
+ * 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.Map;
+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..9c27ffd5
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/Registry.java
@@ -0,0 +1,104 @@
+/*
+ * 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..11120ad6
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistryRoutes.java
@@ -0,0 +1,91 @@
+/*
+ * 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..50b6ba6a
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/registry/RegistrySocketAcceptorFilter.java
@@ -0,0 +1,51 @@
+/*
+ * 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..f9108d66
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java
@@ -0,0 +1,198 @@
+/*
+ * 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 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);
+ }
+ }
+
+ 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();
+ }
+}
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..9279ad98
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java
@@ -0,0 +1,57 @@
+/*
+ * 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 org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Spencer Gibb
+ */
+public interface Routes {
+
+ 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..a5f09b64
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayExchange.java
@@ -0,0 +1,93 @@
+/*
+ * 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;
+
+public class GatewayExchange extends AbstractRSocketExchange {
+
+ private static final Log log = LogFactory.getLog(GatewayExchange.class);
+ 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..fc7090ee
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilter.java
@@ -0,0 +1,22 @@
+/*
+ * 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..8dfcd125
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayFilterChain.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.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..bc7b54c7
--- /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..08810a44
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocket.java
@@ -0,0 +1,321 @@
+/*
+ * 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 responderName = this.metadata.getName();
+ String responderId = this.metadata.get("id");
+ String requestorName = exchange.getRoutingMetadata().getName();
+ Assert.hasText(responderName, "responderName must not be empty");
+ Assert.hasText(responderId, "responderId must not be empty");
+ Assert.hasText(requestorName, "requestorName must not be empty");
+ //requestor.id happens in a callback, later
+ return Tags.of("requester.name", requestorName,
+ "responder.name", responderName, "responder.id", responderId,
+ "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
+ * @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("requester.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 -> {
+ return 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
+ * @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("requester.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..dc4435d7
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketServer.java
@@ -0,0 +1,145 @@
+/*
+ * 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..da568449
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/server/PendingRequestRSocket.java
@@ -0,0 +1,144 @@
+/*
+ * 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
+ */
+ @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
+ * @param 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..c3a11c13
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
@@ -0,0 +1,88 @@
+/*
+ * 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 sendingSocket 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.empty();
+ 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..4083323d
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java
@@ -0,0 +1,55 @@
+/*
+ * 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..2564a93e
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java
@@ -0,0 +1,22 @@
+/*
+ * 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..dd429450
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java
@@ -0,0 +1,44 @@
+/*
+ * 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.AbstractFilterChain;
+
+import java.util.List;
+
+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..5137f1b2
--- /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..7c81fb34
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.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.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..b38d2f2b
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java
@@ -0,0 +1,51 @@
+/*
+ * 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..e2ca8358
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/main/java/org/springframework/cloud/gateway/rsocket/support/Metadata.java
@@ -0,0 +1,194 @@
+/*
+ * 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 {
+
+ public static final String ROUTING_MIME_TYPE = "message/x.rsocket.routing.v0";
+
+ private final String 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..1b4fa204
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
@@ -0,0 +1,60 @@
+/*
+ * 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.metrics.MicrometerResponderRSocketInterceptor;
+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.GatewayRSocket;
+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..3e55d5cb
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketIntegrationTests.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.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..530caa53
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/server/GatewayRSocketTests.java
@@ -0,0 +1,285 @@
+/*
+ * 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.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+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);
+
+ assertTrue(filter1.invoked());
+ assertTrue(filter2.invoked());
+ assertTrue(filter3.invoked());
+ assertNotNull(payload);
+ }
+
+ @Test
+ public void zeroFilters() {
+ Payload payload = new TestGatewayRSocket(registry, new TestRoutes())
+ .requestResponse(incomingPayload)
+ .block(Duration.ZERO);
+
+ assertNotNull(payload);
+ }
+
+ @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();
+
+ assertTrue(filter1.invoked());
+ assertTrue(filter2.invoked());
+ assertFalse(filter3.invoked());
+ }
+
+ @Test
+ public void asyncFilter() {
+
+ AsyncFilter filter = new AsyncFilter();
+
+ Payload payload = new TestGatewayRSocket(registry, new TestRoutes(filter))
+ .requestResponse(incomingPayload)
+ .block(Duration.ofSeconds(5));
+
+ assertTrue(filter.invoked());
+ assertNotNull(payload);
+ }
+
+ //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 class TestGatewayRSocket extends GatewayRSocket {
+
+ private final MonoProcessor processor = MonoProcessor.create();
+
+ public 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("requester.id", map.get("id"));
+ exchange.setTags(tags);
+ }, processor);
+ }
+
+ public MonoProcessor getProcessor() {
+ return processor;
+ }
+ }
+
+ private static Metadata getMetadata() {
+ return Metadata.from("service")
+ .with("id", "service1")
+ .build();
+ }
+
+ private static class TestRoutes implements Routes {
+ private final Route route;
+ private List filters;
+
+ public TestRoutes() {
+ this(Collections.emptyList());
+ }
+
+ public TestRoutes(GatewayFilter... filters) {
+ this(Arrays.asList(filters));
+ }
+
+ public 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..6328d1fb
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java
@@ -0,0 +1,205 @@
+/*
+ * 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..3fd8f358
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java
@@ -0,0 +1,136 @@
+/*
+ * 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;
+
+ public 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..f329e79d
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/support/MetadataTests.java
@@ -0,0 +1,88 @@
+/*
+ * 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..e8380bf0
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
@@ -0,0 +1,279 @@
+/*
+ * 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..f9895fa6
--- /dev/null
+++ b/spring-cloud-gateway-rsocket/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java
@@ -0,0 +1,47 @@
+/*
+ * 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