Merge branch 'rsocket'

This commit is contained in:
Spencer Gibb
2019-03-11 14:48:23 -04:00
49 changed files with 4620 additions and 101 deletions

Binary file not shown.

View File

@@ -1 +1 @@
distributionUrl=https://repo1.maven.org/maven2/org/apache/maven/apache-maven/3.5.0/apache-maven-3.5.0-bin.zip
distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.5.4/apache-maven-3.5.4-bin.zip

201
LICENSE Normal file
View File

@@ -0,0 +1,201 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

185
mvnw vendored
View File

@@ -1,23 +1,4 @@
#!/bin/sh
# ----------------------------------------------------------------------------
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# ----------------------------------------------------------------------------
# ----------------------------------------------------------------------------
# Maven2 Start Up Batch script
#
@@ -54,38 +35,16 @@ case "`uname`" in
CYGWIN*) cygwin=true ;;
MINGW*) mingw=true;;
Darwin*) darwin=true
#
# Look for the Apple JDKs first to preserve the existing behaviour, and then look
# for the new JDKs provided by Oracle.
#
if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK ] ; then
#
# Apple JDKs
#
export JAVA_HOME=/System/Library/Frameworks/JavaVM.framework/Versions/CurrentJDK/Home
fi
if [ -z "$JAVA_HOME" ] && [ -L /System/Library/Java/JavaVirtualMachines/CurrentJDK ] ; then
#
# Apple JDKs
#
export JAVA_HOME=/System/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
fi
if [ -z "$JAVA_HOME" ] && [ -L "/Library/Java/JavaVirtualMachines/CurrentJDK" ] ; then
#
# Oracle JDKs
#
export JAVA_HOME=/Library/Java/JavaVirtualMachines/CurrentJDK/Contents/Home
fi
if [ -z "$JAVA_HOME" ] && [ -x "/usr/libexec/java_home" ]; then
#
# Apple JDKs
#
export JAVA_HOME=`/usr/libexec/java_home`
fi
;;
# Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
# See https://developer.apple.com/library/mac/qa/qa1170/_index.html
if [ -z "$JAVA_HOME" ]; then
if [ -x "/usr/libexec/java_home" ]; then
export JAVA_HOME="`/usr/libexec/java_home`"
else
export JAVA_HOME="/Library/Java/Home"
fi
fi
;;
esac
if [ -z "$JAVA_HOME" ] ; then
@@ -130,7 +89,7 @@ if $cygwin ; then
CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
fi
# For Migwn, ensure paths are in UNIX format before anything is touched
# For Mingw, ensure paths are in UNIX format before anything is touched
if $mingw ; then
[ -n "$M2_HOME" ] &&
M2_HOME="`(cd "$M2_HOME"; pwd)`"
@@ -184,27 +143,28 @@ fi
CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
# For Cygwin, switch paths to Windows format before running java
if $cygwin; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --path --windows "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
fi
# traverses directory structure from process work directory to filesystem root
# first directory with .mvn subdirectory is considered project base directory
find_maven_basedir() {
local basedir=$(pwd)
local wdir=$(pwd)
if [ -z "$1" ]
then
echo "Path not specified to find_maven_basedir"
return 1
fi
basedir="$1"
wdir="$1"
while [ "$wdir" != '/' ] ; do
if [ -d "$wdir"/.mvn ] ; then
basedir=$wdir
break
fi
wdir=$(cd "$wdir/.."; pwd)
# workaround for JBEAP-8937 (on Solaris 10/Sparc)
if [ -d "${wdir}" ]; then
wdir=`cd "$wdir/.."; pwd`
fi
# end of workaround
done
echo "${basedir}"
}
@@ -216,13 +176,87 @@ concat_lines() {
fi
}
export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-$(find_maven_basedir)}
BASE_DIR=`find_maven_basedir "$(pwd)"`
if [ -z "$BASE_DIR" ]; then
exit 1;
fi
##########################################################################################
# Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
# This allows using the maven wrapper in projects that prohibit checking in binary data.
##########################################################################################
if [ -r "$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" ]; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found .mvn/wrapper/maven-wrapper.jar"
fi
else
if [ "$MVNW_VERBOSE" = true ]; then
echo "Couldn't find .mvn/wrapper/maven-wrapper.jar, downloading it ..."
fi
jarUrl="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.4.2/maven-wrapper-0.4.2.jar"
while IFS="=" read key value; do
case "$key" in (wrapperUrl) jarUrl="$value"; break ;;
esac
done < "$BASE_DIR/.mvn/wrapper/maven-wrapper.properties"
if [ "$MVNW_VERBOSE" = true ]; then
echo "Downloading from: $jarUrl"
fi
wrapperJarPath="$BASE_DIR/.mvn/wrapper/maven-wrapper.jar"
if command -v wget > /dev/null; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found wget ... using wget"
fi
wget "$jarUrl" -O "$wrapperJarPath"
elif command -v curl > /dev/null; then
if [ "$MVNW_VERBOSE" = true ]; then
echo "Found curl ... using curl"
fi
curl -o "$wrapperJarPath" "$jarUrl"
else
if [ "$MVNW_VERBOSE" = true ]; then
echo "Falling back to using Java to download"
fi
javaClass="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.java"
if [ -e "$javaClass" ]; then
if [ ! -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
if [ "$MVNW_VERBOSE" = true ]; then
echo " - Compiling MavenWrapperDownloader.java ..."
fi
# Compiling the Java class
("$JAVA_HOME/bin/javac" "$javaClass")
fi
if [ -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
# Running the downloader
if [ "$MVNW_VERBOSE" = true ]; then
echo " - Running MavenWrapperDownloader.java ..."
fi
("$JAVA_HOME/bin/java" -cp .mvn/wrapper MavenWrapperDownloader "$MAVEN_PROJECTBASEDIR")
fi
fi
fi
fi
##########################################################################################
# End of extension
##########################################################################################
export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
if [ "$MVNW_VERBOSE" = true ]; then
echo $MAVEN_PROJECTBASEDIR
fi
MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
# Provide a "standardized" way to retrieve the CLI args that will
# work with both Windows and non-Windows executions.
MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
export MAVEN_CMD_LINE_ARGS
# For Cygwin, switch paths to Windows format before running java
if $cygwin; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --path --windows "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
[ -n "$MAVEN_PROJECTBASEDIR" ] &&
MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
fi
WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
@@ -232,22 +266,23 @@ echo "The found version is [${VERSION}]"
if echo $VERSION | egrep -q 'M|RC'; then
echo Activating \"milestone\" profile for version=\"$VERSION\"
echo $MAVEN_ARGS | grep -q milestone || MAVEN_ARGS="$MAVEN_ARGS -Pmilestone"
echo $MAVEN_CONFIG | grep -q milestone || MAVEN_CONFIG="$MAVEN_CONFIG -Pmilestone"
else
echo Deactivating \"milestone\" profile for version=\"$VERSION\"
echo $MAVEN_ARGS | grep -q milestone && MAVEN_ARGS=$(echo $MAVEN_ARGS | sed -e 's/-Pmilestone//')
echo $MAVEN_CONFIG | grep -q milestone && MAVEN_CONFIG=$(echo $MAVEN_CONFIG | sed -e 's/-Pmilestone//')
fi
if echo $VERSION | egrep -q 'RELEASE'; then
echo Activating \"central\" profile for version=\"$VERSION\"
echo $MAVEN_ARGS | grep -q milestone || MAVEN_ARGS="$MAVEN_ARGS -Pcentral"
echo $MAVEN_CONFIG | grep -q milestone || MAVEN_CONFIG="$MAVEN_CONFIG -Pcentral"
else
echo Deactivating \"central\" profile for version=\"$VERSION\"
echo $MAVEN_ARGS | grep -q central && MAVEN_ARGS=$(echo $MAVEN_ARGS | sed -e 's/-Pcentral//')
echo $MAVEN_CONFIG | grep -q central && MAVEN_CONFIG=$(echo $MAVEN_CONFIG | sed -e 's/-Pcentral//')
fi
exec "$JAVACMD" \
$MAVEN_OPTS \
-classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
"-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
${WRAPPER_LAUNCHER} ${MAVEN_ARGS} "$@"
${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"

47
mvnw.cmd vendored
View File

@@ -1,22 +1,3 @@
@REM ----------------------------------------------------------------------------
@REM Licensed to the Apache Software Foundation (ASF) under one
@REM or more contributor license agreements. See the NOTICE file
@REM distributed with this work for additional information
@REM regarding copyright ownership. The ASF licenses this file
@REM to you under the Apache License, Version 2.0 (the
@REM "License"); you may not use this file except in compliance
@REM with the License. You may obtain a copy of the License at
@REM
@REM http://www.apache.org/licenses/LICENSE-2.0
@REM
@REM Unless required by applicable law or agreed to in writing,
@REM software distributed under the License is distributed on an
@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@REM KIND, either express or implied. See the License for the
@REM specific language governing permissions and limitations
@REM under the License.
@REM ----------------------------------------------------------------------------
@REM ----------------------------------------------------------------------------
@REM Maven2 Start Up Batch script
@REM
@@ -35,6 +16,8 @@
@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
@echo off
@REM set title of command window
title %0
@REM enable echoing my setting MAVEN_BATCH_ECHO to 'on'
@if "%MAVEN_BATCH_ECHO%" == "on" echo %MAVEN_BATCH_ECHO%
@@ -80,8 +63,6 @@ goto error
:init
set MAVEN_CMD_LINE_ARGS=%*
@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
@REM Fallback to current working directory if not found.
@@ -117,11 +98,27 @@ for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do s
:endReadAdditionalConfig
SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
set WRAPPER_JAR="".\.mvn\wrapper\maven-wrapper.jar""
set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CMD_LINE_ARGS%
set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.4.2/maven-wrapper-0.4.2.jar"
FOR /F "tokens=1,2 delims==" %%A IN (%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties) DO (
IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
)
@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
if exist %WRAPPER_JAR% (
echo Found %WRAPPER_JAR%
) else (
echo Couldn't find %WRAPPER_JAR%, downloading it ...
echo Downloading from: %DOWNLOAD_URL%
powershell -Command "(New-Object Net.WebClient).DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"
echo Finished downloading %WRAPPER_JAR%
)
@REM End of extension
%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
if ERRORLEVEL 1 goto error
goto end
@@ -142,4 +139,4 @@ if "%MAVEN_BATCH_PAUSE%" == "on" pause
if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE%
exit /B %ERROR_CODE%
exit /B %ERROR_CODE%

View File

@@ -129,6 +129,7 @@
<module>spring-cloud-gateway-mvc</module>
<module>spring-cloud-gateway-webflux</module>
<module>spring-cloud-gateway-core</module>
<module>spring-cloud-gateway-rsocket</module>
<module>spring-cloud-starter-gateway</module>
<module>spring-cloud-gateway-sample</module>
<module>docs</module>

View File

@@ -18,6 +18,7 @@
<description>Spring Cloud Gateway Dependencies</description>
<properties>
<rsocket.version>0.11.17</rsocket.version>
</properties>
<dependencyManagement>
@@ -37,11 +38,31 @@
<artifactId>spring-cloud-gateway-core</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gateway-rsocket</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-gateway</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
<version>${rsocket.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-micrometer</artifactId>
<version>${rsocket.version}</version>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
<version>${rsocket.version}</version>
</dependency>
</dependencies>
</dependencyManagement>

View File

@@ -0,0 +1,170 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright 2018-2019 the original author or authors.
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
~
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gateway</artifactId>
<version>2.2.0.BUILD-SNAPSHOT</version>
<relativePath>..</relativePath> <!-- lookup parent from repository -->
</parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gateway-rsocket</artifactId>
<name>Spring Cloud Gateway RSocket</name>
<description>Spring Cloud Gateway RSocket</description>
<properties>
<reactor.version>Californium-SR5</reactor.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-bom</artifactId>
<version>${reactor.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<!-- TODO: optional -->
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-micrometer</artifactId>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-transport-netty</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<compilerArgs>
<arg>-parameters</arg>
</compilerArgs>
</configuration>
<executions>
<!-- Replacing default-compile as it is treated specially by maven -->
<execution>
<id>default-compile</id>
<phase>none</phase>
</execution>
<!-- Replacing default-testCompile as it is treated specially by maven -->
<execution>
<id>default-testCompile</id>
<phase>none</phase>
</execution>
<execution>
<id>java-compile</id>
<phase>compile</phase>
<goals>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>java-test-compile</id>
<phase>test-compile</phase>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>3.1.0</version>
<executions>
<execution>
<goals>
<goal>test-jar</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>java8plus</id>
<activation>
<jdk>[1.8,2.0)</jdk>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<compilerArgs>
<arg>-parameters</arg>
</compilerArgs>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

View File

@@ -0,0 +1,104 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.autoconfigure;
import java.util.List;
import io.micrometer.core.instrument.MeterRegistry;
import io.rsocket.RSocket;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.gateway.rsocket.registry.Registry;
import org.springframework.cloud.gateway.rsocket.registry.RegistryRoutes;
import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.route.Routes;
import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket;
import org.springframework.cloud.gateway.rsocket.server.GatewayRSocketServer;
import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
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 org.springframework.core.env.Environment;
/**
* @author Spencer Gibb
*/
@Configuration
@ConditionalOnProperty(name = "spring.cloud.gateway.rsocket.enabled", matchIfMissing = true)
@EnableConfigurationProperties
@ConditionalOnClass(RSocket.class)
public class GatewayRSocketAutoConfiguration {
@Bean
public Registry registry() {
return new Registry();
}
// TODO: CompositeRoutes
@Bean
public RegistryRoutes registryRoutes(Registry registry) {
RegistryRoutes registryRoutes = new RegistryRoutes();
registry.addListener(registryRoutes);
return registryRoutes;
}
@Bean
public RegistrySocketAcceptorFilter registrySocketAcceptorFilter(Registry registry) {
return new RegistrySocketAcceptorFilter(registry);
}
@Bean
public GatewayRSocket.Factory gatewayRSocketFactory(Registry registry, Routes routes,
MeterRegistry meterRegistry, GatewayRSocketProperties properties) {
return new GatewayRSocket.Factory(registry, routes, meterRegistry, properties);
}
@Bean
public GatewayRSocketProperties gatewayRSocketProperties(Environment env) {
GatewayRSocketProperties properties = new GatewayRSocketProperties();
if (env.containsProperty("spring.application.name")) {
properties.setId(env.getProperty("spring.application.name")); // set default
// from env
}
return properties;
}
@Bean
public SocketAcceptorPredicateFilter socketAcceptorPredicateFilter(
List<SocketAcceptorPredicate> predicates) {
return new SocketAcceptorPredicateFilter(predicates);
}
@Bean
public GatewaySocketAcceptor socketAcceptor(GatewayRSocket.Factory rsocketFactory,
List<SocketAcceptorFilter> filters, MeterRegistry meterRegistry,
GatewayRSocketProperties properties) {
return new GatewaySocketAcceptor(rsocketFactory, filters, meterRegistry,
properties);
}
@Bean
public GatewayRSocketServer gatewayApp(GatewaySocketAcceptor socketAcceptor,
GatewayRSocketProperties properties, MeterRegistry meterRegistry) {
return new GatewayRSocketServer(properties, socketAcceptor, meterRegistry);
}
}

View File

@@ -0,0 +1,130 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.autoconfigure;
import java.util.ArrayList;
import java.util.List;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.style.ToStringCreator;
@ConfigurationProperties("spring.cloud.gateway.rsocket")
public class GatewayRSocketProperties {
/**
* Enable Gateway RSocket.
*/
private boolean enabled = true;
private String id = "gateway"; // TODO: + UUID?
private final Server server = new Server();
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String getId() {
return this.id;
}
public void setId(String id) {
this.id = id;
}
public Server getServer() {
return server;
}
@Override
public String toString() {
return new ToStringCreator(this).append("enabled", enabled).append("id", id)
.append("server", server).toString();
}
/**
* Server properties.
*/
public static class Server {
// TODO: other transports
public enum TransportType {
/**
* TCP Transport type.
*/
TCP
}
/**
* Tag names and values to be supplied to Micrometer Interceptor.
*/
private List<String> micrometerTags = new ArrayList<>();
/**
* Server port.
*/
private int port = 7002; // TODO: different default port?
public Server() {
micrometerTags.add("component");
micrometerTags.add("gateway");
}
/**
* Server transport type. Defaults to TCP.
*/
private TransportType transport = TransportType.TCP;
public List<String> getMicrometerTags() {
return micrometerTags;
}
public void setMicrometerTags(List<String> micrometerTags) {
this.micrometerTags = micrometerTags;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public TransportType getTransport() {
return transport;
}
public void setTransport(TransportType transport) {
this.transport = transport;
}
@Override
public String toString() {
return new ToStringCreator(this).append("micrometerTags", micrometerTags)
.append("port", port).append("transport", transport).toString();
}
}
}

View File

@@ -0,0 +1,117 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.filter;
import java.util.Collections;
import java.util.List;
import java.util.ListIterator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
import org.springframework.lang.Nullable;
/**
* Default implementation of {@link FilterChain}.
*
* <p>
* 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.
*
* <p>
* 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<F extends RSocketFilter, E extends RSocketExchange, FC extends AbstractFilterChain>
implements FilterChain<E> {
private final Log log = LogFactory.getLog(getClass());
protected final List<F> 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<F> filters) {
this.allFilters = Collections.unmodifiableList(filters);
FC chain = initChain(filters);
this.currentFilter = (F) chain.currentFilter;
this.next = (FC) chain.next;
}
private FC initChain(List<F> filters) {
FC chain = create(filters, null, null);
ListIterator<? extends F> 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<F> 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<F> allFilters, @Nullable F currentFilter,
@Nullable FC next);
public List<F> getFilters() {
return this.allFilters;
}
@Override
@SuppressWarnings("unchecked")
public Mono<Success> filter(E exchange) {
return Mono.defer(() -> this.currentFilter != null && this.next != null
? this.currentFilter.filter(exchange, this.next) : getMonoSuccess());
}
private Mono<Success> getMonoSuccess() {
if (log.isDebugEnabled()) {
log.debug("filter chain completed with success");
}
return MONO_SUCCESS;
}
private static final Mono<Success> MONO_SUCCESS = Mono.just(Success.INSTANCE);
}

View File

@@ -0,0 +1,31 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.filter;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public abstract class AbstractRSocketExchange implements RSocketExchange {
private final Map<String, Object> attributes = new ConcurrentHashMap<>();
@Override
public Map<String, Object> getAttributes() {
return this.attributes;
}
}

View File

@@ -0,0 +1,37 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.filter;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
/**
* Contract to allow a {@link RSocketFilter} to delegate to the next in the chain.
*
* @author Spencer Gibb
*/
public interface FilterChain<E extends RSocketExchange> {
/**
* Delegate to the next {@code WebFilter} in the chain.
* @param exchange the current server exchange
* @return {@code Mono<Void>} to indicate when request handling is complete
*/
Mono<Success> filter(E exchange);
}

View File

@@ -0,0 +1,70 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.filter;
import java.util.Map;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
public interface RSocketExchange {
/**
* Return a mutable map of request attributes for the current exchange.
* @return current attributes.
*/
Map<String, Object> getAttributes();
/**
* Return the request attribute value if present.
* @param name the attribute name
* @param <T> the attribute type
* @return the attribute value
*/
@SuppressWarnings("unchecked")
@Nullable
default <T> 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 <T> the attribute type
* @return the attribute value
*/
@SuppressWarnings("unchecked")
default <T> 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 <T> the attribute type
* @return the attribute value
*/
@SuppressWarnings("unchecked")
default <T> T getAttributeOrDefault(String name, T defaultValue) {
return (T) getAttributes().getOrDefault(name, defaultValue);
}
}

View File

@@ -0,0 +1,52 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.filter;
import reactor.core.publisher.Mono;
/**
* Contract for interception-style, chained processing of Web requests that may be used to
* implement cross-cutting, application-agnostic requirements such as security, timeouts,
* and others.
*
* Copied from WebFilter
*
* @author Spencer Gibb
*/
public interface RSocketFilter<E extends RSocketExchange, FC extends FilterChain<E>> {
/**
* Enum to signal successful end of chain reached without the end being empty, i.e.
* Mono&lt;Void&gt; 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<Success>} to indicate when request processing is complete.
*/
Mono<Success> filter(E exchange, FC chain);
}

View File

@@ -0,0 +1,249 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.metrics;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Meter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.Timer;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.ResponderRSocket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import org.springframework.util.Assert;
import static reactor.core.publisher.SignalType.CANCEL;
import static reactor.core.publisher.SignalType.ON_COMPLETE;
import static reactor.core.publisher.SignalType.ON_ERROR;
public class MicrometerResponderRSocket implements ResponderRSocket {
private static final Log log = LogFactory.getLog(MicrometerResponderRSocket.class);
private final RSocket delegate;
private final InteractionCounters metadataPush;
private final InteractionCounters requestChannel;
private final InteractionCounters requestFireAndForget;
private final InteractionTimers requestResponse;
private final InteractionCounters requestStream;
/**
* Creates a new {@link RSocket}.
* @param delegate the {@link RSocket} to delegate to
* @param meterRegistry the {@link MeterRegistry} to use
* @param tags additional tags to attach to {@link Meter}s
* @throws IllegalArgumentException if {@code delegate} or {@code meterRegistry} is
* {@code null}
*/
public MicrometerResponderRSocket(RSocket delegate, MeterRegistry meterRegistry,
Tag... tags) {
Assert.notNull(delegate, "delegate must not be null");
Assert.notNull(meterRegistry, "meterRegistry must not be null");
this.delegate = delegate;
this.metadataPush = new InteractionCounters(meterRegistry, "metadata.push", tags);
this.requestChannel = new InteractionCounters(meterRegistry, "request.channel",
tags);
this.requestFireAndForget = new InteractionCounters(meterRegistry, "request.fnf",
tags);
this.requestResponse = new InteractionTimers(meterRegistry, "request.response",
tags);
this.requestStream = new InteractionCounters(meterRegistry, "request.stream",
tags);
}
@Override
public void dispose() {
delegate.dispose();
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
return delegate.fireAndForget(payload).doFinally(requestFireAndForget);
}
@Override
public Mono<Void> metadataPush(Payload payload) {
return delegate.metadataPush(payload).doFinally(metadataPush);
}
@Override
public Mono<Void> onClose() {
return delegate.onClose();
}
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return delegate.requestChannel(payloads).doFinally(requestChannel);
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
return Mono.defer(() -> {
Timer.Sample sample = requestResponse.start();
return delegate.requestResponse(payload)
.doFinally(signalType -> requestResponse.accept(sample, signalType));
});
}
@Override
public Flux<Payload> requestStream(Payload payload) {
return delegate.requestStream(payload).doFinally(requestStream);
}
@Override
public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> 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<SignalType> {
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<Timer.Sample, SignalType> {
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;
}
}
}
}

View File

@@ -0,0 +1,53 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.metrics;
import java.util.Objects;
import io.micrometer.core.instrument.Meter;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.rsocket.RSocket;
import io.rsocket.plugins.RSocketInterceptor;
public class MicrometerResponderRSocketInterceptor implements RSocketInterceptor {
private final MeterRegistry meterRegistry;
private final Tag[] tags;
/**
* Creates a new {@link RSocketInterceptor}.
* @param meterRegistry the {@link MeterRegistry} to use to create {@link Meter}s.
* @param tags the additional tags to attach to each {@link Meter}
* @throws NullPointerException if {@code meterRegistry} is {@code null}
*/
public MicrometerResponderRSocketInterceptor(MeterRegistry meterRegistry,
Tag... tags) {
this.meterRegistry = Objects.requireNonNull(meterRegistry,
"meterRegistry must not be null");
this.tags = tags;
}
@Override
public MicrometerResponderRSocket apply(RSocket delegate) {
Objects.requireNonNull(delegate, "delegate must not be null");
return new MicrometerResponderRSocket(delegate, meterRegistry, tags);
}
}

View File

@@ -0,0 +1,129 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.registry;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import io.rsocket.RSocket;
import io.rsocket.util.RSocketProxy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
public class LoadBalancedRSocket {
private static final Log log = LogFactory.getLog(LoadBalancedRSocket.class);
private final List<EnrichedRSocket> delegates = new CopyOnWriteArrayList<>();
private final String serviceName;
private final LoadBalancer loadBalancer;
public LoadBalancedRSocket(String serviceName) {
this(serviceName, new RoundRobinLoadBalancer(serviceName));
}
public LoadBalancedRSocket(String serviceName, LoadBalancer loadBalancer) {
this.serviceName = serviceName;
this.loadBalancer = loadBalancer;
}
public Mono<EnrichedRSocket> choose() {
return this.loadBalancer.apply(this.delegates);
}
public void addRSocket(RSocket rsocket, Metadata metadata) {
this.delegates.add(new EnrichedRSocket(rsocket, metadata));
}
public void remove(Metadata metadata) {
// TODO: move delegates to a map for easy removal
this.delegates.stream()
.filter(enriched -> metadata.matches(enriched.getMetadata())).findFirst()
.ifPresent(this.delegates::remove);
}
public List<EnrichedRSocket> getDelegates() {
return this.delegates;
}
public static class EnrichedRSocket extends RSocketProxy {
private final Metadata metadata;
public EnrichedRSocket(RSocket source, Metadata metadata) {
super(source);
this.metadata = metadata;
}
public Metadata getMetadata() {
return this.metadata;
}
public RSocket getSource() {
return this.source;
}
}
// TODO: Flux<RSocket> as input?
// TODO: reuse commons load balancer?
public interface LoadBalancer
extends Function<List<EnrichedRSocket>, Mono<EnrichedRSocket>> {
}
public static class RoundRobinLoadBalancer implements LoadBalancer {
private final AtomicInteger position;
private final String serviceName;
public RoundRobinLoadBalancer(String serviceName) {
this(serviceName, new Random().nextInt(1000));
}
public RoundRobinLoadBalancer(String serviceName, int seedPosition) {
this.serviceName = serviceName;
this.position = new AtomicInteger(seedPosition);
}
@Override
public Mono<EnrichedRSocket> apply(List<EnrichedRSocket> rSockets) {
if (rSockets.isEmpty()) {
if (log.isWarnEnabled()) {
log.warn("No servers available for: " + this.serviceName);
}
return Mono.empty();
}
// TODO: enforce order?
int pos = Math.abs(this.position.incrementAndGet());
EnrichedRSocket rSocket = rSockets.get(pos % rSockets.size());
return Mono.just(rSocket);
}
}
}

View File

@@ -0,0 +1,110 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.registry;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import io.rsocket.RSocket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.Disposable;
import reactor.core.publisher.DirectProcessor;
import reactor.core.publisher.FluxSink;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import org.springframework.util.Assert;
/**
* The Registry handles all RSocket connections that have been made that have associated
* announcement metadata. RSocket connections can then be found based on routing metadata.
* When a new RSocket is registered, a RegisteredEvent is pushed onto a DirectProcessor
* that is acting as an event bus for registered Consumers.
*/
// TODO: name?
public class Registry {
private static final Log log = LogFactory.getLog(Registry.class);
private final Map<String, LoadBalancedRSocket> rsockets = new ConcurrentHashMap<>();
private final DirectProcessor<RegisteredEvent> registeredEvents = DirectProcessor
.create();
private final FluxSink<RegisteredEvent> registeredEventsSink = registeredEvents
.sink(FluxSink.OverflowStrategy.DROP);
public Registry() {
}
// TODO: Mono<Void>?
public void register(Metadata metadata, RSocket rsocket) {
Assert.notNull(metadata, "metadata may not be null");
Assert.notNull(rsocket, "RSocket may not be null");
if (log.isDebugEnabled()) {
log.debug("Registering RSocket: " + metadata);
}
LoadBalancedRSocket composite = rsockets.computeIfAbsent(metadata.getName(),
s -> new LoadBalancedRSocket(metadata.getName()));
composite.addRSocket(rsocket, metadata);
registeredEventsSink.next(new RegisteredEvent(metadata, rsocket));
}
public void deregister(Metadata metadata) {
Assert.notNull(metadata, "metadata may not be null");
if (log.isDebugEnabled()) {
log.debug("Deregistering RSocket: " + metadata);
}
LoadBalancedRSocket loadBalanced = this.rsockets.get(metadata.getName());
if (loadBalanced != null) {
loadBalanced.remove(metadata);
}
}
public LoadBalancedRSocket getRegistered(Metadata metadata) {
return rsockets.get(metadata.getName());
}
public Disposable addListener(Consumer<RegisteredEvent> consumer) {
return this.registeredEvents.subscribe(consumer);
}
public static class RegisteredEvent {
private final Metadata routingMetadata;
private final RSocket rSocket;
public RegisteredEvent(Metadata routingMetadata, RSocket rSocket) {
Assert.notNull(routingMetadata, "routingMetadata may not be null");
Assert.notNull(rSocket, "RSocket may not be null");
this.routingMetadata = routingMetadata;
this.rSocket = rSocket;
}
public Metadata getRoutingMetadata() {
return routingMetadata;
}
public RSocket getRSocket() {
return rSocket;
}
}
}

View File

@@ -0,0 +1,90 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.registry;
import java.util.Collection;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.route.Route;
import org.springframework.cloud.gateway.rsocket.route.Routes;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
/**
* Creates routes from RegisteredEvents.
*/
public class RegistryRoutes implements Routes, Consumer<Registry.RegisteredEvent> {
private static final Log log = LogFactory.getLog(RegistryRoutes.class);
private Map<String, Route> routes = new ConcurrentHashMap<>();
@Override
public Flux<Route> getRoutes() {
// TODO: sorting
// TODO: caching
Collection<Route> routeCollection = routes.values();
if (log.isDebugEnabled()) {
log.debug("Found routes: " + routeCollection);
}
return Flux.fromIterable(routeCollection);
}
@Override
public void accept(Registry.RegisteredEvent registeredEvent) {
Metadata routingMetadata = registeredEvent.getRoutingMetadata();
String id = getId(routingMetadata);
routes.computeIfAbsent(id, key -> createRoute(id, routingMetadata));
}
private String getId(Metadata routingMetadata) {
String id = routingMetadata.getName();
if (id == null) {
id = UUID.randomUUID().toString();
}
return id;
}
private Route createRoute(String id, Metadata routingMetadata) {
Route route = Route.builder().id(id).routingMetadata(routingMetadata)
.predicate(exchange -> {
// TODO: standard predicates
// TODO: allow customized predicates
Metadata incomingRouting = exchange.getRoutingMetadata();
boolean matches = incomingRouting.getName()
.equalsIgnoreCase(routingMetadata.getName());
return Mono.just(matches);
})
// TODO: allow customized filters
.build();
if (log.isDebugEnabled()) {
log.debug("Created Route for registered service " + route);
}
return route;
}
}

View File

@@ -0,0 +1,54 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.registry;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorExchange;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilterChain;
import org.springframework.core.Ordered;
import org.springframework.util.StringUtils;
/**
* Filter that registers the SendingSocket.
*/
public class RegistrySocketAcceptorFilter implements SocketAcceptorFilter, Ordered {
private final Registry registry;
public RegistrySocketAcceptorFilter(Registry registry) {
this.registry = registry;
}
@Override
public Mono<Success> filter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
if (exchange.getMetadata() != null
&& StringUtils.hasLength(exchange.getMetadata().getName())) {
this.registry.register(exchange.getMetadata(), exchange.getSendingSocket());
}
return chain.filter(exchange);
}
@Override
public int getOrder() {
return HIGHEST_PRECEDENCE + 1000;
}
}

View File

@@ -0,0 +1,203 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.route;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
import org.springframework.cloud.gateway.rsocket.server.GatewayFilter;
import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import org.springframework.core.Ordered;
import org.springframework.core.style.ToStringCreator;
import org.springframework.util.Assert;
/**
* @author Spencer Gibb
*/
public class Route implements Ordered {
private final String id;
private final Metadata targetMetadata;
private final int order;
private final AsyncPredicate<GatewayExchange> predicate;
private final List<GatewayFilter> gatewayFilters;
public static Builder builder() {
return new Builder();
}
private Route(String id, Metadata targetMetadata, int order,
AsyncPredicate<GatewayExchange> predicate,
List<GatewayFilter> gatewayFilters) {
this.id = id;
this.targetMetadata = targetMetadata;
this.order = order;
this.predicate = predicate;
this.gatewayFilters = gatewayFilters;
}
public String getId() {
return this.id;
}
public Metadata getTargetMetadata() {
return this.targetMetadata;
}
public int getOrder() {
return order;
}
public AsyncPredicate<GatewayExchange> getPredicate() {
return this.predicate;
}
public List<GatewayFilter> getFilters() {
return Collections.unmodifiableList(this.gatewayFilters);
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
Route route = (Route) o;
return Objects.equals(id, route.id)
&& Objects.equals(targetMetadata, route.targetMetadata)
&& Objects.equals(order, route.order)
&& Objects.equals(predicate, route.predicate)
&& Objects.equals(gatewayFilters, route.gatewayFilters);
}
@Override
public int hashCode() {
return Objects.hash(id, targetMetadata, predicate, gatewayFilters);
}
@Override
public String toString() {
return new ToStringCreator(this).append("id", id)
.append("targetMetadata", targetMetadata).append("order", order)
.append("predicate", predicate).append("gatewayFilters", gatewayFilters)
.toString();
}
public static class Builder {
protected String id;
protected Metadata routingMetadata;
protected int order = 0;
protected AsyncPredicate<GatewayExchange> predicate;
protected List<GatewayFilter> 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<GatewayExchange> getPredicate() {
return this.predicate;
}
public Builder routingMetadata(Metadata routingMetadata) {
this.routingMetadata = routingMetadata;
return this;
}
public Builder setFilters(List<GatewayFilter> gatewayFilters) {
this.gatewayFilters = gatewayFilters;
return this;
}
public Builder filter(GatewayFilter gatewayFilter) {
this.gatewayFilters.add(gatewayFilter);
return this;
}
public Builder filters(Collection<GatewayFilter> gatewayFilters) {
this.gatewayFilters.addAll(gatewayFilters);
return this;
}
public Builder filters(GatewayFilter... gatewayFilters) {
return filters(Arrays.asList(gatewayFilters));
}
public Builder predicate(AsyncPredicate<GatewayExchange> predicate) {
this.predicate = predicate;
return this;
}
public Builder and(AsyncPredicate<GatewayExchange> predicate) {
Assert.notNull(this.predicate, "can not call and() on null predicate");
this.predicate = this.predicate.and(predicate);
return this;
}
public Builder or(AsyncPredicate<GatewayExchange> predicate) {
Assert.notNull(this.predicate, "can not call or() on null predicate");
this.predicate = this.predicate.or(predicate);
return this;
}
public Builder negate() {
Assert.notNull(this.predicate, "can not call negate() on null predicate");
this.predicate = this.predicate.negate();
return this;
}
public Route build() {
Assert.notNull(this.id, "id can not be null");
Assert.notNull(this.routingMetadata, "targetMetadata can not be null");
Assert.notNull(this.predicate, "predicate can not be null");
return new Route(this.id, this.routingMetadata, this.order, predicate,
this.gatewayFilters);
}
}
}

View File

@@ -0,0 +1,61 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.route;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
/**
* @author Spencer Gibb
*/
public interface Routes {
/** log. */
Log log = LogFactory.getLog(Routes.class);
Flux<Route> getRoutes();
default Mono<Route> 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;
});
}
}

View File

@@ -0,0 +1,103 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import io.micrometer.core.instrument.Tags;
import io.rsocket.Payload;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.gateway.rsocket.filter.AbstractRSocketExchange;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
/**
* Exchange object used in GatewayFilterChain started by GatewayRSocket.
*/
public class GatewayExchange extends AbstractRSocketExchange {
private static final Log log = LogFactory.getLog(GatewayExchange.class);
/**
* Key for the route object in attributes.
*/
public static final String ROUTE_ATTR = "__route_attr_";
enum Type {
FIRE_AND_FORGET("request.fnf"), REQUEST_CHANNEL(
"request.channel"), REQUEST_RESPONSE(
"request.response"), REQUEST_STREAM("request.stream");
private String key;
Type(String key) {
this.key = key;
}
String getKey() {
return this.key;
}
}
private final Type type;
private final Metadata routingMetadata;
private Tags tags = Tags.empty();
public static GatewayExchange fromPayload(Type type, Payload payload) {
return new GatewayExchange(type, getRoutingMetadata(payload));
}
private static Metadata getRoutingMetadata(Payload payload) {
if (payload == null || !payload.hasMetadata()) { // and metadata is routing
return null;
}
// TODO: deal with composite metadata
Metadata metadata = Metadata.decodeMetadata(payload.sliceMetadata());
if (log.isDebugEnabled()) {
log.debug("found routing metadata " + metadata);
}
return metadata;
}
public GatewayExchange(Type type, Metadata routingMetadata) {
this.type = type;
this.routingMetadata = routingMetadata;
}
public Type getType() {
return type;
}
public Metadata getRoutingMetadata() {
return routingMetadata;
}
public Tags getTags() {
return this.tags;
}
public void setTags(Tags tags) {
this.tags = tags;
}
}

View File

@@ -0,0 +1,24 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter;
public interface GatewayFilter
extends RSocketFilter<GatewayExchange, GatewayFilterChain> {
}

View File

@@ -0,0 +1,53 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import java.util.List;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.filter.AbstractFilterChain;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
public class GatewayFilterChain
extends AbstractFilterChain<GatewayFilter, GatewayExchange, GatewayFilterChain> {
/**
* Public constructor with the list of filters and the target handler to use.
* @param filters the filters ahead of the handler
*/
private GatewayFilterChain(List<GatewayFilter> filters) {
super(filters);
}
protected GatewayFilterChain(List<GatewayFilter> allFilters,
GatewayFilter currentFilter, GatewayFilterChain next) {
super(allFilters, currentFilter, next);
}
@Override
protected GatewayFilterChain create(List<GatewayFilter> allFilters,
GatewayFilter currentFilter, GatewayFilterChain next) {
return new GatewayFilterChain(allFilters, currentFilter, next);
}
public static Mono<Success> executeFilterChain(List<GatewayFilter> filters,
GatewayExchange exchange) {
return new GatewayFilterChain(filters).filter(exchange);
}
}

View File

@@ -0,0 +1,23 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
public interface GatewayPredicate extends AsyncPredicate<GatewayExchange> {
}

View File

@@ -0,0 +1,329 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.logging.Level;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.Timer;
import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.ResponderRSocket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket;
import org.springframework.cloud.gateway.rsocket.registry.Registry;
import org.springframework.cloud.gateway.rsocket.route.Route;
import org.springframework.cloud.gateway.rsocket.route.Routes;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.ROUTE_ATTR;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.FIRE_AND_FORGET;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_CHANNEL;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_RESPONSE;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_STREAM;
import static org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain.executeFilterChain;
/**
* Acts as a proxy to other registered sockets. Creates a GatewayExchange and attempts to
* locate a Route. If a Route is found, it is added to the exchange and the filter chains
* is executed againts the Route's filters. If the filter chain is successful, an attempt
* to locate a target RSocket via the Registry is executed. If not found a pending RSocket
* * is returned.
*/
public class GatewayRSocket extends AbstractRSocket implements ResponderRSocket {
private static final Log log = LogFactory.getLog(GatewayRSocket.class);
private final Registry registry;
private final Routes routes;
private final MeterRegistry meterRegistry;
private final GatewayRSocketProperties properties;
private final Metadata metadata;
GatewayRSocket(Registry registry, Routes routes, MeterRegistry meterRegistry,
GatewayRSocketProperties properties, Metadata metadata) {
this.registry = registry;
this.routes = routes;
this.meterRegistry = meterRegistry;
this.properties = properties;
this.metadata = metadata;
this.onClose().doOnSuccess(v -> registry.deregister(metadata))
// .doOnNext(v -> log.error("OnClose doOnNext"))
.doOnError(t -> {
if (log.isErrorEnabled()) {
log.error("Error received, deregistering " + metadata, t);
}
registry.deregister(metadata);
})
// .doOnTerminate(() -> log.error("OnClose doOnTerminate"))
// .doFinally(st -> log.error("OnClose doFinally"))
.subscribe();
}
protected Registry getRegistry() {
return registry;
}
protected Routes getRoutes() {
return routes;
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
GatewayExchange exchange = createExchange(FIRE_AND_FORGET, payload);
return findRSocketOrCreatePending(exchange)
.flatMap(rSocket -> rSocket.fireAndForget(payload))
.doOnError(t -> count(exchange, "error"))
.doFinally(s -> count(exchange, ""));
}
private GatewayExchange createExchange(GatewayExchange.Type type, Payload payload) {
GatewayExchange exchange = GatewayExchange.fromPayload(type, payload);
Tags tags = getTags(exchange);
exchange.setTags(tags);
return exchange;
}
private Tags getTags(GatewayExchange exchange) {
// TODO: add tags to exchange
String requesterName = this.metadata.getName();
String requesterId = this.metadata.get("id");
String responderName = exchange.getRoutingMetadata().getName();
Assert.hasText(responderName, "responderName must not be empty");
Assert.hasText(requesterId, "requesterId must not be empty");
Assert.hasText(requesterName, "requesterName must not be empty");
// responder.id happens in a callback, later
return Tags.of("requester.name", requesterName, "responder.name", responderName,
"requester.id", requesterId, "gateway.id", this.properties.getId());
}
@Override
public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
GatewayExchange exchange = createExchange(REQUEST_CHANNEL, payload);
Tags responderTags = Tags.of("source", "responder");
return findRSocketOrCreatePending(exchange).flatMapMany(rSocket -> {
Tags requesterTags = Tags.of("source", "requester");
Flux<Payload> flux = Flux.from(payloads)
.doOnNext(s -> count(exchange, "payload", requesterTags))
.doOnError(t -> count(exchange, "error", requesterTags))
.doFinally(s -> count(exchange, requesterTags));
if (rSocket instanceof ResponderRSocket) {
ResponderRSocket socket = (ResponderRSocket) rSocket;
return socket.requestChannel(payload, flux).log(
GatewayRSocket.class.getName() + ".request-channel",
Level.FINEST);
}
return rSocket.requestChannel(flux);
}).doOnNext(s -> count(exchange, "payload", responderTags))
.doOnError(t -> count(exchange, "error", responderTags))
.doFinally(s -> count(exchange, responderTags));
}
private void count(GatewayExchange exchange, String suffix) {
count(exchange, suffix, Tags.empty());
}
private void count(GatewayExchange exchange, Tags additionalTags) {
count(exchange, null, additionalTags);
}
private void count(GatewayExchange exchange, String suffix, Tags additionalTags) {
Tags tags = exchange.getTags().and(additionalTags);
String name = getMetricName(exchange, suffix);
this.meterRegistry.counter(name, tags).increment();
}
private String getMetricName(GatewayExchange exchange) {
return getMetricName(exchange, null);
}
private String getMetricName(GatewayExchange exchange, String suffix) {
StringBuilder name = new StringBuilder("forward.");
name.append(exchange.getType().getKey());
if (StringUtils.hasLength(suffix)) {
name.append(".");
name.append(suffix);
}
return name.toString();
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
AtomicReference<Timer.Sample> timer = new AtomicReference<>();
GatewayExchange exchange = createExchange(REQUEST_RESPONSE, payload);
return findRSocketOrCreatePending(exchange)
.flatMap(rSocket -> rSocket.requestResponse(payload))
.doOnSubscribe(s -> timer.set(Timer.start(meterRegistry)))
.doOnError(t -> count(exchange, "error"))
.doFinally(s -> timer.get().stop(meterRegistry
.timer(getMetricName(exchange), exchange.getTags())));
}
@Override
public Flux<Payload> requestStream(Payload payload) {
GatewayExchange exchange = createExchange(REQUEST_STREAM, payload);
return findRSocketOrCreatePending(exchange)
.flatMapMany(rSocket -> rSocket.requestStream(payload))
// S N E F
.doOnNext(s -> count(exchange, "payload"))
.doOnError(t -> count(exchange, "error"))
.doFinally(s -> count(exchange, Tags.empty()));
}
/**
* Attempt to locate target RSocket via filter chain. If not found, create a pending
* RSocket.
* @param exchange GatewayExchange
* @return
*/
private Mono<RSocket> findRSocketOrCreatePending(GatewayExchange exchange) {
return findRSocket(exchange)
// if a route can't be found or registered RSocket, create pending
.switchIfEmpty(createPendingRSocket(exchange));
}
private Mono<RSocket> createPendingRSocket(GatewayExchange exchange) {
if (log.isDebugEnabled()) {
log.debug("creating pending RSocket for " + exchange.getRoutingMetadata());
}
PendingRequestRSocket pending = constructPendingRSocket(exchange);
Disposable disposable = this.registry.addListener(pending);
pending.setSubscriptionDisposable(disposable);
return Mono.just(pending);
}
/* for testing */ PendingRequestRSocket constructPendingRSocket(
GatewayExchange exchange) {
Function<Registry.RegisteredEvent, Mono<Route>> routeFinder = registeredEvent -> getRouteMono(
registeredEvent, exchange);
return new PendingRequestRSocket(routeFinder, map -> {
Tags tags = exchange.getTags().and("responder.id", map.get("id"));
exchange.setTags(tags);
});
}
protected Mono<Route> getRouteMono(Registry.RegisteredEvent registeredEvent,
GatewayExchange exchange) {
return findRoute(exchange)
.log(PendingRequestRSocket.class.getName() + ".find route pending",
Level.FINEST)
// can this be replaced with filter?
.flatMap(
route -> matchRoute(route, registeredEvent.getRoutingMetadata()));
}
private Mono<Route> findRoute(GatewayExchange exchange) {
Mono<Route> routeMono;
/*
* if (this.route != null) { //TODO: cache Route? routeMono = Mono.just(route); }
* else {
*/
routeMono = this.routes.findRoute(exchange);
// }
return routeMono;
}
private Mono<Route> matchRoute(Route route, Metadata annoucementMetadata) {
Metadata targetMetadata = route.getTargetMetadata();
if (targetMetadata.matches(annoucementMetadata)) {
return Mono.just(route);
}
return Mono.empty();
}
/**
* First locate Route. If found, put route in exchange and execute filter chain. If
* successful, locate target RSocket.
* @param exchange GatewayExchange.
* @return Target RSocket or empty.
*/
private Mono<RSocket> findRSocket(GatewayExchange exchange) {
return this.routes.findRoute(exchange)
.log(GatewayRSocket.class.getName() + ".find route", Level.FINEST)
.flatMap(route -> {
// put route in exchange for later use
exchange.getAttributes().put(ROUTE_ATTR, route);
return executeFilterChain(route.getFilters(), exchange)
.flatMap(success -> {
LoadBalancedRSocket loadBalancedRSocket = registry
.getRegistered(exchange.getRoutingMetadata());
return loadBalancedRSocket.choose();
}).map(enrichedRSocket -> {
Metadata metadata = enrichedRSocket.getMetadata();
Tags tags = exchange.getTags().and("responder.id",
metadata.get("id"));
exchange.setTags(tags);
return enrichedRSocket;
}).cast(RSocket.class).switchIfEmpty(doOnEmpty(exchange));
});
// TODO: deal with connecting to cluster?
}
private Mono<RSocket> doOnEmpty(GatewayExchange exchange) {
if (log.isDebugEnabled()) {
log.debug("Unable to find destination RSocket for "
+ exchange.getRoutingMetadata());
}
return Mono.empty();
}
public static class Factory {
private final Registry registry;
private final Routes routes;
private final MeterRegistry meterRegistry;
private final GatewayRSocketProperties properties;
public Factory(Registry registry, Routes routes, MeterRegistry meterRegistry,
GatewayRSocketProperties properties) {
this.registry = registry;
this.routes = routes;
this.meterRegistry = meterRegistry;
this.properties = properties;
}
public GatewayRSocket create(Metadata metadata) {
return new GatewayRSocket(this.registry, this.routes, this.meterRegistry,
this.properties, metadata);
}
}
}

View File

@@ -0,0 +1,153 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import io.rsocket.RSocketFactory;
import io.rsocket.SocketAcceptor;
import io.rsocket.micrometer.MicrometerDuplexConnectionInterceptor;
import io.rsocket.plugins.RSocketInterceptor;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties.Server.TransportType;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.Ordered;
import org.springframework.util.Assert;
public class GatewayRSocketServer implements Ordered, SmartLifecycle {
private static final Log log = LogFactory.getLog(GatewayRSocketServer.class);
private static final RSocketInterceptor[] EMPTY_INTERCEPTORS = new RSocketInterceptor[0];
private final GatewayRSocketProperties properties;
private final SocketAcceptor socketAcceptor;
private final List<RSocketInterceptor> serverInterceptors;
private final AtomicBoolean running = new AtomicBoolean();
private CloseableChannel closeableChannel;
private final MeterRegistry meterRegistry;
public GatewayRSocketServer(GatewayRSocketProperties properties,
SocketAcceptor socketAcceptor, MeterRegistry meterRegistry) {
this(properties, socketAcceptor, meterRegistry, EMPTY_INTERCEPTORS);
}
public GatewayRSocketServer(GatewayRSocketProperties properties,
SocketAcceptor socketAcceptor, MeterRegistry meterRegistry,
RSocketInterceptor... interceptors) {
Assert.notNull(properties, "properties may not be null");
Assert.notNull(socketAcceptor, "socketAcceptor may not be null");
Assert.notNull(meterRegistry, "meterRegistry may not be null");
Assert.notNull(interceptors, "interceptors may not be null");
this.properties = properties;
this.socketAcceptor = socketAcceptor;
this.meterRegistry = meterRegistry;
this.serverInterceptors = Arrays.asList(interceptors);
}
@Override
public int getOrder() {
// return 0;
return HIGHEST_PRECEDENCE;
}
@Override
public void start() {
if (running.compareAndSet(false, true)) {
startServer();
}
}
@Override
public void stop() {
if (running.compareAndSet(true, false)) {
if (log.isInfoEnabled()) {
log.info("Stopping Gateway RSocket Server");
}
if (closeableChannel != null) {
closeableChannel.dispose();
}
}
}
@Override
public boolean isRunning() {
return running.get();
}
protected void startServer() {
GatewayRSocketProperties.Server server = properties.getServer();
int port = server.getPort();
TransportType transportType = server.getTransport();
TcpServerTransport transport;
switch (transportType) {
case TCP:
transport = TcpServerTransport.create(port);
break;
default:
throw new IllegalArgumentException(
"No support for server transport " + transportType);
}
if (log.isInfoEnabled()) {
log.info("Starting Gateway RSocket Server on port: " + port + ", transport: "
+ transportType);
}
RSocketFactory.ServerRSocketFactory factory = RSocketFactory.receive();
serverInterceptors.forEach(factory::addServerPlugin);
List<String> micrometerTags = server.getMicrometerTags();
Tag[] tags = Tags.of(micrometerTags.toArray(new String[] {}))
.and("gateway.id", properties.getId()).stream()
.collect(Collectors.toList()).toArray(new Tag[] {});
factory
// TODO: add as bean like serverInterceptors above
.addConnectionPlugin(
new MicrometerDuplexConnectionInterceptor(meterRegistry, tags))
.errorConsumer(throwable -> {
if (log.isDebugEnabled()) {
log.debug("Error with connection", throwable);
}
}) // TODO: add configurable errorConsumer
.acceptor(this.socketAcceptor).transport(transport).start()
.map(closeableChannel -> {
this.closeableChannel = closeableChannel;
return closeableChannel;
}).subscribe();
}
}

View File

@@ -0,0 +1,152 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.logging.Level;
import io.rsocket.AbstractRSocket;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.ResponderRSocket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.util.function.Tuple2;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
import org.springframework.cloud.gateway.rsocket.registry.Registry.RegisteredEvent;
import org.springframework.cloud.gateway.rsocket.route.Route;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.ROUTE_ATTR;
import static org.springframework.cloud.gateway.rsocket.server.GatewayExchange.Type.REQUEST_STREAM;
import static org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain.executeFilterChain;
public class PendingRequestRSocket extends AbstractRSocket
implements ResponderRSocket, Consumer<RegisteredEvent> {
private static final Log log = LogFactory.getLog(PendingRequestRSocket.class);
private final Function<RegisteredEvent, Mono<Route>> routeFinder;
private final Consumer<Metadata> metadataCallback;
private final MonoProcessor<RSocket> rSocketProcessor;
private Disposable subscriptionDisposable;
private Route route;
public PendingRequestRSocket(Function<RegisteredEvent, Mono<Route>> routeFinder,
Consumer<Metadata> metadataCallback) {
this(routeFinder, metadataCallback, MonoProcessor.create());
}
/* for testing */ PendingRequestRSocket(
Function<RegisteredEvent, Mono<Route>> routeFinder,
Consumer<Metadata> metadataCallback,
MonoProcessor<RSocket> rSocketProcessor) {
this.routeFinder = routeFinder;
this.metadataCallback = metadataCallback;
this.rSocketProcessor = rSocketProcessor;
}
/**
* Find route (if needed) using pendingExchange. If found, see if the route target
* matches the registered service. If it matches, send registered RSocket to
* processor. Then execute normal filter chain. If filter chain is successful, execute
* request.
* @param registeredEvent the RegisteredEvent
*/
@Override
public void accept(RegisteredEvent registeredEvent) {
this.routeFinder.apply(registeredEvent).subscribe(route -> {
this.route = route;
this.metadataCallback.accept(registeredEvent.getRoutingMetadata());
this.rSocketProcessor.onNext(registeredEvent.getRSocket());
this.rSocketProcessor.onComplete();
if (this.subscriptionDisposable != null) {
this.subscriptionDisposable.dispose();
}
});
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
return processor("pending-request-faf", payload)
.flatMap(tuple -> tuple.getT1().fireAndForget(payload));
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
return processor("pending-request-rr", payload)
.flatMap(tuple -> tuple.getT1().requestResponse(payload));
}
@Override
public Flux<Payload> requestStream(Payload payload) {
return processor("pending-request-rs", payload)
.flatMapMany(tuple -> tuple.getT1().requestStream(payload));
}
@Override
public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
return processor("pending-request-rc", payload).flatMapMany(tuple -> {
RSocket rSocket = tuple.getT1();
if (rSocket instanceof ResponderRSocket) {
ResponderRSocket socket = (ResponderRSocket) rSocket;
return socket.requestChannel(payload, payloads);
}
return rSocket.requestChannel(payloads);
});
}
/**
* After processor receives onNext signal, get route from exchange attrs, create a new
* exchange from payload. Copy exchange attrs. Execute filter chain, if successful,
* execute request.
* @param logCategory log category
* @param payload payload.
* @return
*/
protected Mono<Tuple2<RSocket, Success>> processor(String logCategory,
Payload payload) {
return rSocketProcessor
.log(PendingRequestRSocket.class.getName() + "." + logCategory,
Level.FINEST)
.flatMap(rSocket -> {
GatewayExchange exchange = GatewayExchange.fromPayload(REQUEST_STREAM,
payload);
exchange.getAttributes().put(ROUTE_ATTR, route);
// exchange.getAttributes().putAll(pendingExchange.getAttributes());
return Mono.just(rSocket)
.zipWith(executeFilterChain(route.getFilters(), exchange));
});
}
public void setSubscriptionDisposable(Disposable subscriptionDisposable) {
this.subscriptionDisposable = subscriptionDisposable;
}
}

View File

@@ -0,0 +1,102 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.util.List;
import java.util.logging.Level;
import java.util.stream.Collectors;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.micrometer.core.instrument.Tags;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.metrics.MicrometerResponderRSocket;
import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
public class GatewaySocketAcceptor implements SocketAcceptor {
private final SocketAcceptorFilterChain filterChain;
private final GatewayRSocket.Factory rSocketFactory;
private final MeterRegistry meterRegistry;
private final GatewayRSocketProperties properties;
public GatewaySocketAcceptor(GatewayRSocket.Factory rSocketFactory,
List<SocketAcceptorFilter> filters, MeterRegistry meterRegistry,
GatewayRSocketProperties properties) {
this.rSocketFactory = rSocketFactory;
this.filterChain = new SocketAcceptorFilterChain(filters);
this.meterRegistry = meterRegistry;
this.properties = properties;
}
@Override
@SuppressWarnings("Duplicates")
public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
// decorate GatewayRSocket with metrics
// current gateway id, type requester, service name (from metadata), service id
Tags requesterTags = Tags.of("gateway.id", properties.getId(), "type",
"requester");
Tags metadataTags;
SocketAcceptorExchange exchange;
if (setup.hasMetadata()) { // TODO: and setup.metadataMimeType() is Announcement
// metadata or composite
Metadata metadata = Metadata.decodeMetadata(setup.sliceMetadata());
metadataTags = Tags.of("service.name", metadata.getName()).and("service.id",
metadata.get("id"));
// enrich exchange to have metadata
exchange = new SocketAcceptorExchange(setup,
decorate(sendingSocket, requesterTags.and(metadataTags)), metadata);
}
else {
metadataTags = Tags.of("service.name", "UNKNOWN").and("service.id",
"UNKNOWN");
exchange = new SocketAcceptorExchange(setup,
decorate(sendingSocket, requesterTags));
}
Tags responderTags = Tags
.of("gateway.id", properties.getId(), "type", "responder")
.and(metadataTags);
// decorate with metrics gateway id, type responder, service name, service id
// (instance id)
return this.filterChain.filter(exchange)
.log(GatewaySocketAcceptor.class.getName()
+ ".socket acceptor filter chain", Level.FINEST)
.map(success -> decorate(
this.rSocketFactory.create(exchange.getMetadata()),
responderTags));
}
private RSocket decorate(RSocket rSocket, Tags tags) {
Tag[] tagArray = tags.stream().collect(Collectors.toList()).toArray(new Tag[] {});
return new MicrometerResponderRSocket(rSocket, meterRegistry, tagArray);
}
}

View File

@@ -0,0 +1,58 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.util.Collections;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.RSocket;
import org.springframework.cloud.gateway.rsocket.filter.AbstractRSocketExchange;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
public class SocketAcceptorExchange extends AbstractRSocketExchange {
private final ConnectionSetupPayload setup;
private final RSocket sendingSocket;
private final Metadata metadata;
public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket) {
this(setup, sendingSocket, new Metadata(null, Collections.emptyMap()));
}
public SocketAcceptorExchange(ConnectionSetupPayload setup, RSocket sendingSocket,
Metadata metadata) {
this.setup = setup;
this.sendingSocket = sendingSocket;
this.metadata = metadata;
}
public ConnectionSetupPayload getSetup() {
return setup;
}
public RSocket getSendingSocket() {
return sendingSocket;
}
public Metadata getMetadata() {
return metadata;
}
}

View File

@@ -0,0 +1,24 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter;
public interface SocketAcceptorFilter
extends RSocketFilter<SocketAcceptorExchange, SocketAcceptorFilterChain> {
}

View File

@@ -0,0 +1,45 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.util.List;
import org.springframework.cloud.gateway.rsocket.filter.AbstractFilterChain;
public class SocketAcceptorFilterChain extends
AbstractFilterChain<SocketAcceptorFilter, SocketAcceptorExchange, SocketAcceptorFilterChain> {
/**
* Public constructor with the list of filters and the target handler to use.
* @param filters the filters ahead of the handler
*/
public SocketAcceptorFilterChain(List<SocketAcceptorFilter> filters) {
super(filters);
}
public SocketAcceptorFilterChain(List<SocketAcceptorFilter> allFilters,
SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
super(allFilters, currentFilter, next);
}
@Override
protected SocketAcceptorFilterChain create(List<SocketAcceptorFilter> allFilters,
SocketAcceptorFilter currentFilter, SocketAcceptorFilterChain next) {
return new SocketAcceptorFilterChain(allFilters, currentFilter, next);
}
}

View File

@@ -0,0 +1,23 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
public interface SocketAcceptorPredicate extends AsyncPredicate<SocketAcceptorExchange> {
}

View File

@@ -0,0 +1,62 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.util.List;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.support.AsyncPredicate;
import org.springframework.core.Ordered;
import org.springframework.util.Assert;
public class SocketAcceptorPredicateFilter implements SocketAcceptorFilter, Ordered {
private final AsyncPredicate<SocketAcceptorExchange> predicate;
// TODO: change from List to Flux?
public SocketAcceptorPredicateFilter(List<SocketAcceptorPredicate> predicates) {
Assert.notNull(predicates, "predicates may not be null");
if (predicates.isEmpty()) {
predicate = exchange -> Mono.just(true);
}
else {
AsyncPredicate<SocketAcceptorExchange> 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<Success> filter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
return Mono.from(predicate.apply(exchange)).flatMap(test -> {
if (test) {
return chain.filter(exchange);
}
return Mono.empty();
});
}
}

View File

@@ -0,0 +1,50 @@
/*
* Copyright 2013-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.support;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.util.Assert;
/**
* @author Ben Hale
*/
public interface AsyncPredicate<T> extends Function<T, Publisher<Boolean>> {
default AsyncPredicate<T> and(AsyncPredicate<? super T> 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<T> negate() {
return t -> Mono.from(apply(t)).map(b -> !b);
}
default AsyncPredicate<T> or(AsyncPredicate<? super T> other) {
Assert.notNull(other, "other must not be null");
return t -> Flux.zip(apply(t), other.apply(t))
.map(tuple -> tuple.getT1() || tuple.getT2());
}
}

View File

@@ -0,0 +1,207 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.support;
import java.nio.charset.StandardCharsets;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.rsocket.util.NumberUtils;
import org.springframework.core.style.ToStringCreator;
import org.springframework.util.Assert;
public class Metadata {
/**
* Mime type of routing extension.
*/
public static final String ROUTING_MIME_TYPE = "message/x.rsocket.routing.v0";
/**
* The logical name.
*/
private final String name;
/**
* Keys and values associated with name.
*/
private final Map<String, String> properties;
public Metadata(String name, Map<String, String> properties) {
this.name = name;
this.properties = properties;
}
public String getName() {
return this.name;
}
public Map<String, String> getProperties() {
return this.properties;
}
public String get(String key) {
return this.properties.get(key);
}
public String put(String key, String value) {
return this.properties.put(key, value);
}
@Override
public String toString() {
return new ToStringCreator(this).append("name", name)
.append("properties", properties).toString();
}
public static Builder from(String name) {
return new Builder(name);
}
public static ByteBuf encode(Metadata metadata) {
return encode(ByteBufAllocator.DEFAULT, metadata);
}
public static ByteBuf encode(ByteBufAllocator allocator, Metadata metadata) {
return encode(allocator, metadata.getName(), metadata.getProperties());
}
public static ByteBuf encode(String name, Map<String, String> properties) {
return encode(ByteBufAllocator.DEFAULT, name, properties);
}
public static ByteBuf encode(ByteBufAllocator allocator, String name,
Map<String, String> properties) {
Assert.hasText(name, "name may not be empty");
Assert.notNull(properties, "properties may not be null");
Assert.notNull(allocator, "allocator may not be null");
ByteBuf byteBuf = allocator.buffer();
encodeString(byteBuf, name);
properties.entrySet().stream().forEach(entry -> {
encodeString(byteBuf, entry.getKey());
encodeString(byteBuf, entry.getValue());
});
return byteBuf;
}
private static void encodeString(ByteBuf byteBuf, String s) {
int length = NumberUtils.requireUnsignedByte(ByteBufUtil.utf8Bytes(s));
byteBuf.writeByte(length);
ByteBufUtil.reserveAndWriteUtf8(byteBuf, s, length);
}
public static Metadata decodeMetadata(ByteBuf byteBuf) {
AtomicInteger offset = new AtomicInteger(0);
String name = decodeString(byteBuf, offset);
Map<String, String> properties = new LinkedHashMap<>();
while (offset.get() < byteBuf.readableBytes()) { // TODO: What is the best
// conditional here?
String key = decodeString(byteBuf, offset);
String value = null;
if (offset.get() < byteBuf.readableBytes()) {
value = decodeString(byteBuf, offset);
}
properties.put(key, value);
}
return new Metadata(name, properties);
}
private static String decodeString(ByteBuf byteBuf, AtomicInteger offset) {
int length = byteBuf.getByte(offset.get());
int index = offset.addAndGet(Byte.BYTES);
String s = byteBuf.toString(index, length, StandardCharsets.UTF_8);
offset.addAndGet(length);
return s;
}
public boolean matches(Metadata other) {
if (other == null) {
return false;
}
if (other.getName() == null) {
return false;
}
if (!getName().equalsIgnoreCase(other.getName())) {
return false;
}
return matches(getProperties(), other.getProperties());
}
/**
* Matches leftMetadata to rightMetadata. rightMetadata must contain all key with
* equal values (ignoring case) of leftMetadata.
* @param leftMetadata first metadata to compare.
* @param rightMetadata second metadata to compare.
* @return true if all keys and values (case-insensitive) from leftMetadata are in
* rightMetadata.
*/
// TODO: find a way to make this more performant
public static boolean matches(Map<String, String> leftMetadata,
Map<String, String> rightMetadata) {
if (leftMetadata == null || rightMetadata == null) {
return false;
}
for (Map.Entry<String, String> entry : leftMetadata.entrySet()) {
String enrichedValue = rightMetadata.get(entry.getKey());
if (enrichedValue == null ||
// TODO: regex and possibly SpEL?
!enrichedValue.equalsIgnoreCase(entry.getValue())) {
return false;
}
}
// all entries in metadata exist and match corresponding entries in
// enriched.metadata
return true;
}
public static class Builder {
private final Metadata metadata;
public Builder(String name) {
Assert.hasText(name, "Name must not be empty.");
this.metadata = new Metadata(name, new LinkedHashMap<>());
}
public Builder with(String key, String value) {
this.metadata.put(key, value);
return this;
}
public Metadata build() {
return this.metadata;
}
public ByteBuf encode() {
return Metadata.encode(build());
}
}
}

View File

@@ -0,0 +1,3 @@
# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketAutoConfiguration

View File

@@ -0,0 +1,54 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.autoconfigure;
import org.junit.Test;
import org.springframework.boot.actuate.autoconfigure.metrics.CompositeMeterRegistryAutoConfiguration;
import org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ReactiveWebApplicationContextRunner;
import org.springframework.cloud.gateway.rsocket.registry.Registry;
import org.springframework.cloud.gateway.rsocket.registry.RegistryRoutes;
import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.server.GatewayRSocketServer;
import org.springframework.cloud.gateway.rsocket.socketacceptor.GatewaySocketAcceptor;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicate;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter;
import static org.assertj.core.api.Assertions.assertThat;
public class GatewayRSocketAutoConfigurationTests {
@Test
public void gatewayRSocketConfigured() {
new ReactiveWebApplicationContextRunner()
.withConfiguration(
AutoConfigurations.of(GatewayRSocketAutoConfiguration.class,
CompositeMeterRegistryAutoConfiguration.class,
MetricsAutoConfiguration.class))
.run(context -> assertThat(context).hasSingleBean(Registry.class)
.hasSingleBean(RegistryRoutes.class)
.hasSingleBean(RegistrySocketAcceptorFilter.class)
.hasSingleBean(GatewayRSocketServer.class)
.hasSingleBean(GatewayRSocketProperties.class)
.hasSingleBean(GatewaySocketAcceptor.class)
.hasSingleBean(SocketAcceptorPredicateFilter.class)
.doesNotHaveBean(SocketAcceptorPredicate.class));
}
}

View File

@@ -0,0 +1,81 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import reactor.test.StepVerifier;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.test.PingPongApp;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.SocketUtils;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = PingPongApp.class, properties = {
"ping.take=5" }, webEnvironment = WebEnvironment.RANDOM_PORT)
public class GatewayRSocketIntegrationTests {
private static int port;
@Autowired
private PingPongApp.Ping ping;
@Autowired
private PingPongApp.Pong pong;
@Autowired
private GatewayRSocketProperties properties;
@Autowired
private PingPongApp.MySocketAcceptorFilter mySocketAcceptorFilter;
@Autowired
private GatewayRSocketServer server;
@BeforeClass
public static void init() {
port = SocketUtils.findAvailableTcpPort();
System.setProperty("spring.cloud.gateway.rsocket.server.port",
String.valueOf(port));
}
@AfterClass
public static void after() {
System.clearProperty("spring.cloud.gateway.rsocket.server.port");
}
@Test
public void contextLoads() {
StepVerifier.create(ping.getPongFlux()).expectSubscription()
.then(() -> server.stop()).thenConsumeWhile(s -> true).verifyComplete();
assertThat(ping.getPongsReceived()).isGreaterThan(0);
assertThat(pong.getPingsReceived()).isGreaterThan(0);
assertThat(properties.getServer().getPort()).isNotEqualTo(7002);
assertThat(mySocketAcceptorFilter.invoked()).isTrue();
assertThat(server.isRunning()).isFalse();
}
}

View File

@@ -0,0 +1,278 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.server;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.function.Function;
import io.micrometer.core.instrument.Tags;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.netty.buffer.Unpooled;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.test.StepVerifier;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket;
import org.springframework.cloud.gateway.rsocket.registry.LoadBalancedRSocket.EnrichedRSocket;
import org.springframework.cloud.gateway.rsocket.registry.Registry;
import org.springframework.cloud.gateway.rsocket.route.Route;
import org.springframework.cloud.gateway.rsocket.route.Routes;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Rossen Stoyanchev
*/
public class GatewayRSocketTests {
private static Log logger = LogFactory.getLog(GatewayRSocketTests.class);
private Registry registry;
private Payload incomingPayload;
// TODO: add tests for metrics and other request types
@Before
public void init() {
registry = mock(Registry.class);
incomingPayload = DefaultPayload.create(Unpooled.EMPTY_BUFFER,
Metadata.from("mock").with("id", "mock1").encode());
RSocket rSocket = mock(RSocket.class);
LoadBalancedRSocket loadBalancedRSocket = mock(LoadBalancedRSocket.class);
when(registry.getRegistered(any(Metadata.class))).thenReturn(loadBalancedRSocket);
Mono<EnrichedRSocket> mono = Mono
.just(new EnrichedRSocket(rSocket, getMetadata()));
when(loadBalancedRSocket.choose()).thenReturn(mono);
when(rSocket.requestResponse(any(Payload.class)))
.thenReturn(Mono.just(DefaultPayload.create("response")));
}
@Test
public void multipleFilters() {
TestFilter filter1 = new TestFilter();
TestFilter filter2 = new TestFilter();
TestFilter filter3 = new TestFilter();
Payload payload = new TestGatewayRSocket(registry,
new TestRoutes(filter1, filter2, filter3))
.requestResponse(incomingPayload).block(Duration.ZERO);
assertThat(filter1.invoked()).isTrue();
assertThat(filter2.invoked()).isTrue();
assertThat(filter3.invoked()).isTrue();
assertThat(payload).isNotNull();
}
@Test
public void zeroFilters() {
Payload payload = new TestGatewayRSocket(registry, new TestRoutes())
.requestResponse(incomingPayload).block(Duration.ZERO);
assertThat(payload).isNotNull();
}
@Test
public void shortcircuitFilter() {
TestFilter filter1 = new TestFilter();
ShortcircuitingFilter filter2 = new ShortcircuitingFilter();
TestFilter filter3 = new TestFilter();
TestGatewayRSocket gatewayRSocket = new TestGatewayRSocket(registry,
new TestRoutes(filter1, filter2, filter3));
Mono<Payload> response = gatewayRSocket.requestResponse(incomingPayload);
// a false filter will create a pending rsocket that blocks forever
// this tweaks the rsocket to compelte.
gatewayRSocket.processor.onNext(null);
StepVerifier.withVirtualTime(() -> response).expectSubscription()
.verifyComplete();
assertThat(filter1.invoked()).isTrue();
assertThat(filter2.invoked()).isTrue();
assertThat(filter3.invoked()).isFalse();
}
@Test
public void asyncFilter() {
AsyncFilter filter = new AsyncFilter();
Payload payload = new TestGatewayRSocket(registry, new TestRoutes(filter))
.requestResponse(incomingPayload).block(Duration.ofSeconds(5));
assertThat(filter.invoked()).isTrue();
assertThat(payload).isNotNull();
}
// TODO: add exception handlers?
@Test(expected = IllegalStateException.class)
public void handleErrorFromFilter() {
ExceptionFilter filter = new ExceptionFilter();
new TestGatewayRSocket(registry, new TestRoutes(filter))
.requestResponse(incomingPayload).block(Duration.ofSeconds(5));
// assertNull(socket);
}
private static Metadata getMetadata() {
return Metadata.from("service").with("id", "service1").build();
}
private static class TestGatewayRSocket extends GatewayRSocket {
private final MonoProcessor<RSocket> processor = MonoProcessor.create();
TestGatewayRSocket(Registry registry, Routes routes) {
super(registry, routes, new SimpleMeterRegistry(),
new GatewayRSocketProperties(), getMetadata());
}
@Override
PendingRequestRSocket constructPendingRSocket(GatewayExchange exchange) {
Function<Registry.RegisteredEvent, Mono<Route>> routeFinder = registeredEvent -> getRouteMono(
registeredEvent, exchange);
return new PendingRequestRSocket(routeFinder, map -> {
Tags tags = exchange.getTags().and("responder.id", map.get("id"));
exchange.setTags(tags);
}, processor);
}
public MonoProcessor<RSocket> getProcessor() {
return processor;
}
}
private static class TestRoutes implements Routes {
private final Route route;
private List<GatewayFilter> filters;
TestRoutes() {
this(Collections.emptyList());
}
TestRoutes(GatewayFilter... filters) {
this(Arrays.asList(filters));
}
TestRoutes(List<GatewayFilter> filters) {
this.filters = filters;
route = Route.builder().id("route1")
.routingMetadata(Metadata.from("mock").build())
.predicate(exchange -> Mono.just(true)).filters(filters).build();
}
@Override
public Flux<Route> getRoutes() {
return Flux.just(route);
}
}
private static class TestFilter implements GatewayFilter {
private volatile boolean invoked;
public boolean invoked() {
return this.invoked;
}
@Override
public Mono<Success> filter(GatewayExchange exchange, GatewayFilterChain chain) {
this.invoked = true;
return doFilter(exchange, chain);
}
public Mono<Success> doFilter(GatewayExchange exchange,
GatewayFilterChain chain) {
return chain.filter(exchange);
}
}
private static class ShortcircuitingFilter extends TestFilter {
@Override
public Mono<Success> doFilter(GatewayExchange exchange,
GatewayFilterChain chain) {
return Mono.empty();
}
}
private static class AsyncFilter extends TestFilter {
@Override
public Mono<Success> doFilter(GatewayExchange exchange,
GatewayFilterChain chain) {
return doAsyncWork().flatMap(asyncResult -> {
logger.debug("Async result: " + asyncResult);
return chain.filter(exchange);
});
}
private Mono<String> doAsyncWork() {
return Mono.delay(Duration.ofMillis(100L)).map(l -> "123");
}
}
private static class ExceptionFilter implements GatewayFilter {
@Override
public Mono<Success> filter(GatewayExchange exchange, GatewayFilterChain chain) {
return Mono.error(new IllegalStateException("boo"));
}
}
/*
* private static class TestExceptionHandler implements WebExceptionHandler {
*
* private Throwable ex;
*
* @Override public Mono<Void> handle(GatewayExchange exchange, Throwable ex) {
* this.ex = ex; return Mono.error(ex); } }
*/
}

View File

@@ -0,0 +1,212 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.RSocket;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Before;
import org.junit.Test;
import reactor.core.publisher.Mono;
import org.springframework.cloud.gateway.rsocket.autoconfigure.GatewayRSocketProperties;
import org.springframework.cloud.gateway.rsocket.server.GatewayRSocket;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import static java.util.Collections.singletonList;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
/**
* @author Rossen Stoyanchev
*/
public class GatewaySocketAcceptorTests {
private static Log logger = LogFactory.getLog(GatewaySocketAcceptorTests.class);
private GatewayRSocket.Factory factory;
private ConnectionSetupPayload setupPayload;
private RSocket sendingSocket;
private MeterRegistry meterRegistry;
private GatewayRSocketProperties properties = new GatewayRSocketProperties();
@Before
public void init() {
this.factory = mock(GatewayRSocket.Factory.class);
this.setupPayload = mock(ConnectionSetupPayload.class);
this.sendingSocket = mock(RSocket.class);
this.meterRegistry = new SimpleMeterRegistry();
when(this.factory.create(any(Metadata.class)))
.thenReturn(mock(GatewayRSocket.class));
}
// TODO: test metrics
@Test
public void multipleFilters() {
TestFilter filter1 = new TestFilter();
TestFilter filter2 = new TestFilter();
TestFilter filter3 = new TestFilter();
RSocket socket = new GatewaySocketAcceptor(this.factory,
Arrays.asList(filter1, filter2, filter3), this.meterRegistry,
this.properties).accept(this.setupPayload, this.sendingSocket)
.block(Duration.ZERO);
assertThat(filter1.invoked()).isTrue();
assertThat(filter2.invoked()).isTrue();
assertThat(filter3.invoked()).isTrue();
assertThat(socket).isNotNull();
}
@Test
public void zeroFilters() {
RSocket socket = new GatewaySocketAcceptor(this.factory, Collections.emptyList(),
this.meterRegistry, this.properties)
.accept(this.setupPayload, this.sendingSocket)
.block(Duration.ZERO);
assertThat(socket).isNotNull();
}
@Test
public void shortcircuitFilter() {
TestFilter filter1 = new TestFilter();
ShortcircuitingFilter filter2 = new ShortcircuitingFilter();
TestFilter filter3 = new TestFilter();
RSocket socket = new GatewaySocketAcceptor(this.factory,
Arrays.asList(filter1, filter2, filter3), this.meterRegistry,
this.properties).accept(this.setupPayload, this.sendingSocket)
.block(Duration.ZERO);
assertThat(filter1.invoked()).isTrue();
assertThat(filter2.invoked()).isTrue();
assertThat(filter3.invoked()).isFalse();
assertThat(socket).isNull();
}
@Test
public void asyncFilter() {
AsyncFilter filter = new AsyncFilter();
RSocket socket = new GatewaySocketAcceptor(this.factory, singletonList(filter),
this.meterRegistry, this.properties)
.accept(this.setupPayload, this.sendingSocket)
.block(Duration.ofSeconds(5));
assertThat(filter.invoked()).isTrue();
assertThat(socket).isNotNull();
}
// TODO: add exception handlers?
@Test(expected = IllegalStateException.class)
public void handleErrorFromFilter() {
ExceptionFilter filter = new ExceptionFilter();
new GatewaySocketAcceptor(this.factory, singletonList(filter), this.meterRegistry,
this.properties).accept(this.setupPayload, this.sendingSocket)
.block(Duration.ofSeconds(5));
}
private static class TestFilter implements SocketAcceptorFilter {
private volatile boolean invoked;
public boolean invoked() {
return this.invoked;
}
@Override
public Mono<Success> filter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
this.invoked = true;
return doFilter(exchange, chain);
}
public Mono<Success> doFilter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
return chain.filter(exchange);
}
}
private static class ShortcircuitingFilter extends TestFilter {
@Override
public Mono<Success> doFilter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
return Mono.empty();
}
}
private static class AsyncFilter extends TestFilter {
@Override
public Mono<Success> doFilter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
return doAsyncWork().flatMap(asyncResult -> {
logger.debug("Async result: " + asyncResult);
return chain.filter(exchange);
});
}
private Mono<String> doAsyncWork() {
return Mono.delay(Duration.ofMillis(100L)).map(l -> "123");
}
}
private static class ExceptionFilter implements SocketAcceptorFilter {
@Override
public Mono<Success> filter(SocketAcceptorExchange exchange,
SocketAcceptorFilterChain chain) {
return Mono.error(new IllegalStateException("boo"));
}
}
/*
* private static class TestExceptionHandler implements WebExceptionHandler {
*
* private Throwable ex;
*
* @Override public Mono<Void> handle(SocketAcceptorExchange exchange, Throwable ex) {
* this.ex = ex; return Mono.error(ex); } }
*/
}

View File

@@ -0,0 +1,131 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.socketacceptor;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.RSocket;
import org.junit.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import org.springframework.cloud.gateway.rsocket.filter.RSocketFilter.Success;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
public class SocketAcceptorPredicateFilterTests {
@Test
public void noPredicateWorks() {
Mono<Success> result = runFilter(Collections.emptyList());
StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
}
@Test
public void singleTruePredicateWorks() {
TestPredicate predicate = new TestPredicate(true);
Mono<Success> result = runFilter(predicate);
StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
assertThat(predicate.invoked()).isTrue();
}
@Test
public void singleFalsePredicateWorks() {
TestPredicate predicate = new TestPredicate(false);
Mono<Success> 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<Success> 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<Success> 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<Success> result = runFilter(truePredicate, truePredicate2);
StepVerifier.create(result).expectNext(Success.INSTANCE).verifyComplete();
assertThat(truePredicate.invoked()).isTrue();
assertThat(truePredicate2.invoked()).isTrue();
}
private Mono<Success> runFilter(SocketAcceptorPredicate predicate) {
return runFilter(Collections.singletonList(predicate));
}
private Mono<Success> runFilter(SocketAcceptorPredicate... predicates) {
return runFilter(Arrays.asList(predicates));
}
private Mono<Success> runFilter(List<SocketAcceptorPredicate> 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<Boolean> test;
TestPredicate(boolean value) {
test = Mono.just(value);
}
@Override
public Publisher<Boolean> apply(SocketAcceptorExchange exchange) {
invoked = true;
return test;
}
public boolean invoked() {
return invoked;
}
}
}

View File

@@ -0,0 +1,85 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.support;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.IntStream;
import io.netty.buffer.ByteBuf;
import org.junit.Test;
import org.springframework.util.Assert;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.cloud.gateway.rsocket.support.Metadata.matches;
public class MetadataTests {
@Test
public void encodeAndDecodeJustName() {
ByteBuf byteBuf = Metadata.from("test").encode();
assertMetadata(byteBuf, "test");
}
@Test
public void encodeAndDecodeWorks() {
ByteBuf byteBuf = Metadata.from("test1").with("key1111", "val111111")
.with("key22", "val222").encode();
Metadata metadata = assertMetadata(byteBuf, "test1");
Map<String, String> properties = metadata.getProperties();
assertThat(properties).hasSize(2).containsOnlyKeys("key1111", "key22")
.containsValues("val111111", "val222");
}
private Metadata assertMetadata(ByteBuf byteBuf, String name) {
Metadata metadata = Metadata.decodeMetadata(byteBuf);
assertThat(metadata).isNotNull();
assertThat(metadata.getName()).isEqualTo(name);
return metadata;
}
@Test
public void nullMetadataDoesNotMatch() {
assertThat(matches(null, new HashMap<>())).isFalse();
assertThat(matches(new HashMap<>(), null)).isFalse();
}
@Test
public void metadataSubsetMatches() {
assertThat(matches(metadata(2), metadata(3))).isTrue();
}
@Test
public void metadataEqualSetMatches() {
assertThat(matches(metadata(3), metadata(3))).isTrue();
}
@Test
public void metadataSuperSetDoesNotMatch() {
assertThat(matches(metadata(3), metadata(2))).isFalse();
}
private Map<String, String> metadata(int size) {
Assert.isTrue(size > 0, "size must be > 0");
HashMap<String, String> metadata = new HashMap<>();
IntStream.rangeClosed(1, size).forEach(i -> metadata.put("key" + i, "val" + i));
return metadata;
}
}

View File

@@ -0,0 +1,293 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.test;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.micrometer.MicrometerRSocketInterceptor;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import io.rsocket.util.RSocketProxy;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.cloud.gateway.rsocket.server.GatewayExchange;
import org.springframework.cloud.gateway.rsocket.server.GatewayFilter;
import org.springframework.cloud.gateway.rsocket.server.GatewayFilterChain;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorExchange;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilterChain;
import org.springframework.cloud.gateway.rsocket.support.Metadata;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
import org.springframework.core.Ordered;
import org.springframework.core.env.ConfigurableEnvironment;
import static io.netty.buffer.Unpooled.EMPTY_BUFFER;
@SpringBootApplication
public class PingPongApp {
@Bean
public Ping ping1() {
return new Ping("1");
}
@Bean
@ConditionalOnProperty("ping.two.enabled")
public Ping ping2() {
return new Ping("2");
}
@Bean
public Pong pong() {
return new Pong();
}
@Bean
public MySocketAcceptorFilter mySocketAcceptorFilter() {
return new MySocketAcceptorFilter();
}
public static void main(String[] args) {
SpringApplication.run(PingPongApp.class, args);
}
static String reply(String in) {
if (in.length() > 4) {
in = in.substring(0, 4);
}
switch (in.toLowerCase()) {
case "ping":
return "pong";
case "pong":
return "ping";
default:
throw new IllegalArgumentException("Value must be ping or pong, not " + in);
}
}
@Slf4j
public static class Ping
implements Ordered, ApplicationListener<ApplicationReadyEvent> {
@Autowired
private MeterRegistry meterRegistry;
private final String id;
private final AtomicInteger pongsReceived = new AtomicInteger();
private Flux<String> pongFlux;
public Ping(String id) {
this.id = id;
}
@Override
public int getOrder() {
return 0;
}
@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
log.info("Starting Ping" + id);
ConfigurableEnvironment env = event.getApplicationContext().getEnvironment();
Integer take = env.getProperty("ping.take", Integer.class, null);
Integer gatewayPort = env.getProperty(
"spring.cloud.gateway.rsocket.server.port", Integer.class, 7002);
log.debug("ping.take: " + take);
MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
meterRegistry, Tag.of("component", "ping"));
ByteBuf announcementMetadata = Metadata.from("ping").with("id", "ping" + id)
.encode();
pongFlux = RSocketFactory.connect()
.metadataMimeType(Metadata.ROUTING_MIME_TYPE)
.setupPayload(
DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
.addClientPlugin(interceptor)
.transport(TcpClientTransport.create(gatewayPort)) // proxy
.start().flatMapMany(socket -> {
Flux<String> pong = socket.requestChannel(
Flux.interval(Duration.ofSeconds(1)).map(i -> {
ByteBuf data = ByteBufUtil.writeUtf8(
ByteBufAllocator.DEFAULT, "ping" + id);
ByteBuf routingMetadata = Metadata.from("pong")
.encode();
return DefaultPayload.create(data, routingMetadata);
}).onBackpressureDrop(payload -> log.debug(
"Dropped payload " + payload.getDataUtf8())) // this
// is
// needed
// in
// case
// pong
// is
// not
// available
// yet
).map(Payload::getDataUtf8).doOnNext(str -> {
int received = pongsReceived.incrementAndGet();
log.info("received " + str + "(" + received + ") in Ping"
+ id);
}).doFinally(signal -> socket.dispose());
if (take != null) {
return pong.take(take);
}
return pong;
});
pongFlux.subscribe();
}
public Flux<String> getPongFlux() {
return pongFlux;
}
public int getPongsReceived() {
return pongsReceived.get();
}
}
@Slf4j
public static class Pong
implements Ordered, ApplicationListener<ApplicationReadyEvent> {
@Autowired
private MeterRegistry meterRegistry;
private final AtomicInteger pingsReceived = new AtomicInteger();
@Override
public int getOrder() {
return 1;
}
@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
ConfigurableEnvironment env = event.getApplicationContext().getEnvironment();
Integer pongDelay = env.getProperty("pong.delay", Integer.class, 5000);
try {
Thread.sleep(pongDelay);
}
catch (InterruptedException e) {
e.printStackTrace();
}
log.info("Starting Pong");
Integer gatewayPort = env.getProperty(
"spring.cloud.gateway.rsocket.server.port", Integer.class, 7002);
MicrometerRSocketInterceptor interceptor = new MicrometerRSocketInterceptor(
meterRegistry, Tag.of("component", "pong"));
ByteBuf announcementMetadata = Metadata.from("pong").with("id", "pong1")
.encode();
RSocketFactory.connect().metadataMimeType(Metadata.ROUTING_MIME_TYPE)
.setupPayload(
DefaultPayload.create(EMPTY_BUFFER, announcementMetadata))
.addClientPlugin(interceptor).acceptor(this::accept)
.transport(TcpClientTransport.create(gatewayPort)) // proxy
.start().block();
}
@SuppressWarnings("Duplicates")
RSocket accept(RSocket rSocket) {
RSocket pong = new RSocketProxy(rSocket) {
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return Flux.from(payloads).map(Payload::getDataUtf8).doOnNext(str -> {
int received = pingsReceived.incrementAndGet();
log.info("received " + str + "(" + received + ") in Pong");
}).map(PingPongApp::reply).map(reply -> {
ByteBuf data = ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT,
reply);
ByteBuf routingMetadata = Metadata.from("ping").encode();
return DefaultPayload.create(data, routingMetadata);
});
}
};
return pong;
}
public int getPingsReceived() {
return pingsReceived.get();
}
}
@Slf4j
public static class MyGatewayFilter implements GatewayFilter {
private AtomicBoolean invoked = new AtomicBoolean(false);
@Override
public Mono<Success> 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<Success> 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();
}
}
}

View File

@@ -0,0 +1,49 @@
/*
* Copyright 2018-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.gateway.rsocket.test;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.springframework.cloud.gateway.rsocket.registry.Registry;
import org.springframework.cloud.gateway.rsocket.registry.RegistrySocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorFilter;
import org.springframework.cloud.gateway.rsocket.socketacceptor.SocketAcceptorPredicateFilter;
import org.springframework.core.OrderComparator;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
public class SocketAcceptorFilterOrderTests {
@Test
public void predicateFilterAfterRegistryFilter() {
SocketAcceptorFilter predicateFilter = new SocketAcceptorPredicateFilter(
Collections.emptyList());
SocketAcceptorFilter registryFilter = new RegistrySocketAcceptorFilter(
mock(Registry.class));
List<SocketAcceptorFilter> filters = Arrays.asList(predicateFilter,
registryFilter);
OrderComparator.sort(filters);
assertThat(filters).containsExactly(registryFilter, predicateFilter);
}
}

View File

@@ -0,0 +1,9 @@
logging:
level:
org.springframework.cloud.gateway.rsocket: DEBUG
management:
endpoints:
web:
exposure:
include: '*'

View File

@@ -3,6 +3,7 @@
"-//Puppy Crawl//DTD Suppressions 1.1//EN"
"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress files=".*AbstractFilterChain\.java" checks="LineLengthCheck"/>
<suppress files=".*AbstractNameValueGatewayFilterFactory\.java" checks="LineLengthCheck"/>
<suppress files=".*AdhocTestSuite\.java" checks="LineLengthCheck"/>
<suppress files=".*CloudFoundryRouteServiceRoutePredicateFactory\.java" checks="JavadocTagContinuationIndentationCheck"/>