routeIds = this.routingTable.findRouteIds(tagsMetadata);
- if (routeIds.contains(route.getId())) {
- return Mono.just(route);
- }
- return Mono.empty();
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java
deleted file mode 100644
index 024c6b92..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractFilterChain.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java
deleted file mode 100644
index 584cf4ff..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/AbstractRSocketExchange.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java
deleted file mode 100644
index 53fcafd5..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/FilterChain.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java
deleted file mode 100644
index bb30820e..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketExchange.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.filter;
-
-import java.util.Map;
-
-import org.springframework.lang.Nullable;
-import org.springframework.util.Assert;
-
-public interface RSocketExchange {
-
- /**
- * Return a mutable map of request attributes for the current exchange.
- * @return current attributes.
- */
- Map getAttributes();
-
- /**
- * Return the request attribute value if present.
- * @param name the attribute name
- * @param the attribute type
- * @return the attribute value
- */
- @SuppressWarnings("unchecked")
- @Nullable
- default T getAttribute(String name) {
- return (T) getAttributes().get(name);
- }
-
- /**
- * Return the request attribute value or if not present raise an
- * {@link IllegalArgumentException}.
- * @param name the attribute name
- * @param the attribute type
- * @return the attribute value
- */
- @SuppressWarnings("unchecked")
- default T getRequiredAttribute(String name) {
- T value = getAttribute(name);
- Assert.notNull(value, () -> "Required attribute '" + name + "' is missing");
- return value;
- }
-
- /**
- * Return the request attribute value, or a default, fallback value.
- * @param name the attribute name
- * @param defaultValue a default value to return instead
- * @param the attribute type
- * @return the attribute value
- */
- @SuppressWarnings("unchecked")
- default T getAttributeOrDefault(String name, T defaultValue) {
- return (T) getAttributes().getOrDefault(name, defaultValue);
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java
deleted file mode 100644
index 7833a9b4..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/filter/RSocketFilter.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java
deleted file mode 100644
index 3c7b981b..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocket.java
+++ /dev/null
@@ -1,249 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.metrics;
-
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
-
-import io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.Meter;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Tag;
-import io.micrometer.core.instrument.Tags;
-import io.micrometer.core.instrument.Timer;
-import io.rsocket.Payload;
-import io.rsocket.RSocket;
-import io.rsocket.ResponderRSocket;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.SignalType;
-
-import org.springframework.util.Assert;
-
-import static reactor.core.publisher.SignalType.CANCEL;
-import static reactor.core.publisher.SignalType.ON_COMPLETE;
-import static reactor.core.publisher.SignalType.ON_ERROR;
-
-public class MicrometerResponderRSocket implements ResponderRSocket {
-
- private static final Log log = LogFactory.getLog(MicrometerResponderRSocket.class);
-
- private final RSocket delegate;
-
- private final InteractionCounters metadataPush;
-
- private final InteractionCounters requestChannel;
-
- private final InteractionCounters requestFireAndForget;
-
- private final InteractionTimers requestResponse;
-
- private final InteractionCounters requestStream;
-
- /**
- * Creates a new {@link RSocket}.
- * @param delegate the {@link RSocket} to delegate to
- * @param meterRegistry the {@link MeterRegistry} to use
- * @param tags additional tags to attach to {@link Meter}s
- * @throws IllegalArgumentException if {@code delegate} or {@code meterRegistry} is
- * {@code null}
- */
- public MicrometerResponderRSocket(RSocket delegate, MeterRegistry meterRegistry,
- Tag... tags) {
- Assert.notNull(delegate, "delegate must not be null");
- Assert.notNull(meterRegistry, "meterRegistry must not be null");
-
- this.delegate = delegate;
- this.metadataPush = new InteractionCounters(meterRegistry, "metadata.push", tags);
- this.requestChannel = new InteractionCounters(meterRegistry, "request.channel",
- tags);
- this.requestFireAndForget = new InteractionCounters(meterRegistry, "request.fnf",
- tags);
- this.requestResponse = new InteractionTimers(meterRegistry, "request.response",
- tags);
- this.requestStream = new InteractionCounters(meterRegistry, "request.stream",
- tags);
- }
-
- @Override
- public void dispose() {
- delegate.dispose();
- }
-
- @Override
- public Mono fireAndForget(Payload payload) {
- return delegate.fireAndForget(payload).doFinally(requestFireAndForget);
- }
-
- @Override
- public Mono metadataPush(Payload payload) {
- return delegate.metadataPush(payload).doFinally(metadataPush);
- }
-
- @Override
- public Mono onClose() {
- return delegate.onClose();
- }
-
- @Override
- public Flux requestChannel(Publisher payloads) {
- return delegate.requestChannel(payloads).doFinally(requestChannel);
- }
-
- @Override
- public Mono requestResponse(Payload payload) {
- return Mono.defer(() -> {
- Timer.Sample sample = requestResponse.start();
-
- return delegate.requestResponse(payload)
- .doFinally(signalType -> requestResponse.accept(sample, signalType));
- });
- }
-
- @Override
- public Flux requestStream(Payload payload) {
- return delegate.requestStream(payload).doFinally(requestStream);
- }
-
- @Override
- public Flux requestChannel(Payload payload, Publisher payloads) {
- if (delegate instanceof ResponderRSocket) {
- ResponderRSocket rSocket = (ResponderRSocket) delegate;
- return rSocket.requestChannel(payload, payloads).doFinally(requestChannel);
- }
- return delegate.requestChannel(payloads).doFinally(requestChannel);
- }
-
- private static final class InteractionCounters implements Consumer {
-
- private final Counter cancel;
-
- private final Counter onComplete;
-
- private final Counter onError;
-
- private InteractionCounters(MeterRegistry meterRegistry, String interactionModel,
- Tag... tags) {
- this.cancel = counter(meterRegistry, interactionModel, CANCEL, tags);
- this.onComplete = counter(meterRegistry, interactionModel, ON_COMPLETE, tags);
- this.onError = counter(meterRegistry, interactionModel, ON_ERROR, tags);
- }
-
- @Override
- public void accept(SignalType signalType) {
- switch (signalType) {
- case CANCEL:
- if (this.cancel != null) {
- this.cancel.increment();
- }
- break;
- case ON_COMPLETE:
- if (this.onComplete != null) {
- this.onComplete.increment();
- }
- break;
- case ON_ERROR:
- if (this.onError != null) {
- this.onError.increment();
- }
- break;
- }
- }
-
- private Counter counter(MeterRegistry meterRegistry, String interactionModel,
- SignalType signalType, Tag... tags) {
-
- Tags withType = Tags.of(tags).and("signal.type", signalType.name());
- try {
- return meterRegistry.counter("rsocket." + interactionModel, withType);
- }
- catch (Exception e) {
- if (log.isTraceEnabled()) {
- log.trace("Error creating counter with tags: " + withType, e);
- }
- return null;
- }
- }
-
- }
-
- private static final class InteractionTimers
- implements BiConsumer {
-
- private final Timer cancel;
-
- private final MeterRegistry meterRegistry;
-
- private final Timer onComplete;
-
- private final Timer onError;
-
- private InteractionTimers(MeterRegistry meterRegistry, String interactionModel,
- Tag... tags) {
- this.meterRegistry = meterRegistry;
-
- this.cancel = timer(meterRegistry, interactionModel, CANCEL, tags);
- this.onComplete = timer(meterRegistry, interactionModel, ON_COMPLETE, tags);
- this.onError = timer(meterRegistry, interactionModel, ON_ERROR, tags);
- }
-
- @Override
- public void accept(Timer.Sample sample, SignalType signalType) {
- switch (signalType) {
- case CANCEL:
- if (this.cancel != null) {
- sample.stop(this.cancel);
- }
- break;
- case ON_COMPLETE:
- if (this.onComplete != null) {
- sample.stop(this.onComplete);
- }
- break;
- case ON_ERROR:
- if (this.onError != null) {
- sample.stop(this.onError);
- }
- break;
- }
- }
-
- Timer.Sample start() {
- return Timer.start(meterRegistry);
- }
-
- private static Timer timer(MeterRegistry meterRegistry, String interactionModel,
- SignalType signalType, Tag... tags) {
-
- Tags withType = Tags.of(tags).and("signal.type", signalType.name());
- try {
- return meterRegistry.timer("rsocket." + interactionModel, withType);
- }
- catch (Exception e) {
- if (log.isTraceEnabled()) {
- log.trace("Error creating timer with tags: " + withType, e);
- }
- return null;
- }
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java
deleted file mode 100644
index f2363211..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/metrics/MicrometerResponderRSocketInterceptor.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/DefaultRoute.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/DefaultRoute.java
deleted file mode 100644
index 33271ba9..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/DefaultRoute.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-import org.springframework.cloud.gateway.rsocket.core.GatewayFilter;
-import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
-import org.springframework.core.style.ToStringCreator;
-import org.springframework.util.Assert;
-
-/**
- * @author Spencer Gibb
- */
-public class DefaultRoute implements Route {
-
- private final String id;
-
- private final RouteSetup targetMetadata;
-
- private final int order;
-
- private final AsyncPredicate predicate;
-
- private final List gatewayFilters;
-
- public static Builder builder() {
- return new Builder();
- }
-
- private DefaultRoute(String id, RouteSetup targetMetadata, int order,
- AsyncPredicate predicate,
- List gatewayFilters) {
- this.id = id;
- this.targetMetadata = targetMetadata;
- this.order = order;
- this.predicate = predicate;
- this.gatewayFilters = gatewayFilters;
- }
-
- public String getId() {
- return this.id;
- }
-
- public 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.getId())
- && Objects.equals(order, route.getOrder())
- && Objects.equals(predicate, route.getPredicate())
- && Objects.equals(gatewayFilters, route.getFilters());
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(id, targetMetadata, predicate, gatewayFilters);
- }
-
- @Override
- public String toString() {
- return new ToStringCreator(this).append("id", id)
- .append("targetMetadata", targetMetadata).append("order", order)
- .append("predicate", predicate).append("gatewayFilters", gatewayFilters)
- .toString();
- }
-
- public static class Builder {
-
- protected String id;
-
- protected RouteSetup 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(RouteSetup 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 DefaultRoute(this.id, this.routingMetadata, this.order, predicate,
- this.gatewayFilters);
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java
deleted file mode 100644
index 404fff23..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Route.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.route;
-
-import java.util.List;
-
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-import org.springframework.cloud.gateway.rsocket.core.GatewayFilter;
-import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
-import org.springframework.core.Ordered;
-
-/**
- * @author Spencer Gibb
- */
-public interface Route extends Ordered {
-
- String getId();
-
- default int getOrder() {
- return 0;
- }
-
- AsyncPredicate getPredicate();
-
- List getFilters();
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java
deleted file mode 100644
index 47bdeb9e..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/route/Routes.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.route;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-
-/**
- * @author Spencer Gibb
- */
-public interface Routes {
-
- /** log. */
- Log log = LogFactory.getLog(Routes.class);
-
- Flux getRoutes();
-
- default Mono findRoute(GatewayExchange exchange) {
- return getRoutes()
- // individually filter routes so that filterWhen error delaying is not a
- // problem
- .concatMap(route -> Mono.just(route).filterWhen(r -> {
- // add the current route we are testing
- // TODO: exchange attributes
- // exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR,
- // r.getId());
- return r.getPredicate().apply(exchange);
- })
- // instead of immediately stopping main flux due to error, log and
- // swallow it
- .doOnError(e -> log.error(
- "Error applying predicate for route: " + route.getId(),
- e))
- .onErrorResume(e -> Mono.empty()))
- .next().map(route -> {
- if (log.isDebugEnabled()) {
- log.debug("Route matched: " + route.getId());
- }
- return route;
- });
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/LoadBalancerFactory.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/LoadBalancerFactory.java
deleted file mode 100644
index 2cc6f1d6..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/LoadBalancerFactory.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-import java.util.List;
-import java.util.Random;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Function;
-
-import io.rsocket.RSocket;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-
-public class LoadBalancerFactory {
-
- private static final Log log = LogFactory.getLog(LoadBalancerFactory.class);
-
- private final RoutingTable routingTable;
-
- public LoadBalancerFactory(RoutingTable routingTable) {
- this.routingTable = routingTable;
- }
-
- public List> find(TagsMetadata tagsMetadata) {
- List> rSockets = this.routingTable
- .findRSockets(tagsMetadata);
- return rSockets;
- }
-
- // TODO: potentially GatewayExchange or return a new Result Object?
- public Mono> choose(TagsMetadata tagsMetadata) {
- List> rSockets = this.routingTable
- .findRSockets(tagsMetadata);
- // TODO: change loadbalancer impl based on tags
- // TODO: cache loadbalancers based on tags
- return new RoundRobinLoadBalancer(tagsMetadata).apply(rSockets);
- }
-
- // TODO: Flux as input?
- // TODO: reuse commons load balancer?
- public interface LoadBalancer extends
- Function>, Mono>> {
-
- }
-
- public static class RoundRobinLoadBalancer implements LoadBalancer {
-
- private final TagsMetadata tagsMetadata;
-
- private final AtomicInteger position;
-
- public RoundRobinLoadBalancer(TagsMetadata tagsMetadata) {
- this(tagsMetadata, new Random().nextInt(1000));
- }
-
- public RoundRobinLoadBalancer(TagsMetadata tagsMetadata, int seedPosition) {
- this.tagsMetadata = tagsMetadata;
- this.position = new AtomicInteger(seedPosition);
- }
-
- @Override
- public Mono> apply(
- List> rSockets) {
- if (rSockets.isEmpty()) {
- if (log.isWarnEnabled()) {
- log.warn("No servers available for: " + this.tagsMetadata);
- }
- return Mono.empty();
- }
- // TODO: enforce order?
- int pos = Math.abs(this.position.incrementAndGet());
-
- Tuple2 tuple = rSockets.get(pos % rSockets.size());
- return Mono.just(tuple);
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTable.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTable.java
deleted file mode 100644
index cbcaca67..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTable.java
+++ /dev/null
@@ -1,333 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Consumer;
-
-import io.rsocket.RSocket;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.roaringbitmap.IntConsumer;
-import org.roaringbitmap.RoaringBitmap;
-import reactor.core.Disposable;
-import reactor.core.publisher.DirectProcessor;
-import reactor.core.publisher.FluxSink;
-import reactor.util.function.Tuple2;
-import reactor.util.function.Tuples;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.WellKnownKey;
-import org.springframework.core.style.ToStringCreator;
-import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
-
-/**
- * The RoutingTable handles all RSocket connections that have been made that have
- * associated RouteSetup metadata. RSocket connections can then be found based on
- * Forwarding metadata. When a new RSocket is registered, a RegisteredEvent is pushed onto
- * a DirectProcessor that is acting as an event bus for registered Consumers.
- */
-public class RoutingTable {
-
- private static final Log log = LogFactory.getLog(RoutingTable.class);
-
- AtomicInteger internalRouteId = new AtomicInteger();
-
- final Map internalRouteIdToRouteId = new ConcurrentHashMap<>();
-
- final Map tagsToBitmaps = new ConcurrentHashMap<>();
-
- final Map routeEntries = new ConcurrentHashMap<>();
-
- private final DirectProcessor registeredEvents = DirectProcessor
- .create();
-
- private final FluxSink registeredEventsSink = registeredEvents
- .sink(FluxSink.OverflowStrategy.DROP);
-
- public RoutingTable() {
- }
-
- // TODO: Mono?
- public void register(TagsMetadata tagsMetadata, RSocket rsocket) {
- register(new RouteEntry(rsocket, tagsMetadata));
- }
-
- private void register(RouteEntry routeEntry) {
- if (log.isInfoEnabled()) {
- log.info("Registering RSocket: " + routeEntry.tagsMetadata);
- }
-
- // TODO: only register new route if timestamp is newer
- String routeId = routeEntry.getRouteId();
-
- if (routeEntries.containsKey(routeId)) {
- throw new IllegalStateException("Route Id already registered: " + routeId);
- }
-
- int internalId = internalRouteId.incrementAndGet();
- internalRouteIdToRouteId.put(internalId, routeId);
- routeEntries.put(routeId, routeEntry);
-
- routeEntry.getTags().forEach((key, value) -> {
- // TODO: deal with string keys?
- RoaringBitmap bitmap = tagsToBitmaps.computeIfAbsent(new TagKey(key, value),
- k -> new RoaringBitmap());
- bitmap.add(internalId);
- });
-
- registeredEventsSink.next(new RegisteredEvent(routeEntry));
- }
-
- public boolean deregister(TagsMetadata metadata) {
- Assert.notNull(metadata, "metadata may not be null");
- String routeId = metadata.getRouteId();
- if (!StringUtils.hasText(routeId)) {
- if (log.isDebugEnabled()) {
- log.debug("Unable to deregister, no RouteId: " + metadata);
- }
- return false;
- }
- if (log.isInfoEnabled()) {
- log.info("Deregistering RSocket: " + metadata);
- }
-
- TagsMetadata findByRouteId = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, routeId).build();
- RoaringBitmap found = find(findByRouteId);
-
- if (found.isEmpty() || found.getLongCardinality() > 1) {
- if (log.isWarnEnabled()) {
- log.warn("Unable to deregister " + metadata + ", found: "
- + found.getLongCardinality());
- }
- return false;
- }
-
- int internalId = found.first();
- internalRouteIdToRouteId.remove(internalId);
- routeEntries.remove(routeId);
-
- metadata.getTags().forEach((key, value) -> {
- // TODO: deal with string keys?
- TagKey tagKey = new TagKey(key, value);
- if (tagsToBitmaps.containsKey(tagKey)) {
- RoaringBitmap bitmap = tagsToBitmaps.get(tagKey);
- bitmap.remove(internalId);
- }
- });
-
- // TODO: deregistered event
- return true;
- }
-
- /**
- * Finds routeIds of matching routes.
- * @param tagsMetadata tags to match.
- * @return all matching routeIds or empty list.
- */
- public Set findRouteIds(TagsMetadata tagsMetadata) {
- RoaringBitmap found = find(tagsMetadata);
- if (found.isEmpty()) {
- return Collections.emptySet();
- }
- HashSet routeIds = new HashSet<>();
- found.forEach((IntConsumer) internalId -> {
- String routeId = internalRouteIdToRouteId.get(internalId);
- routeIds.add(routeId);
- });
- return routeIds;
- }
-
- /**
- * Finds tuples of routeIds and RSockets of matching routes.
- * @param tagsMetadata tags to match.
- * @return all matching routeId and RSocket tuples or empty list.
- */
- public List> findRSockets(TagsMetadata tagsMetadata) {
- RoaringBitmap found = find(tagsMetadata);
- if (found.isEmpty()) {
- return Collections.emptyList();
- }
- ArrayList> rSockets = new ArrayList<>();
- found.forEach((IntConsumer) internalId -> {
- String routeId = internalRouteIdToRouteId.get(internalId);
- RouteEntry routeEntry = routeEntries.get(routeId);
- RSocket rSocket = routeEntry.getRSocket();
- rSockets.add(Tuples.of(routeId, rSocket));
- });
- return rSockets;
- }
-
- /**
- * Finds internal ids of routes.
- * @param tagsMetadata tags to match
- * @return bitmap of all internal ids of routes.
- */
- RoaringBitmap find(TagsMetadata tagsMetadata) {
- RoaringBitmap found = new RoaringBitmap();
- AtomicBoolean first = new AtomicBoolean(true);
- tagsMetadata.getTags().forEach((key, value) -> {
- TagKey tagKey = new TagKey(key, value);
- if (tagsToBitmaps.containsKey(tagKey)) {
- RoaringBitmap search = tagsToBitmaps.get(tagKey);
- if (first.get()) {
- // initiliaze found bitmap with current search
- found.or(search);
- first.compareAndSet(true, false);
- }
- else {
- found.and(search);
- }
- }
- });
- return found;
- }
-
- public Disposable addListener(Consumer consumer) {
- return this.registeredEvents.subscribe(consumer);
- }
-
- public static class RegisteredEvent {
-
- private final RouteEntry routeEntry;
-
- public RegisteredEvent(RouteEntry routeEntry) {
- Assert.notNull(routeEntry, "routeEntry may not be null");
- this.routeEntry = routeEntry;
- }
-
- public TagsMetadata getRoutingMetadata() {
- return this.routeEntry.getTagsMetadata();
- }
-
- public RSocket getRSocket() {
- return this.routeEntry.getRSocket();
- }
-
- }
-
- static class RouteEntry {
-
- private final RSocket rSocket;
-
- private final TagsMetadata tagsMetadata;
-
- private final Long timestamp;
-
- RouteEntry(RSocket rSocket, TagsMetadata tagsMetadata) {
- this(rSocket, tagsMetadata, System.currentTimeMillis());
- }
-
- RouteEntry(RSocket rSocket, TagsMetadata tagsMetadata, Long timestamp) {
- Assert.notNull(tagsMetadata, "tagsMetadata may not be null");
- Assert.notNull(rSocket, "RSocket may not be null");
- this.rSocket = rSocket;
- this.tagsMetadata = tagsMetadata;
- this.timestamp = timestamp;
- }
-
- public RSocket getRSocket() {
- return this.rSocket;
- }
-
- public TagsMetadata getTagsMetadata() {
- return this.tagsMetadata;
- }
-
- public Long getTimestamp() {
- return this.timestamp;
- }
-
- public String getRouteId() {
- return this.tagsMetadata.getRouteId();
- }
-
- public Map getTags() {
- return this.getTagsMetadata().getTags();
- }
-
- @Override
- public String toString() {
- // @formatter:off
- return new ToStringCreator(this)
- .append("rSocket", rSocket)
- .append("tagsMetadata", tagsMetadata)
- .toString();
- // @formatter:on
- }
-
- }
-
- static class TagKey {
-
- final TagsMetadata.Key key;
-
- final String value;
-
- TagKey(TagsMetadata.Key key, String value) {
- // TODO: Assert non null
- this.key = key;
- this.value = value.toLowerCase();
- }
-
- public TagsMetadata.Key getKey() {
- return this.key;
- }
-
- public String getValue() {
- return this.value;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- TagKey that = (TagKey) o;
- return Objects.equals(this.key, that.key)
- && Objects.equals(this.value, that.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(this.key, this.value);
- }
-
- @Override
- public String toString() {
- return new ToStringCreator(this).append("key", key).append("value", value)
- .toString();
-
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutes.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutes.java
deleted file mode 100644
index bbc09c68..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutes.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Consumer;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-import org.springframework.cloud.gateway.rsocket.core.GatewayFilter;
-import org.springframework.cloud.gateway.rsocket.route.Route;
-import org.springframework.cloud.gateway.rsocket.route.Routes;
-import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
-import org.springframework.core.style.ToStringCreator;
-
-/**
- * View of RoutingTable as Route objects.
- */
-public class RoutingTableRoutes
- implements Routes, Consumer {
-
- private static final Log log = LogFactory.getLog(RoutingTableRoutes.class);
-
- private Map routes = new ConcurrentHashMap<>();
-
- private final RoutingTable routingTable;
-
- public RoutingTableRoutes(RoutingTable routingTable) {
- this.routingTable = routingTable;
- this.routingTable.addListener(this);
- }
-
- @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(RoutingTable.RegisteredEvent registeredEvent) {
- TagsMetadata routingMetadata = registeredEvent.getRoutingMetadata();
- String routeId = routingMetadata.getRouteId();
-
- routes.computeIfAbsent(routeId, key -> createRoute(routeId));
- }
-
- private Route createRoute(String routeId) {
- AsyncPredicate predicate = new RoutIdPredicate(routingTable,
- routeId);
-
- RegistryRoute route = new RegistryRoute(routeId, predicate);
-
- if (log.isDebugEnabled()) {
- log.debug("Created Route for registered service " + route);
- }
-
- return route;
- }
-
- static class RoutIdPredicate implements AsyncPredicate {
-
- private final RoutingTable routingTable;
-
- private final String routeId;
-
- RoutIdPredicate(RoutingTable routingTable, String routeId) {
- this.routingTable = routingTable;
- this.routeId = routeId;
- }
-
- @Override
- public Publisher apply(GatewayExchange exchange) {
- // TODO: standard predicates
- // TODO: allow customized predicates
- Set routeIds = routingTable
- .findRouteIds(exchange.getRoutingMetadata());
- return Mono.just(routeIds.contains(routeId));
- }
-
- @Override
- public String toString() {
- return String.format("[RoutIdPredicate %s]", routeId);
- }
-
- }
-
- static class RegistryRoute implements Route {
-
- final String id;
-
- final AsyncPredicate predicate;
-
- RegistryRoute(String id, AsyncPredicate predicate) {
- this.id = id;
- this.predicate = predicate;
- }
-
- @Override
- public String getId() {
- return this.id;
- }
-
- @Override
- public AsyncPredicate getPredicate() {
- return this.predicate;
- }
-
- @Override
- public List getFilters() {
- return Collections.emptyList();
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- RegistryRoute that = (RegistryRoute) o;
- return Objects.equals(this.id, that.id)
- && Objects.equals(this.predicate, that.predicate);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(this.id, this.predicate);
- }
-
- @Override
- public String toString() {
- return new ToStringCreator(this).append("id", id)
- .append("predicate", predicate).toString();
-
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableSocketAcceptorFilter.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableSocketAcceptorFilter.java
deleted file mode 100644
index ddc968a3..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableSocketAcceptorFilter.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-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;
-
-/**
- * Filter that registers the SendingSocket.
- */
-public class RoutingTableSocketAcceptorFilter implements SocketAcceptorFilter, Ordered {
-
- private final RoutingTable routingTable;
-
- public RoutingTableSocketAcceptorFilter(RoutingTable routingTable) {
- this.routingTable = routingTable;
- }
-
- @Override
- public Mono filter(SocketAcceptorExchange exchange,
- SocketAcceptorFilterChain chain) {
- if (exchange.getMetadata() != null) {
- // TODO: needed? &&
- // StringUtils.hasLength(exchange.getMetadata().getServiceName())) {
- this.routingTable.register(exchange.getMetadata().getEnrichedTagsMetadata(),
- exchange.getSendingSocket());
- }
-
- return chain.filter(exchange);
- }
-
- @Override
- public int getOrder() {
- return HIGHEST_PRECEDENCE + 1000;
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
deleted file mode 100644
index 984fa7ae..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptor.java
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import java.util.List;
-import java.util.Map;
-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 org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Mono;
-
-import org.springframework.cloud.gateway.rsocket.autoconfigure.BrokerProperties;
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.core.GatewayRSocketFactory;
-import org.springframework.cloud.gateway.rsocket.metrics.MicrometerResponderRSocket;
-import org.springframework.messaging.rsocket.MetadataExtractor;
-import org.springframework.util.MimeType;
-
-public class GatewaySocketAcceptor implements SocketAcceptor {
-
- private static final Log log = LogFactory.getLog(GatewaySocketAcceptor.class);
-
- private final SocketAcceptorFilterChain filterChain;
-
- private final GatewayRSocketFactory rSocketFactory;
-
- private final MeterRegistry meterRegistry;
-
- private final BrokerProperties properties;
-
- private final MetadataExtractor metadataExtractor;
-
- public GatewaySocketAcceptor(GatewayRSocketFactory rSocketFactory,
- List filters, MeterRegistry meterRegistry,
- BrokerProperties properties, MetadataExtractor metadataExtractor) {
- this.rSocketFactory = rSocketFactory;
- this.filterChain = new SocketAcceptorFilterChain(filters);
- this.meterRegistry = meterRegistry;
- this.properties = properties;
- this.metadataExtractor = metadataExtractor;
- }
-
- @Override
- @SuppressWarnings("Duplicates")
- public Mono accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
- if (log.isTraceEnabled()) {
- log.trace("accept()");
- }
- // decorate GatewayRSocket with metrics
- // current gateway id, type requester, service name (from metadata), service id
-
- Tags requesterTags = Tags.of("gateway.id", properties.getId(), "type",
- "requester");
-
- Tags metadataTags;
- SocketAcceptorExchange exchange;
-
- Map metadataMap = null;
- try {
- metadataMap = this.metadataExtractor.extract(setup,
- MimeType.valueOf(setup.metadataMimeType()));
- }
- catch (Exception e) {
- if (log.isDebugEnabled()) {
- log.debug("Error extracting metadata", e);
- }
- return Mono.error(e);
- }
- if (metadataMap.containsKey("routesetup")) {
- RouteSetup metadata = (RouteSetup) metadataMap.get("routesetup");
- metadataTags = Tags.of("service.name", metadata.getServiceName())
- .and("service.id", metadata.getId().toString());
- // enrich exchange to have metadata
- exchange = new SocketAcceptorExchange(setup,
- decorate(sendingSocket, requesterTags.and(metadataTags)), metadata);
- }
- else {
- metadataTags = Tags.of("service.name", "UNKNOWN").and("service.id",
- "UNKNOWN");
- exchange = new SocketAcceptorExchange(setup,
- decorate(sendingSocket, requesterTags));
- }
-
- Tags responderTags = Tags
- .of("gateway.id", properties.getId(), "type", "responder")
- .and(metadataTags);
-
- // decorate with metrics gateway id, type responder, service name, service id
- // (instance id)
- return this.filterChain.filter(exchange).log(
- GatewaySocketAcceptor.class.getName() + ".socket acceptor filter chain",
- Level.FINEST).map(success -> {
- TagsMetadata tags = exchange.getMetadata().getEnrichedTagsMetadata();
- return decorate(this.rSocketFactory.create(tags), 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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java
deleted file mode 100644
index 0c0f5625..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorExchange.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import java.math.BigInteger;
-
-import io.rsocket.ConnectionSetupPayload;
-import io.rsocket.RSocket;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.filter.AbstractRSocketExchange;
-
-public class SocketAcceptorExchange extends AbstractRSocketExchange {
-
- private final ConnectionSetupPayload setup;
-
- private final RSocket sendingSocket;
-
- private final RouteSetup metadata;
-
- public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket) {
- this(setup, sendingSocket, RouteSetup.of((BigInteger) null, null).build());
- }
-
- public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket,
- RouteSetup metadata) {
- this.setup = setup;
- this.sendingSocket = sendingSocket;
- this.metadata = metadata;
- }
-
- public ConnectionSetupPayload getSetup() {
- return setup;
- }
-
- public RSocket getSendingSocket() {
- return sendingSocket;
- }
-
- public RouteSetup getMetadata() {
- return metadata;
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java
deleted file mode 100644
index 31923172..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilter.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter;
-
-public interface SocketAcceptorFilter
- extends RSocketFilter {
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java
deleted file mode 100644
index 470eee62..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorFilterChain.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import java.util.List;
-
-import org.springframework.cloud.gateway.rsocket.filter.AbstractFilterChain;
-
-public class SocketAcceptorFilterChain extends
- AbstractFilterChain {
-
- /**
- * Public constructor with the list of filters and the target handler to use.
- * @param filters the filters ahead of the handler
- */
- public SocketAcceptorFilterChain(List filters) {
- super(filters);
- }
-
- public SocketAcceptorFilterChain(List allFilters,
- SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
- super(allFilters, currentFilter, next);
- }
-
- @Override
- protected SocketAcceptorFilterChain create(List allFilters,
- SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
- return new SocketAcceptorFilterChain(allFilters, currentFilter, next);
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java
deleted file mode 100644
index af37eddd..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicate.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
-
-public interface SocketAcceptorPredicate extends AsyncPredicate {
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java
deleted file mode 100644
index b9b2df43..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilter.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java
deleted file mode 100644
index 7fdad9ca..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/java/org/springframework/cloud/gateway/rsocket/support/AsyncPredicate.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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/spring-cloud-gateway-rsocket-broker/src/main/resources/META-INF/spring.factories b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/resources/META-INF/spring.factories
deleted file mode 100644
index c7338a85..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/main/resources/META-INF/spring.factories
+++ /dev/null
@@ -1,7 +0,0 @@
-# Auto Configure
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketAutoConfiguration
-
-# Environment Post Processors
-org.springframework.boot.env.EnvironmentPostProcessor=\
-org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketEnvironmentPostProcessor
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorIntegrationTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorIntegrationTests.java
deleted file mode 100644
index 64e1e7a8..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorIntegrationTests.java
+++ /dev/null
@@ -1,194 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.actuate;
-
-import java.math.BigInteger;
-import java.util.Random;
-
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import reactor.core.publisher.Hooks;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.SpringBootConfiguration;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.boot.test.mock.mockito.MockBean;
-import org.springframework.cloud.gateway.rsocket.cluster.ClusterService;
-import org.springframework.cloud.gateway.rsocket.common.metadata.Forwarding;
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.messaging.rsocket.RSocketRequester;
-import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.util.SocketUtils;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
-import static org.springframework.cloud.gateway.rsocket.actuate.BrokerActuator.BROKER_INFO_PATH;
-import static org.springframework.cloud.gateway.rsocket.actuate.BrokerActuator.ROUTE_JOIN_PATH;
-import static org.springframework.cloud.gateway.rsocket.actuate.BrokerActuator.ROUTE_REMOVE_PATH;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = RANDOM_PORT,
- properties = { "spring.cloud.gateway.rsocket.cluster.enabled=false",
- "spring.cloud.gateway.rsocket.broker.actuator.enabled=true" })
-public class BrokerActuatorIntegrationTests {
-
- private final Random random = new Random();
-
- @Autowired
- private RSocketRequester.Builder requesterBuilder;
-
- @Autowired
- private RSocketMessageHandler messageHandler;
-
- @MockBean
- private ClusterService clusterService;
-
- // @LocalServerPort
- private static int port;
-
- @BeforeClass
- public static void init() {
- Hooks.onOperatorDebug();
- port = SocketUtils.findAvailableTcpPort();
- System.setProperty("spring.rsocket.server.port", String.valueOf(port));
- }
-
- @AfterClass
- public static void after() {
- System.clearProperty("spring.rsocket.server.port");
- }
-
- @Test
- public void brokerInfoWorks() {
- long brokerId = random.nextLong();
-
- BrokerInfo data = BrokerInfo.of(brokerId).build();
-
- Mono result = callActuator(brokerId, BigInteger.class, data,
- BROKER_INFO_PATH);
-
- StepVerifier.create(result).consumeNextWith(
- res -> assertThat(res).isNotNull().isEqualTo(BigInteger.valueOf(1234L)))
- .verifyComplete();
-
- // TODO: assert server side calls worked
- }
-
- @Test
- @Ignore // TODO: move to integration tests module
- public void routeJoinRemoveWorks() {
- long brokerId = random.nextLong();
- long routeId = random.nextLong();
-
- RouteJoin data = RouteJoin.builder().brokerId(brokerId).routeId(routeId)
- .serviceName("testServiceName").build();
-
- RSocketRequester requester = getRequester(brokerId);
- Mono result = callActuator(requester, brokerId, RouteJoin.class, data,
- ROUTE_JOIN_PATH);
-
- StepVerifier.create(result)
- .consumeNextWith(res -> assertThat(res).isNotNull().isEqualTo(data))
- .verifyComplete();
- // TODO: assert server side calls worked
-
- routeRemoveWorks(requester, routeId);
- }
-
- public void routeRemoveWorks(RSocketRequester requester, long routeId) {
- long brokerId = random.nextLong();
- RouteRemove data = RouteRemove.builder().brokerId(brokerId).routeId(routeId)
- .build();
-
- Mono result = callActuator(requester, brokerId, Boolean.class, data,
- ROUTE_REMOVE_PATH);
-
- StepVerifier.create(result).consumeNextWith(res -> assertThat(res).isTrue())
- .verifyComplete();
- // TODO: assert server side calls worked
-
- result = callActuator(brokerId, Boolean.class, data, ROUTE_REMOVE_PATH);
-
- StepVerifier.create(result).consumeNextWith(res -> assertThat(res).isTrue())
- .verifyComplete();
- }
-
- @Test
- @Ignore // TODO: move to integration tests module
- public void routeJoinCloseDeregisters() {
- long brokerId = random.nextLong();
- long routeId = random.nextLong();
-
- RouteJoin data = RouteJoin.builder().brokerId(brokerId).routeId(routeId)
- .serviceName("testServiceName").build();
-
- RSocketRequester requester = getRequester(brokerId);
- Mono result = callActuator(requester, brokerId, RouteJoin.class, data,
- ROUTE_JOIN_PATH);
-
- result.block();
- StepVerifier.create(result)
- .consumeNextWith(res -> assertThat(res).isNotNull().isEqualTo(data))
- .verifyComplete();
-
- requester.rsocket().dispose();
-
- // TODO: assert server side calls worked
- }
-
- private Mono callActuator(long brokerId, Class type, D data,
- String path) {
- RSocketRequester requester = getRequester(brokerId);
-
- return callActuator(requester, brokerId, type, data, path);
- }
-
- private Mono callActuator(RSocketRequester requester, long brokerId,
- Class type, D data, String path) {
-
- Forwarding forwarding = Forwarding.of(brokerId).serviceName("gateway")
- .disableProxy().build();
-
- return requester.route(path).metadata(forwarding, Forwarding.FORWARDING_MIME_TYPE)
- .data(data).retrieveMono(type);
- }
-
- private RSocketRequester getRequester(long brokerId) {
- RouteSetup routeSetup = RouteSetup.of(brokerId, "gateway")
- .with("proxy", Boolean.FALSE.toString()).build();
- // mimic rsocket client autoconfig
- return requesterBuilder
- .setupMetadata(routeSetup, RouteSetup.ROUTE_SETUP_MIME_TYPE)
- .rsocketFactory(rsocketFactory -> rsocketFactory
- .acceptor(messageHandler.responder()))
- .connectTcp("localhost", port).block();
- }
-
- @SpringBootConfiguration
- @EnableAutoConfiguration
- static class Config {
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorRegistrarTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorRegistrarTests.java
deleted file mode 100644
index 0a958b8b..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/actuate/BrokerActuatorRegistrarTests.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.actuate;
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.SpringBootConfiguration;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.junit4.SpringRunner;
-
-import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = RANDOM_PORT,
- properties = { "spring.rsocket.server.port=0",
- "spring.cloud.gateway.rsocket.route-id=55",
- "spring.cloud.gateway.rsocket.service-name=gateway" })
-public class BrokerActuatorRegistrarTests {
-
- @Autowired
- private BrokerActuatorHandlerRegistration registrar;
-
- @Test
- public void test() {
- }
-
- @SpringBootConfiguration
- @EnableAutoConfiguration
- static class Config {
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
deleted file mode 100644
index cae682d5..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/autoconfigure/GatewayRSocketAutoConfigurationTests.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.autoconfigure;
-
-import io.rsocket.SocketAcceptor;
-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.autoconfigure.rsocket.RSocketMessagingAutoConfiguration;
-import org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration;
-import org.springframework.boot.rsocket.context.RSocketServerBootstrap;
-import org.springframework.boot.rsocket.server.RSocketServer;
-import org.springframework.boot.rsocket.server.RSocketServerFactory;
-import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
-import org.springframework.cloud.gateway.rsocket.common.autoconfigure.GatewayRSocketCommonAutoConfiguration;
-import org.springframework.cloud.gateway.rsocket.core.GatewayServerRSocketFactoryProcessor;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTable;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTableRoutes;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTableSocketAcceptorFilter;
-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 org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-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;
-
-public class GatewayRSocketAutoConfigurationTests {
-
- @Test
- public void gatewayRSocketConfigured() {
- new ReactiveWebApplicationContextRunner().withUserConfiguration(MyConfig.class)
- .withSystemProperties("spring.cloud.gateway.rsocket.route-id=11")
- .withConfiguration(
- AutoConfigurations.of(RSocketStrategiesAutoConfiguration.class,
- RSocketMessagingAutoConfiguration.class,
- GatewayRSocketCommonAutoConfiguration.class,
- GatewayRSocketAutoConfiguration.class,
- CompositeMeterRegistryAutoConfiguration.class,
- MetricsAutoConfiguration.class))
- .run(context -> assertThat(context).hasSingleBean(RoutingTable.class)
- .hasSingleBean(RoutingTableRoutes.class)
- .hasSingleBean(RoutingTableSocketAcceptorFilter.class)
- .hasSingleBean(GatewayServerRSocketFactoryProcessor.class)
- .hasSingleBean(BrokerProperties.class)
- .hasSingleBean(GatewaySocketAcceptor.class)
- .hasSingleBean(SocketAcceptorPredicateFilter.class)
- .hasSingleBean(RSocketServerBootstrap.class)
- .doesNotHaveBean(SocketAcceptorPredicate.class));
- }
-
- @Configuration
- protected static class MyConfig {
-
- @Bean
- RSocketServerFactory rSocketServerFactory() {
- RSocketServerFactory serverFactory = mock(RSocketServerFactory.class);
- when(serverFactory.create(any(SocketAcceptor.class)))
- .thenReturn(mock(RSocketServer.class));
- return serverFactory;
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/cluster/ClusterServiceTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/cluster/ClusterServiceTests.java
deleted file mode 100644
index 0b57577f..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/cluster/ClusterServiceTests.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.cluster;
-
-import org.junit.Test;
-
-import org.springframework.cloud.gateway.rsocket.actuate.BrokerInfo;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-public class ClusterServiceTests {
-
- @Test
- public void registerIncomingWorks() {
- ClusterService routingTable = new ClusterService();
-
- BrokerInfo brokerInfo = BrokerInfo.of(1L).timestamp(100L).build();
- boolean result = routingTable.registerIncoming(brokerInfo);
-
- String brokerId = brokerInfo.getBrokerId().toString();
- assertThat(result).isTrue();
- assertThat(routingTable.incomingBrokers).containsKey(brokerId);
-
- brokerInfo = BrokerInfo.of(1L).timestamp(10L).build();
- result = routingTable.registerIncoming(brokerInfo);
- assertThat(result).isFalse();
- assertThat(routingTable.incomingBrokers.get(brokerId)).isNotNull()
- .extracting(ClusterService.BrokerEntry::getTimestamp).isEqualTo(100L);
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketIntegrationTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketIntegrationTests.java
deleted file mode 100644
index 454230bf..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketIntegrationTests.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.core;
-
-import java.time.Duration;
-
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import reactor.core.publisher.Hooks;
-import reactor.test.StepVerifier;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.rsocket.RSocketProperties;
-import org.springframework.boot.rsocket.context.RSocketServerBootstrap;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
-import org.springframework.cloud.gateway.rsocket.test.PingPongApp;
-import org.springframework.test.context.junit4.SpringRunner;
-import org.springframework.test.util.ReflectionTestUtils;
-import org.springframework.util.SocketUtils;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest(classes = PingPongApp.class,
- properties = { "ping.take=10", "ping.subscribe=false" },
- webEnvironment = WebEnvironment.RANDOM_PORT)
-public class GatewayRSocketIntegrationTests {
-
- private static int port;
-
- @Autowired
- private PingPongApp.Ping ping;
-
- @Autowired
- private PingPongApp.Pong pong;
-
- @Autowired
- private RSocketProperties properties;
-
- @Autowired
- private PingPongApp.MySocketAcceptorFilter mySocketAcceptorFilter;
-
- @Autowired
- private RSocketServerBootstrap server;
-
- @BeforeClass
- public static void init() {
- Hooks.onOperatorDebug();
- port = SocketUtils.findAvailableTcpPort();
- System.setProperty("spring.rsocket.server.port", String.valueOf(port));
- }
-
- @AfterClass
- public static void after() {
- System.clearProperty("spring.rsocket.server.port");
- }
-
- @Test
- public void contextLoads() {
- // @formatter:off
- StepVerifier.create(ping.getPongFlux())
- .expectSubscription()
- .then(() -> server.stop())
- .thenConsumeWhile(s -> true)
- .expectComplete()
- .verify(Duration.ofSeconds(20));
- // @formatter:on
-
- assertThat(ping.getPongsReceived()).isGreaterThan(0);
- assertThat(pong.getPingsReceived()).isGreaterThan(0);
- Object server = properties.getServer();
- Object port = ReflectionTestUtils.invokeGetterMethod(server, "port");
- assertThat(port).isNotEqualTo(7002);
- assertThat(mySocketAcceptorFilter.invoked()).isTrue();
- assertThat(this.server.isRunning()).isFalse();
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketTests.java
deleted file mode 100644
index dc7bfb79..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/core/GatewayRSocketTests.java
+++ /dev/null
@@ -1,321 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.core;
-
-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.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 reactor.util.function.Tuple2;
-import reactor.util.function.Tuples;
-
-import org.springframework.cloud.gateway.rsocket.autoconfigure.BrokerProperties;
-import org.springframework.cloud.gateway.rsocket.common.metadata.Forwarding;
-import org.springframework.cloud.gateway.rsocket.common.metadata.Metadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.WellKnownKey;
-import org.springframework.cloud.gateway.rsocket.common.test.MetadataEncoder;
-import org.springframework.cloud.gateway.rsocket.route.DefaultRoute;
-import org.springframework.cloud.gateway.rsocket.route.Route;
-import org.springframework.cloud.gateway.rsocket.route.Routes;
-import org.springframework.cloud.gateway.rsocket.routing.LoadBalancerFactory;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTable;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.messaging.rsocket.DefaultMetadataExtractor;
-import org.springframework.messaging.rsocket.MetadataExtractor;
-import org.springframework.messaging.rsocket.PayloadUtils;
-import org.springframework.messaging.rsocket.RSocketStrategies;
-
-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;
-import static org.springframework.cloud.gateway.rsocket.common.metadata.Forwarding.FORWARDING_MIME_TYPE;
-
-/**
- * @author Spencer Gibb
- */
-public class GatewayRSocketTests {
-
- private static Log logger = LogFactory.getLog(GatewayRSocketTests.class);
-
- private RoutingTable routingTable;
-
- private Payload incomingPayload;
-
- private final RSocketStrategies rSocketStrategies = RSocketStrategies.builder()
- .decoder(new Forwarding.Decoder()).encoder(new Forwarding.Encoder()).build();
-
- private DefaultMetadataExtractor metadataExtractor = new DefaultMetadataExtractor(
- rSocketStrategies.decoders());
-
- // TODO: add tests for metrics and other request types
-
- @Before
- public void init() {
- routingTable = mock(RoutingTable.class);
-
- this.metadataExtractor.metadataToExtract(FORWARDING_MIME_TYPE, Forwarding.class,
- "forwarding");
-
- MetadataEncoder encoder = new MetadataEncoder(Metadata.COMPOSITE_MIME_TYPE,
- this.rSocketStrategies);
- Forwarding metadata = Forwarding.of(1).with(WellKnownKey.SERVICE_NAME, "mock")
- .build();
- DataBuffer dataBuffer = encoder.metadata(metadata, FORWARDING_MIME_TYPE).encode();
- DataBuffer data = MetadataEncoder.emptyDataBuffer(rSocketStrategies);
- incomingPayload = PayloadUtils.createPayload(data, dataBuffer);
-
- RSocket rSocket = mock(RSocket.class);
- Tuple2 tuple = Tuples.of("1111", rSocket);
- when(routingTable.findRSockets(any(TagsMetadata.class)))
- .thenReturn(Collections.singletonList(tuple));
-
- 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(routingTable,
- new TestRoutes(filter1, filter2, filter3), metadataExtractor)
- .requestResponse(incomingPayload).block(Duration.ZERO);
-
- assertThat(filter1.invoked()).isTrue();
- assertThat(filter2.invoked()).isTrue();
- assertThat(filter3.invoked()).isTrue();
- assertThat(payload).isNotNull();
- }
-
- @Test
- public void zeroFilters() {
- Payload payload = new TestGatewayRSocket(routingTable, new TestRoutes(),
- metadataExtractor).requestResponse(incomingPayload).block(Duration.ZERO);
-
- assertThat(payload).isNotNull();
- }
-
- @Test
- public void shortcircuitFilter() {
-
- TestFilter filter1 = new TestFilter();
- ShortcircuitingFilter filter2 = new ShortcircuitingFilter();
- TestFilter filter3 = new TestFilter();
-
- TestGatewayRSocket gatewayRSocket = new TestGatewayRSocket(routingTable,
- new TestRoutes(filter1, filter2, filter3), metadataExtractor);
- Mono response = gatewayRSocket.requestResponse(incomingPayload);
-
- // a false filter will create a pending rsocket that blocks forever
- // this tweaks the rsocket to complete.
- gatewayRSocket.getProcessor().onNext(null);
-
- StepVerifier.withVirtualTime(() -> response).expectSubscription()
- .verifyComplete();
-
- assertThat(filter1.invoked()).isTrue();
- assertThat(filter2.invoked()).isTrue();
- assertThat(filter3.invoked()).isFalse();
- }
-
- @Test
- public void asyncFilter() {
-
- AsyncFilter filter = new AsyncFilter();
-
- Payload payload = new TestGatewayRSocket(routingTable, new TestRoutes(filter),
- metadataExtractor).requestResponse(incomingPayload)
- .block(Duration.ofSeconds(5));
-
- assertThat(filter.invoked()).isTrue();
- assertThat(payload).isNotNull();
- }
-
- // TODO: add exception handlers?
- @Test(expected = IllegalStateException.class)
- public void handleErrorFromFilter() {
-
- ExceptionFilter filter = new ExceptionFilter();
-
- new TestGatewayRSocket(routingTable, new TestRoutes(filter), metadataExtractor)
- .requestResponse(incomingPayload).block(Duration.ofSeconds(5));
-
- // assertNull(socket);
- }
-
- private static RouteSetup getMetadata() {
- return RouteSetup.of(1L, "service").build();
- }
-
- private static class TestGatewayRSocket extends GatewayRSocket {
-
- TestGatewayRSocket(RoutingTable routingTable, Routes routes,
- MetadataExtractor metadataExtractor) {
- super(routes, new TestPendingFactory(routingTable, routes, metadataExtractor),
- new LoadBalancerFactory(routingTable), new SimpleMeterRegistry(),
- new BrokerProperties(), metadataExtractor, getMetadata());
- }
-
- private MonoProcessor getProcessor() {
- TestPendingFactory factory = (TestPendingFactory) super.getPendingFactory();
- return factory.processor;
- }
-
- }
-
- private static class TestPendingFactory extends PendingRequestRSocketFactory {
-
- private final MonoProcessor processor = MonoProcessor.create();
-
- private final MetadataExtractor metadataExtractor;
-
- TestPendingFactory(RoutingTable routingTable, Routes routes,
- MetadataExtractor metadataExtractor) {
- super(routingTable, routes, metadataExtractor);
- this.metadataExtractor = metadataExtractor;
- }
-
- @Override
- protected PendingRequestRSocket constructPendingRSocket(
- GatewayExchange exchange) {
- Function> routeFinder = registeredEvent -> getRouteMono(
- registeredEvent, exchange);
- return new PendingRequestRSocket(metadataExtractor, routeFinder,
- tagsMetadata -> {
- Tags tags = exchange.getTags().and("responder.id",
- tagsMetadata.getRouteId());
- exchange.setTags(tags);
- }, processor);
- }
-
- }
-
- private static class TestRoutes implements Routes {
-
- private final Route route;
-
- private List filters;
-
- TestRoutes() {
- this(Collections.emptyList());
- }
-
- TestRoutes(GatewayFilter... filters) {
- this(Arrays.asList(filters));
- }
-
- TestRoutes(List filters) {
- this.filters = filters;
- route = DefaultRoute.builder().id("route1")
- .routingMetadata(RouteSetup.of(1L, "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/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutesTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutesTests.java
deleted file mode 100644
index 1ca106d1..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableRoutesTests.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-import java.util.HashSet;
-
-import io.rsocket.RSocket;
-import org.junit.Test;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.Forwarding;
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-import org.springframework.cloud.gateway.rsocket.route.Route;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTable.RegisteredEvent;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTable.RouteEntry;
-
-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;
-import static org.springframework.cloud.gateway.rsocket.core.GatewayExchange.Type.REQUEST_RESPONSE;
-
-public class RoutingTableRoutesTests {
-
- @Test
- public void routesAreBuilt() {
- RoutingTable routingTable = mock(RoutingTable.class);
- RoutingTableRoutes routes = new RoutingTableRoutes(routingTable);
-
- HashSet routeIds = new HashSet<>();
- routeIds.add("2");
- when(routingTable.findRouteIds(any(TagsMetadata.class))).thenReturn(routeIds);
- addRoute(routes, "1");
- addRoute(routes, "2");
- addRoute(routes, "3");
-
- Forwarding forwarding = Forwarding.of(1L).routeId("2").build();
- Mono routeMono = routes
- .findRoute(new GatewayExchange(REQUEST_RESPONSE, forwarding));
-
- StepVerifier.create(routeMono).consumeNextWith(route -> {
- assertThat(route).isNotNull().extracting(Route::getId).isEqualTo("2");
- }).verifyComplete();
- }
-
- void addRoute(RoutingTableRoutes routes, String routeId) {
- TagsMetadata tagsMetadata = TagsMetadata.builder().routeId(routeId).build();
-
- RSocket rsocket = mock(RSocket.class);
- routes.accept(new RegisteredEvent(new RouteEntry(rsocket, tagsMetadata)));
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableTests.java
deleted file mode 100644
index a80a7b84..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/routing/RoutingTableTests.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * Copyright 2013-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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.routing;
-
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import io.rsocket.AbstractRSocket;
-import io.rsocket.RSocket;
-import org.junit.Test;
-import org.roaringbitmap.RoaringBitmap;
-import reactor.util.function.Tuple2;
-import reactor.util.function.Tuples;
-
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.WellKnownKey;
-import org.springframework.core.style.ToStringCreator;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-public class RoutingTableTests {
-
- @Test
- public void testIndexesCreatedAndSearchWorks() {
- RoutingTable routingTable = new RoutingTable();
- // @formatter:off
- TagsMetadata setupTags1 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "1111")
- .with(WellKnownKey.SERVICE_NAME, "serviceA")
- .with(WellKnownKey.CLUSTER_NAME, "clusterA")
- .build();
- TagsMetadata setupTags2 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "2222")
- .with(WellKnownKey.SERVICE_NAME, "serviceA")
- .with(WellKnownKey.CLUSTER_NAME, "clusterB")
- .build();
- TagsMetadata setupTags3 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "3333")
- .with(WellKnownKey.SERVICE_NAME, "serviceB")
- .with(WellKnownKey.REGION, "region1")
- .build();
- TagsMetadata setupTags4 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "4444")
- .with(WellKnownKey.SERVICE_NAME, "serviceB")
- .with(WellKnownKey.CLUSTER_NAME, "clusterB")
- .with(WellKnownKey.ZONE, "zone1")
- .build();
- TagsMetadata setupTags5 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "5555")
- .with(WellKnownKey.SERVICE_NAME, "serviceA")
- .with(WellKnownKey.CLUSTER_NAME, "clusterB")
- .build();
- // @formatter:on
-
- AtomicInteger internalRouteId = routingTable.internalRouteId;
- RSocket rSocket1 = assertRegister(routingTable, setupTags1,
- internalRouteId.get() + 1);
- internalRouteId.set(99);
- RSocket rSocket2 = assertRegister(routingTable, setupTags2,
- internalRouteId.get() + 1);
- internalRouteId.set(999);
- RSocket rSocket3 = assertRegister(routingTable, setupTags3,
- internalRouteId.get() + 1);
- internalRouteId.set(9999);
- RSocket rSocket4 = assertRegister(routingTable, setupTags4,
- internalRouteId.get() + 1);
- internalRouteId.set(99999);
- RSocket rSocket5 = assertRegister(routingTable, setupTags5,
- internalRouteId.get() + 1);
-
- // @formatter:off
- TagsMetadata searchTags1 = TagsMetadata.builder()
- .with(WellKnownKey.SERVICE_NAME, "serviceA")
- .with(WellKnownKey.CLUSTER_NAME, "clusterB")
- .build();
- // @formatter:on
-
- List> results1 = routingTable.findRSockets(searchTags1);
- assertThat(results1).containsOnly(Tuples.of("2222", rSocket2),
- Tuples.of("5555", rSocket5));
-
- // @formatter:off
- TagsMetadata searchTags2 = TagsMetadata.builder()
- .with(WellKnownKey.ROUTE_ID, "3333")
- .build();
- // @formatter:on
-
- List> results2 = routingTable.findRSockets(searchTags2);
- assertThat(results2).containsOnly(Tuples.of("3333", rSocket3));
-
- // @formatter:off
- TagsMetadata searchTags3 = TagsMetadata.builder()
- .with(WellKnownKey.ZONE, "zone1")
- .with(WellKnownKey.SERVICE_NAME, "serviceB")
- .with(WellKnownKey.CLUSTER_NAME, "clusterB")
- .build();
- // @formatter:on
-
- List> results3 = routingTable.findRSockets(searchTags3);
- assertThat(results3).containsOnly(Tuples.of("4444", rSocket4));
-
- assertDeregister(routingTable, setupTags1);
- assertDeregister(routingTable, setupTags2);
- assertDeregister(routingTable, setupTags3);
- assertDeregister(routingTable, setupTags4);
- assertDeregister(routingTable, setupTags5);
- assertThat(routingTable.deregister(setupTags5)).isFalse();
- }
-
- void assertDeregister(RoutingTable routingTable, TagsMetadata tagsMetadata) {
- boolean result = routingTable.deregister(tagsMetadata);
- assertThat(result).isTrue();
- String routeId = tagsMetadata.getRouteId();
- assertThat(routingTable.internalRouteIdToRouteId).doesNotContainValue(routeId);
- assertThat(routingTable.routeEntries).doesNotContainKey(routeId);
- }
-
- private RSocket assertRegister(RoutingTable routingTable, TagsMetadata tagsMetadata,
- int internalId) {
- String routeId = tagsMetadata.getRouteId();
- RSocket rsocket = new TestRSocket(routeId);
- routingTable.register(tagsMetadata, rsocket);
-
- assertThat(routingTable.internalRouteId).hasValue(internalId);
- assertThat(routingTable.internalRouteIdToRouteId).containsEntry(internalId,
- routeId);
- assertThat(routingTable.routeEntries).containsKey(routeId);
- tagsMetadata.getTags().forEach((key, value) -> {
- RoutingTable.TagKey tagKey = new RoutingTable.TagKey(key, value);
- assertThat(routingTable.tagsToBitmaps).containsKey(tagKey);
- RoaringBitmap bitmap = routingTable.tagsToBitmaps.get(tagKey);
- assertThat(bitmap.contains(internalId));
- });
-
- return rsocket;
- }
-
- static class TestRSocket extends AbstractRSocket {
-
- final String routeId;
-
- TestRSocket(String routeId) {
- this.routeId = routeId;
- }
-
- @Override
- public String toString() {
- return new ToStringCreator(this).append("routeId", routeId).toString();
-
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java
deleted file mode 100644
index 79cc7787..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/GatewaySocketAcceptorTests.java
+++ /dev/null
@@ -1,247 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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.Payload;
-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.BrokerProperties;
-import org.springframework.cloud.gateway.rsocket.common.metadata.Metadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.common.test.MetadataEncoder;
-import org.springframework.cloud.gateway.rsocket.core.GatewayRSocket;
-import org.springframework.cloud.gateway.rsocket.core.GatewayRSocketFactory;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.messaging.rsocket.DefaultMetadataExtractor;
-import org.springframework.messaging.rsocket.PayloadUtils;
-import org.springframework.messaging.rsocket.RSocketStrategies;
-
-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;
-import static org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup.ROUTE_SETUP_MIME_TYPE;
-
-/**
- * @author Spencer Gibb
- */
-public class GatewaySocketAcceptorTests {
-
- private static Log logger = LogFactory.getLog(GatewaySocketAcceptorTests.class);
-
- private GatewayRSocketFactory factory;
-
- private ConnectionSetupPayload setupPayload;
-
- private RSocket sendingSocket;
-
- private MeterRegistry meterRegistry;
-
- private BrokerProperties properties = new BrokerProperties();
-
- private final RSocketStrategies rSocketStrategies = RSocketStrategies.builder()
- .decoder(new RouteSetup.Decoder()).encoder(new RouteSetup.Encoder()).build();
-
- private DefaultMetadataExtractor metadataExtractor = new DefaultMetadataExtractor(
- rSocketStrategies.decoders());
-
- @Before
- public void init() {
- this.factory = mock(GatewayRSocketFactory.class);
- this.setupPayload = mock(ConnectionSetupPayload.class);
- this.sendingSocket = mock(RSocket.class);
- this.meterRegistry = new SimpleMeterRegistry();
-
- this.metadataExtractor.metadataToExtract(ROUTE_SETUP_MIME_TYPE, RouteSetup.class,
- "routesetup");
-
- when(this.factory.create(any(TagsMetadata.class)))
- .thenReturn(mock(GatewayRSocket.class));
-
- when(this.setupPayload.metadataMimeType())
- .thenReturn(Metadata.COMPOSITE_MIME_TYPE.toString());
-
- when(this.setupPayload.hasMetadata()).thenReturn(true);
-
- MetadataEncoder encoder = new MetadataEncoder(Metadata.COMPOSITE_MIME_TYPE,
- this.rSocketStrategies);
- encoder.metadata(RouteSetup.of(1L, "myservice").build(), ROUTE_SETUP_MIME_TYPE);
- DataBuffer dataBuffer = encoder.encode();
- DataBuffer data = MetadataEncoder.emptyDataBuffer(rSocketStrategies);
- Payload payload = PayloadUtils.createPayload(data, dataBuffer);
- when(setupPayload.metadata()).thenReturn(payload.metadata());
- }
-
- // 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, this.metadataExtractor)
- .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, this.metadataExtractor)
- .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, this.metadataExtractor)
- .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, this.metadataExtractor)
- .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, this.metadataExtractor)
- .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/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java
deleted file mode 100644
index 127aafe9..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/socketacceptor/SocketAcceptorPredicateFilterTests.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.socketacceptor;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
-import io.rsocket.ConnectionSetupPayload;
-import io.rsocket.RSocket;
-import org.junit.Test;
-import org.reactivestreams.Publisher;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
-import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
-
-public class SocketAcceptorPredicateFilterTests {
-
- @Test
- public void noPredicateWorks() {
- Mono result = runFilter(Collections.emptyList());
- StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
- }
-
- @Test
- public void singleTruePredicateWorks() {
- TestPredicate predicate = new TestPredicate(true);
- Mono result = runFilter(predicate);
- StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
- assertThat(predicate.invoked()).isTrue();
- }
-
- @Test
- public void singleFalsePredicateWorks() {
- TestPredicate predicate = new TestPredicate(false);
- Mono result = runFilter(predicate);
- StepVerifier.create(result).verifyComplete();
-
- assertThat(predicate.invoked()).isTrue();
- }
-
- @Test
- public void multipleFalsePredicateWorks() {
- TestPredicate predicate = new TestPredicate(false);
- TestPredicate predicate2 = new TestPredicate(false);
- Mono result = runFilter(predicate, predicate2);
- StepVerifier.create(result).verifyComplete();
-
- assertThat(predicate.invoked()).isTrue();
- assertThat(predicate2.invoked()).isTrue(); // Async predicates don't short circuit
- }
-
- @Test
- public void multiplePredicatesNoSuccessWorks() {
- TestPredicate truePredicate = new TestPredicate(true);
- TestPredicate falsePredicate = new TestPredicate(false);
- Mono result = runFilter(truePredicate, falsePredicate);
- StepVerifier.create(result).verifyComplete();
- assertThat(truePredicate.invoked()).isTrue();
- assertThat(falsePredicate.invoked()).isTrue();
- }
-
- @Test
- public void multiplePredicatesSuccessWorks() {
- TestPredicate truePredicate = new TestPredicate(true);
- TestPredicate truePredicate2 = new TestPredicate(true);
- Mono result = runFilter(truePredicate, truePredicate2);
- StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
- assertThat(truePredicate.invoked()).isTrue();
- assertThat(truePredicate2.invoked()).isTrue();
- }
-
- private Mono runFilter(SocketAcceptorPredicate predicate) {
- return runFilter(Collections.singletonList(predicate));
- }
-
- private Mono runFilter(SocketAcceptorPredicate... predicates) {
- return runFilter(Arrays.asList(predicates));
- }
-
- private Mono runFilter(List predicates) {
- SocketAcceptorPredicateFilter filter = new SocketAcceptorPredicateFilter(
- predicates);
- SocketAcceptorExchange exchange = new SocketAcceptorExchange(
- mock(ConnectionSetupPayload.class), mock(RSocket.class));
- SocketAcceptorFilterChain filterChain = new SocketAcceptorFilterChain(
- Collections.singletonList(filter));
- return filter.filter(exchange, filterChain);
- }
-
- private class TestPredicate implements SocketAcceptorPredicate {
-
- private boolean invoked = false;
-
- private final Mono test;
-
- TestPredicate(boolean value) {
- test = Mono.just(value);
- }
-
- @Override
- public Publisher apply(SocketAcceptorExchange exchange) {
- invoked = true;
- return test;
- }
-
- public boolean invoked() {
- return invoked;
- }
-
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
deleted file mode 100644
index 0546226d..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/PingPongApp.java
+++ /dev/null
@@ -1,336 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.gateway.rsocket.test;
-
-import java.time.Duration;
-import java.util.LinkedHashMap;
-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.frame.decoder.PayloadDecoder;
-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.Hooks;
-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.common.metadata.Forwarding;
-import org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup;
-import org.springframework.cloud.gateway.rsocket.common.metadata.TagsMetadata;
-import org.springframework.cloud.gateway.rsocket.common.metadata.WellKnownKey;
-import org.springframework.cloud.gateway.rsocket.common.test.MetadataEncoder;
-import org.springframework.cloud.gateway.rsocket.core.GatewayExchange;
-import org.springframework.cloud.gateway.rsocket.core.GatewayFilter;
-import org.springframework.cloud.gateway.rsocket.core.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.context.ApplicationListener;
-import org.springframework.context.annotation.Bean;
-import org.springframework.core.Ordered;
-import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.messaging.rsocket.RSocketStrategies;
-
-import static io.netty.buffer.Unpooled.EMPTY_BUFFER;
-import static org.springframework.cloud.gateway.rsocket.common.metadata.Metadata.COMPOSITE_MIME_TYPE;
-import static org.springframework.cloud.gateway.rsocket.common.metadata.RouteSetup.ROUTE_SETUP_MIME_TYPE;
-
-@SpringBootApplication
-public class PingPongApp {
-
- @Bean
- public Ping ping1() {
- return new Ping(1L);
- }
-
- @Bean
- @ConditionalOnProperty("ping.two.enabled")
- public Ping ping2() {
- return new Ping(2L);
- }
-
- @Bean
- public Pong pong() {
- return new Pong();
- }
-
- @Bean
- public MySocketAcceptorFilter mySocketAcceptorFilter() {
- return new MySocketAcceptorFilter();
- }
-
- public static void main(String[] args) {
- Hooks.onOperatorDebug();
- 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);
- }
- }
-
- static ByteBuf getRouteSetupMetadata(RSocketStrategies strategies, String name,
- long id) {
- RouteSetup routeSetup = RouteSetup.of(id, name)
- .with("current-time", String.valueOf(System.currentTimeMillis())).build();
- LinkedHashMap tags = new LinkedHashMap<>();
- tags.put(new TagsMetadata.Key(WellKnownKey.TIME_ZONE),
- System.currentTimeMillis() + "");
- DataBuffer dataBuffer = new MetadataEncoder(COMPOSITE_MIME_TYPE, strategies)
- .metadata(routeSetup, ROUTE_SETUP_MIME_TYPE).encode();
- return TagsMetadata.asByteBuf(dataBuffer);
- }
-
- static ByteBuf getForwardingMetadata(RSocketStrategies strategies, String name,
- long id) {
- Forwarding metadata = Forwarding.of(id).serviceName(name).build();
- DataBuffer dataBuffer = new MetadataEncoder(COMPOSITE_MIME_TYPE, strategies)
- .metadata(metadata, Forwarding.FORWARDING_MIME_TYPE).encode();
- return TagsMetadata.asByteBuf(dataBuffer);
- }
-
- @Slf4j
- public static class Ping
- implements Ordered, ApplicationListener {
-
- @Autowired
- private MeterRegistry meterRegistry;
-
- @Autowired
- private RSocketStrategies strategies;
-
- private final Long id;
-
- private final AtomicInteger pongsReceived = new AtomicInteger();
-
- private Flux pongFlux;
-
- public Ping(Long 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.rsocket.server.port",
- Integer.class, 7002);
-
- log.debug("ping.take: " + take);
-
- MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
- meterRegistry, Tag.of("component", "ping"));
- ByteBuf metadata = getRouteSetupMetadata(strategies, "ping", id);
- Payload setupPayload = DefaultPayload.create(EMPTY_BUFFER, metadata);
-
- pongFlux = RSocketFactory.connect().frameDecoder(PayloadDecoder.ZERO_COPY)
- .metadataMimeType(COMPOSITE_MIME_TYPE.toString())
- .setupPayload(setupPayload).addRequesterPlugin(interceptor)
- .transport(TcpClientTransport.create(gatewayPort)) // proxy
- .start().log("startPing" + id)
- .flatMapMany(socket -> doPing(take, socket)).cast(String.class)
- .doOnSubscribe(o -> {
- if (log.isDebugEnabled()) {
- log.debug("ping doOnSubscribe");
- }
- });
-
- boolean subscribe = env.getProperty("ping.subscribe", Boolean.class, true);
-
- if (subscribe) {
- pongFlux.subscribe();
- }
- }
-
- Publisher doPing(Integer take, RSocket socket) {
- Flux pong = socket
- .requestChannel(Flux.interval(Duration.ofSeconds(1)).map(i -> {
- ByteBuf data = ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT,
- "ping" + id);
- ByteBuf routingMetadata = getForwardingMetadata(strategies,
- "pong", id);
- log.debug("Sending ping" + id);
- return DefaultPayload.create(data, routingMetadata);
- // onBackpressure is needed in case pong is not available yet
- }).log("doPing")
- .onBackpressureDrop(payload -> log
- .debug("Dropped payload " + payload.getDataUtf8())))
- .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;
- }
-
- public Flux getPongFlux() {
- return pongFlux;
- }
-
- public int getPongsReceived() {
- return pongsReceived.get();
- }
-
- }
-
- @Slf4j
- public static class Pong
- implements Ordered, ApplicationListener {
-
- @Autowired
- private MeterRegistry meterRegistry;
-
- @Autowired
- private RSocketStrategies strategies;
-
- 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.rsocket.server.port",
- Integer.class, 7002);
- MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
- meterRegistry, Tag.of("component", "pong"));
-
- ByteBuf announcementMetadata = getRouteSetupMetadata(strategies, "pong", 3L);
- RSocketFactory.connect().metadataMimeType(COMPOSITE_MIME_TYPE.toString())
- .setupPayload(
- DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
- .addRequesterPlugin(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 = getForwardingMetadata(strategies,
- "ping", 1L);
- 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/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java
deleted file mode 100644
index 1affc4be..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/java/org/springframework/cloud/gateway/rsocket/test/SocketAcceptorFilterOrderTests.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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
- *
- * https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.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.routing.RoutingTable;
-import org.springframework.cloud.gateway.rsocket.routing.RoutingTableSocketAcceptorFilter;
-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 RoutingTableSocketAcceptorFilter(
- mock(RoutingTable.class));
- List filters = Arrays.asList(predicateFilter,
- registryFilter);
- OrderComparator.sort(filters);
-
- assertThat(filters).containsExactly(registryFilter, predicateFilter);
- }
-
-}
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/resources/application.yml b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/resources/application.yml
deleted file mode 100644
index bf3bf7d1..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-broker/src/test/resources/application.yml
+++ /dev/null
@@ -1,17 +0,0 @@
-
-logging:
- level:
- # org.springframework.cloud.gateway.rsocket: DEBUG
- org.springframework.cloud.gateway.rsocket: TRACE
- org.springframework.messaging.handler.invocation.reactive: TRACE
-
-management:
- endpoints:
- web:
- exposure:
- include: '*'
-spring:
- cloud:
- gateway:
- rsocket:
- route-id: 1234
diff --git a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-client/pom.xml b/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-client/pom.xml
deleted file mode 100644
index 36860a83..00000000
--- a/spring-cloud-gateway-rsocket/spring-cloud-gateway-rsocket-client/pom.xml
+++ /dev/null
@@ -1,145 +0,0 @@
-
-
-
-
- 4.0.0
-
- org.springframework.cloud
- spring-cloud-gateway-rsocket
- 2.2.0.BUILD-SNAPSHOT
- ..
-
- org.springframework.cloud
- spring-cloud-gateway-rsocket-client
- Spring Cloud Gateway RSocket Client
- Spring Cloud Gateway RSocket Client
-
-
-
- org.springframework.boot
- spring-boot-starter-validation
-
-
- org.springframework.boot
- spring-boot-starter-rsocket
-
-
- org.springframework.cloud
- spring-cloud-gateway-rsocket-common
-
-
- org.springframework.boot
- spring-boot-configuration-processor
- true
-
-
- org.springframework.boot
- spring-boot-starter-actuator
- test
-
-
- org.projectlombok
- lombok
- test
-
-
- org.springframework.boot
- spring-boot-starter-test
- test
-
-
- io.projectreactor
- reactor-test
- test
-
-
-
-
-
- org.apache.maven.plugins
- maven-compiler-plugin
-
-