Moving rabbit binder out of SCSt main repo

This commit is contained in:
Eric Bottard
2016-06-02 17:58:12 +02:00
committed by Ilayaperumal Gopinathan
parent 77f1f80655
commit 4f841f59f5
33 changed files with 4010 additions and 0 deletions

24
.gitignore vendored Normal file
View File

@@ -0,0 +1,24 @@
/application.yml
/application.properties
asciidoctor.css
*~
.#*
*#
target/
build/
bin/
_site/
.classpath
.project
.settings
.springBeans
.DS_Store
*.sw*
*.iml
*.ipr
*.iws
.idea/*
.factorypath
dump.rdb
.apt_generated
artifacts

1
.mvn/jvm.config Normal file
View File

@@ -0,0 +1 @@
-Xmx1024m -XX:CICompilerCount=1 -XX:TieredStopAtLevel=1 -Djava.security.egd=file:/dev/./urandom

1
.mvn/maven.config Normal file
View File

@@ -0,0 +1 @@
-DaltSnapshotDeploymentRepository=repo.spring.io::default::https://repo.spring.io/libs-snapshot-local -P spring

BIN
.mvn/wrapper/maven-wrapper.jar vendored Normal file

Binary file not shown.

1
.mvn/wrapper/maven-wrapper.properties vendored Normal file
View File

@@ -0,0 +1 @@
distributionUrl=https://repo1.maven.org/maven2/org/apache/maven/apache-maven/3.3.3/apache-maven-3.3.3-bin.zip

66
.settings.xml Normal file
View File

@@ -0,0 +1,66 @@
<?xml version="1.0" encoding="UTF-8"?>
<settings>
<servers>
<server>
<id>repo.spring.io</id>
<username>${env.CI_DEPLOY_USERNAME}</username>
<password>${env.CI_DEPLOY_PASSWORD}</password>
</server>
</servers>
<profiles>
<profile>
<!--
N.B. this profile is only here to support users and IDEs that do not use Maven 3.3.
It isn't needed on the command line if you use the wrapper script (mvnw) or if you use
a native Maven with the right version. Eclipse users should points their Maven tooling to
this settings file, or copy the profile into their ~/.m2/settings.xml.
-->
<id>spring</id>
<activation><activeByDefault>true</activeByDefault></activation>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>http://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</profile>
</profiles>
</settings>

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.

3
README.adoc Normal file
View File

@@ -0,0 +1,3 @@
# spring-cloud-stream-binder-rabbit
Spring Cloud Stream Binder implementation for Rabbit

234
mvnw vendored Executable file
View File

@@ -0,0 +1,234 @@
#!/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
#
# Required ENV vars:
# ------------------
# JAVA_HOME - location of a JDK home dir
#
# Optional ENV vars
# -----------------
# M2_HOME - location of maven2's installed home dir
# MAVEN_OPTS - parameters passed to the Java VM when running Maven
# e.g. to debug Maven itself, use
# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
# MAVEN_SKIP_RC - flag to disable loading of mavenrc files
# ----------------------------------------------------------------------------
if [ -z "$MAVEN_SKIP_RC" ] ; then
if [ -f /etc/mavenrc ] ; then
. /etc/mavenrc
fi
if [ -f "$HOME/.mavenrc" ] ; then
. "$HOME/.mavenrc"
fi
fi
# OS specific support. $var _must_ be set to either true or false.
cygwin=false;
darwin=false;
mingw=false
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
;;
esac
if [ -z "$JAVA_HOME" ] ; then
if [ -r /etc/gentoo-release ] ; then
JAVA_HOME=`java-config --jre-home`
fi
fi
if [ -z "$M2_HOME" ] ; then
## resolve links - $0 may be a link to maven's home
PRG="$0"
# need this for relative symlinks
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG="`dirname "$PRG"`/$link"
fi
done
saveddir=`pwd`
M2_HOME=`dirname "$PRG"`/..
# make it fully qualified
M2_HOME=`cd "$M2_HOME" && pwd`
cd "$saveddir"
# echo Using m2 at $M2_HOME
fi
# For Cygwin, ensure paths are in UNIX format before anything is touched
if $cygwin ; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --unix "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
fi
# For Migwn, ensure paths are in UNIX format before anything is touched
if $mingw ; then
[ -n "$M2_HOME" ] &&
M2_HOME="`(cd "$M2_HOME"; pwd)`"
[ -n "$JAVA_HOME" ] &&
JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
# TODO classpath?
fi
if [ -z "$JAVA_HOME" ]; then
javaExecutable="`which javac`"
if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
# readlink(1) is not available as standard on Solaris 10.
readLink=`which readlink`
if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
if $darwin ; then
javaHome="`dirname \"$javaExecutable\"`"
javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
else
javaExecutable="`readlink -f \"$javaExecutable\"`"
fi
javaHome="`dirname \"$javaExecutable\"`"
javaHome=`expr "$javaHome" : '\(.*\)/bin'`
JAVA_HOME="$javaHome"
export JAVA_HOME
fi
fi
fi
if [ -z "$JAVACMD" ] ; then
if [ -n "$JAVA_HOME" ] ; then
if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
# IBM's JDK on AIX uses strange locations for the executables
JAVACMD="$JAVA_HOME/jre/sh/java"
else
JAVACMD="$JAVA_HOME/bin/java"
fi
else
JAVACMD="`which java`"
fi
fi
if [ ! -x "$JAVACMD" ] ; then
echo "Error: JAVA_HOME is not defined correctly." >&2
echo " We cannot execute $JAVACMD" >&2
exit 1
fi
if [ -z "$JAVA_HOME" ] ; then
echo "Warning: JAVA_HOME environment variable is not set."
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)
while [ "$wdir" != '/' ] ; do
if [ -d "$wdir"/.mvn ] ; then
basedir=$wdir
break
fi
wdir=$(cd "$wdir/.."; pwd)
done
echo "${basedir}"
}
# concatenates all lines of a file
concat_lines() {
if [ -f "$1" ]; then
echo "$(tr -s '\n' ' ' < "$1")"
fi
}
export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-$(find_maven_basedir)}
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
WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
exec "$JAVACMD" \
$MAVEN_OPTS \
-classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
"-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
${WRAPPER_LAUNCHER} "$@"

234
mvnw.cmd vendored Normal file
View File

@@ -0,0 +1,234 @@
#!/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
#
# Required ENV vars:
# ------------------
# JAVA_HOME - location of a JDK home dir
#
# Optional ENV vars
# -----------------
# M2_HOME - location of maven2's installed home dir
# MAVEN_OPTS - parameters passed to the Java VM when running Maven
# e.g. to debug Maven itself, use
# set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
# MAVEN_SKIP_RC - flag to disable loading of mavenrc files
# ----------------------------------------------------------------------------
if [ -z "$MAVEN_SKIP_RC" ] ; then
if [ -f /etc/mavenrc ] ; then
. /etc/mavenrc
fi
if [ -f "$HOME/.mavenrc" ] ; then
. "$HOME/.mavenrc"
fi
fi
# OS specific support. $var _must_ be set to either true or false.
cygwin=false;
darwin=false;
mingw=false
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
;;
esac
if [ -z "$JAVA_HOME" ] ; then
if [ -r /etc/gentoo-release ] ; then
JAVA_HOME=`java-config --jre-home`
fi
fi
if [ -z "$M2_HOME" ] ; then
## resolve links - $0 may be a link to maven's home
PRG="$0"
# need this for relative symlinks
while [ -h "$PRG" ] ; do
ls=`ls -ld "$PRG"`
link=`expr "$ls" : '.*-> \(.*\)$'`
if expr "$link" : '/.*' > /dev/null; then
PRG="$link"
else
PRG="`dirname "$PRG"`/$link"
fi
done
saveddir=`pwd`
M2_HOME=`dirname "$PRG"`/..
# make it fully qualified
M2_HOME=`cd "$M2_HOME" && pwd`
cd "$saveddir"
# echo Using m2 at $M2_HOME
fi
# For Cygwin, ensure paths are in UNIX format before anything is touched
if $cygwin ; then
[ -n "$M2_HOME" ] &&
M2_HOME=`cygpath --unix "$M2_HOME"`
[ -n "$JAVA_HOME" ] &&
JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
[ -n "$CLASSPATH" ] &&
CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
fi
# For Migwn, ensure paths are in UNIX format before anything is touched
if $mingw ; then
[ -n "$M2_HOME" ] &&
M2_HOME="`(cd "$M2_HOME"; pwd)`"
[ -n "$JAVA_HOME" ] &&
JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
# TODO classpath?
fi
if [ -z "$JAVA_HOME" ]; then
javaExecutable="`which javac`"
if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
# readlink(1) is not available as standard on Solaris 10.
readLink=`which readlink`
if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
if $darwin ; then
javaHome="`dirname \"$javaExecutable\"`"
javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
else
javaExecutable="`readlink -f \"$javaExecutable\"`"
fi
javaHome="`dirname \"$javaExecutable\"`"
javaHome=`expr "$javaHome" : '\(.*\)/bin'`
JAVA_HOME="$javaHome"
export JAVA_HOME
fi
fi
fi
if [ -z "$JAVACMD" ] ; then
if [ -n "$JAVA_HOME" ] ; then
if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
# IBM's JDK on AIX uses strange locations for the executables
JAVACMD="$JAVA_HOME/jre/sh/java"
else
JAVACMD="$JAVA_HOME/bin/java"
fi
else
JAVACMD="`which java`"
fi
fi
if [ ! -x "$JAVACMD" ] ; then
echo "Error: JAVA_HOME is not defined correctly." >&2
echo " We cannot execute $JAVACMD" >&2
exit 1
fi
if [ -z "$JAVA_HOME" ] ; then
echo "Warning: JAVA_HOME environment variable is not set."
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)
while [ "$wdir" != '/' ] ; do
if [ -d "$wdir"/.mvn ] ; then
basedir=$wdir
break
fi
wdir=$(cd "$wdir/.."; pwd)
done
echo "${basedir}"
}
# concatenates all lines of a file
concat_lines() {
if [ -f "$1" ]; then
echo "$(tr -s '\n' ' ' < "$1")"
fi
}
export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-$(find_maven_basedir)}
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
WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
exec "$JAVACMD" \
$MAVEN_OPTS \
-classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
"-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
${WRAPPER_LAUNCHER} "$@"

94
pom.xml Normal file
View File

@@ -0,0 +1,94 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
<packaging>pom</packaging>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build</artifactId>
<version>1.1.1.RELEASE</version>
<relativePath />
</parent>
<properties>
<spring-boot.version>1.4.0.BUILD-SNAPSHOT</spring-boot.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-dependencies</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<modules>
<module>spring-cloud-stream-binder-rabbit</module>
<module>spring-cloud-starter-stream-rabbit</module>
</modules>
<profiles>
<profile>
<id>spring</id>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
<repository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>http://repo.spring.io/release</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/libs-snapshot-local</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
<releases>
<enabled>false</enabled>
</releases>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/libs-milestone-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-releases</id>
<name>Spring Releases</name>
<url>http://repo.spring.io/libs-release-local</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</profile>
</profiles>
</project>

View File

@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<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-stream-binder-rabbit-parent</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
</parent>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
<description>Spring Cloud Starter Stream Rabbit</description>
<url>http://projects.spring.io/spring-cloud</url>
<organization>
<name>Pivotal Software, Inc.</name>
<url>http://www.spring.io</url>
</organization>
<properties>
<main.basedir>${basedir}/../..</main.basedir>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1 @@
provides: spring-cloud-starter-stream-rabbit

View File

@@ -0,0 +1,59 @@
<?xml version="1.0" encoding="UTF-8"?>
<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>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-binder-rabbit</name>
<description>RabbitMQ binder implementation</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit-parent</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-codec</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-core</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-amqp</artifactId>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,36 @@
/*
* Copyright 2015 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.stream.binder.rabbit;
/**
* Exceptions thrown while interfacing with the RabbitMQ admin plugin.
* @author Gary Russell
* @since 1.2
*/
@SuppressWarnings("serial")
public class RabbitAdminException extends RuntimeException {
public RabbitAdminException(String message, Throwable cause) {
super(message, cause);
}
public RabbitAdminException(String message) {
super(message);
}
}

View File

@@ -0,0 +1,195 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.BindingCleaner;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
/**
* Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for the {@code RabbitBinder}.
* @author Gary Russell
* @author David Turanski
* @since 1.2
*/
public class RabbitBindingCleaner implements BindingCleaner {
private static final Log logger = LogFactory.getLog(RabbitBindingCleaner.class);
private static final String PREFIX_DELIMITER = ".";
public static final String BINDER_PREFIX = "binder" + PREFIX_DELIMITER;
@Override
public Map<String, List<String>> clean(String entity, boolean isJob) {
return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX, entity, isJob);
}
public Map<String, List<String>> clean(String adminUri, String user, String pw, String vhost,
String binderPrefix, String entity, boolean isJob) {
return doClean(
adminUri == null ? "http://localhost:15672" : adminUri,
user == null ? "guest" : user,
pw == null ? "guest" : pw,
vhost == null ? "/" : vhost,
binderPrefix == null ? BINDER_PREFIX : binderPrefix,
entity, isJob);
}
private Map<String, List<String>> doClean(String adminUri, String user, String pw, String vhost,
String binderPrefix, String entity, boolean isJob) {
RestTemplate restTemplate = RabbitManagementUtils.buildRestTemplate(adminUri, user, pw);
List<String> removedQueues = isJob
? null
: findStreamQueues(adminUri, vhost, binderPrefix, entity, restTemplate);
List<String> removedExchanges = findExchanges(adminUri, vhost, binderPrefix, entity, restTemplate);
// Delete the queues in reverse order to enable re-running after a partial success.
// The queue search above starts with 0 and terminates on a not found.
for (int i = removedQueues.size() - 1; i >= 0; i--) {
String queueName = removedQueues.get(i);
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("queues", "{vhost}", "{stream}")
.buildAndExpand(vhost, queueName).encode().toUri();
restTemplate.delete(uri);
if (logger.isDebugEnabled()) {
logger.debug("deleted queue: " + queueName);
}
}
Map<String, List<String>> results = new HashMap<>();
if (removedQueues.size() > 0) {
results.put("queues", removedQueues);
}
// Fanout exchanges for taps
for (String exchange : removedExchanges) {
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}", "{name}")
.buildAndExpand(vhost, exchange).encode().toUri();
restTemplate.delete(uri);
if (logger.isDebugEnabled()) {
logger.debug("deleted exchange: " + exchange);
}
}
if (removedExchanges.size() > 0) {
results.put("exchanges", removedExchanges);
}
return results;
}
private List<String> findStreamQueues(String adminUri, String vhost, String binderPrefix, String stream,
RestTemplate restTemplate) {
String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream));
List<Map<String, Object>> queues = listAllQueues(adminUri, vhost, restTemplate);
List<String> removedQueues = new ArrayList<>();
for (Map<String, Object> queue : queues) {
String queueName = (String) queue.get("name");
if (queueName.startsWith(queueNamePrefix)) {
checkNoConsumers(queueName, queue);
removedQueues.add(queueName);
}
}
return removedQueues;
}
private List<Map<String, Object>> listAllQueues(String adminUri, String vhost, RestTemplate restTemplate) {
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("queues", "{vhost}")
.buildAndExpand(vhost).encode().toUri();
@SuppressWarnings("unchecked")
List<Map<String, Object>> queues = restTemplate.getForObject(uri, List.class);
return queues;
}
private String adjustPrefix(String prefix) {
if (prefix.endsWith("*")) {
return prefix.substring(0, prefix.length() - 1);
}
else {
return prefix + PREFIX_DELIMITER;
}
}
private void checkNoConsumers(String queueName, Map<String, Object> queue) {
if (!queue.get("consumers").equals(Integer.valueOf(0))) {
throw new RabbitAdminException("Queue " + queueName + " is in use");
}
}
@SuppressWarnings("unchecked")
private List<String> findExchanges(String adminUri, String vhost, String binderPrefix, String entity,
RestTemplate restTemplate) {
List<String> removedExchanges = new ArrayList<>();
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}")
.buildAndExpand(vhost).encode().toUri();
List<Map<String, Object>> exchanges = restTemplate.getForObject(uri, List.class);
String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity));
for (Map<String, Object> exchange : exchanges) {
String exchangeName = (String) exchange.get("name");
if (exchangeName.startsWith(exchangeNamePrefix)) {
uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings", "source")
.buildAndExpand(vhost, exchangeName).encode().toUri();
List<Map<String, Object>> bindings = restTemplate.getForObject(uri, List.class);
if (hasNoForeignBindings(bindings, exchangeNamePrefix)) {
uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings", "destination")
.buildAndExpand(vhost, exchangeName).encode().toUri();
bindings = restTemplate.getForObject(uri, List.class);
if (bindings.size() == 0) {
removedExchanges.add((String) exchange.get("name"));
}
else {
throw new RabbitAdminException("Cannot delete exchange " + exchangeName
+ "; it is a destination: " + bindings);
}
}
else {
throw new RabbitAdminException("Cannot delete exchange " + exchangeName + "; it has bindings: "
+ bindings);
}
}
}
return removedExchanges;
}
private boolean hasNoForeignBindings(List<Map<String, Object>> bindings, String exchangeNamePrefix) {
if (bindings.size() == 0) {
return true;
}
boolean noForeign = true;
for (Map<String, Object> binding : bindings) {
if (!("queue".equals(binding.get("destination_type")))
|| !((String) binding.get("destination")).startsWith(exchangeNamePrefix)) {
noForeign = false;
break;
}
}
return noForeign;
}
}

View File

@@ -0,0 +1,43 @@
/*
* Copyright 2016 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.stream.binder.rabbit;
/**
* @author Marius Bogoevici
*/
public class RabbitBindingProperties {
private RabbitConsumerProperties consumer = new RabbitConsumerProperties();
private RabbitProducerProperties producer = new RabbitProducerProperties();
public RabbitConsumerProperties getConsumer() {
return consumer;
}
public void setConsumer(RabbitConsumerProperties consumer) {
this.consumer = consumer;
}
public RabbitProducerProperties getProducer() {
return producer;
}
public void setProducer(RabbitProducerProperties producer) {
this.producer = producer;
}
}

View File

@@ -0,0 +1,162 @@
/*
* Copyright 2016 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.stream.binder.rabbit;
import javax.validation.constraints.Min;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.util.Assert;
/**
* @author Marius Bogoevici
*/
public class RabbitConsumerProperties {
private String prefix = "";
private boolean transacted;
private AcknowledgeMode acknowledgeMode = AcknowledgeMode.AUTO;
private int maxConcurrency = 1;
private int prefetch = 1;
private String[] requestHeaderPatterns = new String[] {"STANDARD_REQUEST_HEADERS", "*"};
private int txSize = 1;
private boolean autoBindDlq;
private boolean durableSubscription = true;
private boolean republishToDlq;
private boolean requeueRejected = true;
private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"};
private long recoveryInterval = 5000;
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public boolean isTransacted() {
return transacted;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
public AcknowledgeMode getAcknowledgeMode() {
return acknowledgeMode;
}
public void setAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
Assert.notNull("Acknowledge mode cannot be null");
this.acknowledgeMode = acknowledgeMode;
}
@Min(value = 1, message = "Max Concurrency should be greater than zero.")
public int getMaxConcurrency() {
return maxConcurrency;
}
public void setMaxConcurrency(int maxConcurrency) {
this.maxConcurrency = maxConcurrency;
}
@Min(value = 1, message = "Prefetch should be greater than zero.")
public int getPrefetch() {
return prefetch;
}
public void setPrefetch(int prefetch) {
this.prefetch = prefetch;
}
public String[] getRequestHeaderPatterns() {
return requestHeaderPatterns;
}
public void setRequestHeaderPatterns(String[] requestHeaderPatterns) {
this.requestHeaderPatterns = requestHeaderPatterns;
}
@Min(value = 1, message = "Tx Size should be greater than zero.")
public int getTxSize() {
return txSize;
}
public void setTxSize(int txSize) {
this.txSize = txSize;
}
public boolean isAutoBindDlq() {
return autoBindDlq;
}
public void setAutoBindDlq(boolean autoBindDlq) {
this.autoBindDlq = autoBindDlq;
}
public boolean isDurableSubscription() {
return durableSubscription;
}
public void setDurableSubscription(boolean durableSubscription) {
this.durableSubscription = durableSubscription;
}
public boolean isRepublishToDlq() {
return republishToDlq;
}
public void setRepublishToDlq(boolean republishToDlq) {
this.republishToDlq = republishToDlq;
}
public boolean isRequeueRejected() {
return requeueRejected;
}
public void setRequeueRejected(boolean requeueRejected) {
this.requeueRejected = requeueRejected;
}
public String[] getReplyHeaderPatterns() {
return replyHeaderPatterns;
}
public void setReplyHeaderPatterns(String[] replyHeaderPatterns) {
this.replyHeaderPatterns = replyHeaderPatterns;
}
public long getRecoveryInterval() {
return recoveryInterval;
}
public void setRecoveryInterval(long recoveryInterval) {
this.recoveryInterval = recoveryInterval;
}
}

View File

@@ -0,0 +1,60 @@
/*
* Copyright 2016 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.stream.binder.rabbit;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
/**
* @author Marius Bogoevici
*/
@ConfigurationProperties("spring.cloud.stream.rabbit")
public class RabbitExtendedBindingProperties implements ExtendedBindingProperties<RabbitConsumerProperties, RabbitProducerProperties> {
private Map<String, RabbitBindingProperties> bindings = new HashMap<>();
public Map<String, RabbitBindingProperties> getBindings() {
return bindings;
}
public void setBindings(Map<String, RabbitBindingProperties> bindings) {
this.bindings = bindings;
}
@Override
public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
if (bindings.containsKey(channelName) && bindings.get(channelName).getConsumer() != null) {
return bindings.get(channelName).getConsumer();
}
else {
return new RabbitConsumerProperties();
}
}
@Override
public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
if (bindings.containsKey(channelName) && bindings.get(channelName).getProducer() != null) {
return bindings.get(channelName).getProducer();
}
else {
return new RabbitProducerProperties();
}
}
}

View File

@@ -0,0 +1,83 @@
/*
* Copyright 2015 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.stream.binder.rabbit;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collections;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.AuthCache;
import org.apache.http.client.HttpClient;
import org.apache.http.client.protocol.HttpClientContext;
import org.apache.http.impl.auth.BasicScheme;
import org.apache.http.impl.client.BasicAuthCache;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.protocol.HttpContext;
import org.springframework.http.HttpMethod;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
import org.springframework.web.client.RestTemplate;
/**
* @author Gary Russell
* @since 1.2
*/
public abstract class RabbitManagementUtils {
public static RestTemplate buildRestTemplate(String adminUri, String user, String password) {
BasicCredentialsProvider credsProvider = new BasicCredentialsProvider();
credsProvider.setCredentials(
new AuthScope(AuthScope.ANY_HOST, AuthScope.ANY_PORT),
new UsernamePasswordCredentials(user, password));
HttpClient httpClient = HttpClients.custom().setDefaultCredentialsProvider(credsProvider).build();
// Set up pre-emptive basic Auth because the rabbit plugin doesn't currently support challenge/response for PUT
// Create AuthCache instance
AuthCache authCache = new BasicAuthCache();
// Generate BASIC scheme object and add it to the local; from the apache docs...
// auth cache
BasicScheme basicAuth = new BasicScheme();
URI uri;
try {
uri = new URI(adminUri);
}
catch (URISyntaxException e) {
throw new RabbitAdminException("Invalid URI", e);
}
authCache.put(new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme()), basicAuth);
// Add AuthCache to the execution context
final HttpClientContext localContext = HttpClientContext.create();
localContext.setAuthCache(authCache);
RestTemplate restTemplate = new RestTemplate(new HttpComponentsClientHttpRequestFactory(httpClient) {
@Override
protected HttpContext createHttpContext(HttpMethod httpMethod, URI uri) {
return localContext;
}
});
restTemplate.setMessageConverters(Collections.<HttpMessageConverter<?>>singletonList(
new MappingJackson2HttpMessageConverter()));
return restTemplate;
}
}

View File

@@ -0,0 +1,651 @@
/*
* Copyright 2013-2016 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.stream.binder.rabbit;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import org.aopalliance.aop.Advice;
import org.springframework.amqp.AmqpConnectException;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.DefaultBinding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.MessageValues;
import org.springframework.cloud.stream.binder.PartitionHandler;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ.
*
* @author Mark Fisher
* @author Gary Russell
* @author Jennifer Hickey
* @author Gunnar Hillert
* @author Ilayaperumal Gopinathan
* @author David Turanski
* @author Marius Bogoevici
*/
public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>> implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
= new AnonymousQueue.Base64UrlNamingStrategy("anonymous.");
private static final String DEAD_LETTER_EXCHANGE = "DLX";
private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
new DefaultMessagePropertiesConverter() {
@Override
public MessageProperties toMessageProperties(AMQP.BasicProperties source, Envelope envelope,
String charset) {
MessageProperties properties = super.toMessageProperties(source, envelope, charset);
properties.setDeliveryMode(null);
return properties;
}
};
private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private final RabbitAdmin rabbitAdmin;
private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
private final RabbitProperties rabbitProperties;
private ConnectionFactory connectionFactory;
private MessagePostProcessor decompressingPostProcessor = new DelegatingDecompressingPostProcessor();
private MessagePostProcessor compressingPostProcessor = new GZipPostProcessor();
private volatile String[] adminAddresses;
private volatile String[] nodes;
private volatile boolean clustered;
private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
Assert.notNull(rabbitProperties, "rabbitProperties must not be null");
this.connectionFactory = connectionFactory;
this.rabbitProperties = rabbitProperties;
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
this.autoDeclareContext.refresh();
this.rabbitAdmin.setApplicationContext(this.autoDeclareContext);
this.rabbitAdmin.setIgnoreDeclarationExceptions(true);
this.rabbitAdmin.afterPropertiesSet();
}
/**
* Set a {@link MessagePostProcessor} to decompress messages. Defaults to a
* {@link DelegatingDecompressingPostProcessor} with its default delegates.
* @param decompressingPostProcessor the post processor.
*/
public void setDecompressingPostProcessor(MessagePostProcessor decompressingPostProcessor) {
this.decompressingPostProcessor = decompressingPostProcessor;
}
/**
* Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages. Defaults to a
* {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}.
* @param compressingPostProcessor the post processor.
*/
public void setCompressingPostProcessor(MessagePostProcessor compressingPostProcessor) {
this.compressingPostProcessor = compressingPostProcessor;
}
public void setAdminAddresses(String[] adminAddresses) {
this.adminAddresses = Arrays.copyOf(adminAddresses, adminAddresses.length);
}
public void setNodes(String[] nodes) {
this.nodes = Arrays.copyOf(nodes, nodes.length);
this.clustered = nodes.length > 1;
}
public void setExtendedBindingProperties(RabbitExtendedBindingProperties extendedBindingProperties) {
this.extendedBindingProperties = extendedBindingProperties;
}
@Override
public void onInit() {
if (this.clustered) {
String[] addresses = StringUtils.commaDelimitedListToStringArray(this.rabbitProperties.getAddresses());
Assert.state(addresses.length == this.adminAddresses.length
&& addresses.length == this.nodes.length,
"'addresses', 'adminAddresses', and 'nodes' properties must have equal length");
this.connectionFactory = new LocalizedQueueConnectionFactory(this.connectionFactory, addresses,
this.adminAddresses, this.nodes, rabbitProperties.getVirtualHost(),
this.rabbitProperties.getUsername(), this.rabbitProperties.getPassword(),
this.rabbitProperties.getSsl().isEnabled(), this.rabbitProperties.getSsl().getKeyStore(),
this.rabbitProperties.getSsl().getTrustStore(), this.rabbitProperties.getSsl().getKeyStorePassword(),
this.rabbitProperties.getSsl().getTrustStorePassword());
}
}
@Override
public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
}
@Override
public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
return extendedBindingProperties.getExtendedProducerProperties(channelName);
}
@Override
public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
boolean anonymousConsumer = !StringUtils.hasText(group);
String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
: groupedName(name, group);
if (this.logger.isInfoEnabled()) {
this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
}
String prefix = properties.getExtension().getPrefix();
String exchangeName = applyPrefix(prefix, name);
TopicExchange exchange = new TopicExchange(exchangeName);
declareExchange(exchangeName, exchange);
String queueName = applyPrefix(prefix, baseQueueName);
boolean partitioned = !anonymousConsumer && properties.isPartitioned();
boolean durable = !anonymousConsumer && properties.getExtension().isDurableSubscription();
Queue queue;
if (anonymousConsumer) {
queue = new Queue(queueName, false, true, true);
}
else {
if (partitioned) {
String partitionSuffix = "-" + properties.getInstanceIndex();
queueName += partitionSuffix;
}
if (durable) {
queue = new Queue(queueName, true, false, false,
queueArgs(queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()));
}
else {
queue = new Queue(queueName, false, false, true);
}
}
declareQueue(queueName, queue);
if (partitioned) {
String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex());
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey));
}
else {
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#"));
}
Binding<MessageChannel> binding = doRegisterConsumer(baseQueueName, group, inputChannel, queue, properties);
if (durable) {
autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
}
return binding;
}
private Map<String, Object> queueArgs(String queueName, String prefix, boolean bindDlq) {
Map<String, Object> args = new HashMap<>();
if (bindDlq) {
args.put("x-dead-letter-exchange", applyPrefix(prefix, "DLX"));
args.put("x-dead-letter-routing-key", queueName);
}
return args;
}
private Binding<MessageChannel> doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue,
final ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
DefaultBinding<MessageChannel> consumerBinding;
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
this.connectionFactory);
listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
int concurrency = properties.getConcurrency();
concurrency = concurrency > 0 ? concurrency : 1;
listenerContainer.setConcurrentConsumers(concurrency);
int maxConcurrency = properties.getExtension().getMaxConcurrency();
if (maxConcurrency > concurrency) {
listenerContainer.setMaxConcurrentConsumers(maxConcurrency);
}
listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
listenerContainer.setQueues(queue);
int maxAttempts = properties.getMaxAttempts();
if (maxAttempts > 1 || properties.getExtension().isRepublishToDlq()) {
RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
.maxAttempts(maxAttempts)
.backOffOptions(properties.getBackOffInitialInterval(),
properties.getBackOffMultiplier(),
properties.getBackOffMaxInterval())
.recoverer(determineRecoverer(name, properties.getExtension().getPrefix(), properties.getExtension().isRepublishToDlq()))
.build();
listenerContainer.setAdviceChain(new Advice[] { retryInterceptor });
}
listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
listenerContainer.afterPropertiesSet();
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
adapter.setBeanFactory(this.getBeanFactory());
DirectChannel bridgeToModuleChannel = new DirectChannel();
bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
bridgeToModuleChannel.setBeanName(name + ".bridge");
adapter.setOutputChannel(bridgeToModuleChannel);
adapter.setBeanName("inbound." + name);
DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns());
mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns());
adapter.setHeaderMapper(mapper);
adapter.afterPropertiesSet();
consumerBinding = new DefaultBinding<MessageChannel>(name, group, moduleInputChannel, adapter) {
@Override
protected void afterUnbind() {
cleanAutoDeclareContext(properties.getExtension().getPrefix(), name);
}
};
ReceivingHandler convertingBridge = new ReceivingHandler();
convertingBridge.setOutputChannel(moduleInputChannel);
convertingBridge.setBeanName(name + ".convert.bridge");
convertingBridge.afterPropertiesSet();
bridgeToModuleChannel.subscribe(convertingBridge);
adapter.start();
return consumerBinding;
}
private MessageRecoverer determineRecoverer(String name, String prefix, boolean republish) {
if (republish) {
RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory);
RepublishMessageRecoverer republishMessageRecoverer = new RepublishMessageRecoverer(errorTemplate,
deadLetterExchangeName(prefix),
applyPrefix(prefix, name));
return republishMessageRecoverer;
}
else {
return new RejectAndDontRequeueRecoverer();
}
}
private AmqpOutboundEndpoint buildOutboundEndpoint(final String name,
ExtendedProducerProperties<RabbitProducerProperties> properties,
RabbitTemplate rabbitTemplate) {
String prefix = properties.getExtension().getPrefix();
String exchangeName = applyPrefix(prefix, name);
TopicExchange exchange = new TopicExchange(exchangeName);
declareExchange(exchangeName, exchange);
AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate);
endpoint.setExchangeName(exchange.getName());
if (!properties.isPartitioned()) {
endpoint.setRoutingKey(name);
}
else {
endpoint.setExpressionRoutingKey(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression(name)));
}
for (String requiredGroupName : properties.getRequiredGroups()) {
String baseQueueName = exchangeName + "." + requiredGroupName;
if (!properties.isPartitioned()) {
Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq()));
declareQueue(baseQueueName, queue);
autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(name);
declareBinding(baseQueueName, binding);
}
else {
// if the stream is partitioned, create one queue for each target partition for the default group
for (int i = 0; i < properties.getPartitionCount(); i++) {
String partitionSuffix = "-" + i;
String partitionQueueName = baseQueueName + partitionSuffix;
Queue queue = new Queue(partitionQueueName, true, false, false,
queueArgs(partitionQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()));
declareQueue(queue.getName(), queue);
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(name + partitionSuffix));
}
}
}
configureOutboundHandler(endpoint, properties);
return endpoint;
}
private void configureOutboundHandler(AmqpOutboundEndpoint handler,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper();
mapper.setRequestHeaderNames(producerProperties.getExtension().getRequestHeaderPatterns());
mapper.setReplyHeaderNames(producerProperties.getExtension().getReplyHeaderPatterns());
handler.setHeaderMapper(mapper);
handler.setDefaultDeliveryMode(producerProperties.getExtension().getDeliveryMode());
handler.setBeanFactory(this.getBeanFactory());
handler.afterPropertiesSet();
}
@Override
public Binding<MessageChannel> doBindProducer(String name, MessageChannel outputChannel,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
TopicExchange exchange = new TopicExchange(exchangeName);
declareExchange(exchangeName, exchange);
AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties,
buildRabbitTemplate(producerProperties.getExtension()));
return doRegisterProducer(name, outputChannel, endpoint, producerProperties);
}
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) {
RabbitTemplate rabbitTemplate = null;
if (properties.isBatchingEnabled()) {
BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(
properties.getBatchSize(),
properties.getBatchBufferLimit(),
properties.getBatchTimeout());
rabbitTemplate = new BatchingRabbitTemplate(batchingStrategy,
getApplicationContext().getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME,
TaskScheduler.class));
}
else {
rabbitTemplate = new RabbitTemplate();
}
rabbitTemplate.setConnectionFactory(this.connectionFactory);
if (properties.isCompress()) {
rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
}
rabbitTemplate.setChannelTransacted(properties.isTransacted());
rabbitTemplate.afterPropertiesSet();
return rabbitTemplate;
}
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
AmqpOutboundEndpoint delegate, ExtendedProducerProperties<RabbitProducerProperties> properties) {
return this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties);
}
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
AmqpOutboundEndpoint delegate, String replyTo,
ExtendedProducerProperties<RabbitProducerProperties> properties) {
Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
MessageHandler handler = new SendingHandler(delegate, replyTo, properties);
EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler);
consumer.setBeanFactory(getBeanFactory());
consumer.setBeanName("outbound." + name);
consumer.afterPropertiesSet();
DefaultBinding<MessageChannel> producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer);
consumer.start();
return producerBinding;
}
/**
* If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original
* queue name because we use default exchange routing by queue name for the original message.
* @param queueName The base name for the queue (including the binder prefix, if any).
* @param routingKey The routing key for the queue.
* @param autoBindDlq true if the DLQ should be bound.
*/
private void autoBindDLQ(final String queueName, String routingKey, String prefix, boolean autoBindDlq) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("autoBindDLQ=" + autoBindDlq
+ " for: " + queueName);
}
if (autoBindDlq) {
String dlqName = constructDLQName(queueName);
Queue dlq = new Queue(dlqName);
declareQueue(dlqName, dlq);
final String dlxName = deadLetterExchangeName(prefix);
final DirectExchange dlx = new DirectExchange(dlxName);
declareExchange(dlxName, dlx);
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(routingKey));
}
}
public void declareQueue(String beanName, Queue queue) {
try {
this.rabbitAdmin.declareQueue(queue);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
}
}
addToAutoDeclareContext(beanName, queue);
}
public void declareExchange(final String rootName, final Exchange exchange) {
try {
this.rabbitAdmin.declareExchange(exchange);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
}
}
addToAutoDeclareContext(rootName + ".exchange", exchange);
}
public void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
try {
this.rabbitAdmin.declareBinding(binding);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of binding: " + rootName + ".binding deferred - connection not available");
}
}
addToAutoDeclareContext(rootName + ".binding", binding);
}
private String deadLetterExchangeName(String prefix) {
return prefix + DEAD_LETTER_EXCHANGE;
}
private void addToAutoDeclareContext(String name, Object bean) {
synchronized (this.autoDeclareContext) {
if (!this.autoDeclareContext.containsBean(name)) {
this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean);
}
}
}
public void cleanAutoDeclareContext(String prefix, String name) {
synchronized (this.autoDeclareContext) {
removeSingleton(applyPrefix(prefix, name) + ".binding");
removeSingleton(applyPrefix(prefix, name));
String dlq = applyPrefix(prefix, name) + ".dlq";
removeSingleton(dlq + ".binding");
removeSingleton(dlq);
}
}
private void removeSingleton(String name) {
if (this.autoDeclareContext.containsBean(name)) {
ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
if (beanFactory instanceof DefaultListableBeanFactory) {
((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
}
}
}
@Override
public void doManualAck(LinkedList<MessageHeaders> messageHeadersList) {
Iterator<MessageHeaders> iterator = messageHeadersList.iterator();
Map<Object, Long> channelsToAck = new HashMap<>();
while (iterator.hasNext()) {
MessageHeaders messageHeaders = iterator.next();
if (messageHeaders.containsKey(AmqpHeaders.CHANNEL)) {
Channel channel = (com.rabbitmq.client.Channel) messageHeaders.get(AmqpHeaders.CHANNEL);
Long deliveryTag = (Long) messageHeaders.get(AmqpHeaders.DELIVERY_TAG);
channelsToAck.put(channel, deliveryTag);
}
}
for (Map.Entry<Object, Long> entry : channelsToAck.entrySet()) {
try {
((Channel) entry.getKey()).basicAck(entry.getValue(), true);
}
catch (IOException e) {
this.logger.error("Exception while manually acknowledging " + e);
}
}
}
private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
private final MessageHandler delegate;
private final String replyTo;
private final ExtendedProducerProperties<RabbitProducerProperties> producerProperties;
private final PartitionHandler partitionHandler;
private SendingHandler(MessageHandler delegate, String replyTo,
ExtendedProducerProperties<RabbitProducerProperties> properties) {
this.delegate = delegate;
this.replyTo = replyTo;
producerProperties = properties;
ConfigurableListableBeanFactory beanFactory = RabbitMessageChannelBinder.this.getBeanFactory();
this.setBeanFactory(beanFactory);
this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector,
properties);
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
MessageValues messageToSend = serializePayloadIfNecessary(message);
if (this.replyTo != null) {
messageToSend.put(AmqpHeaders.REPLY_TO, this.replyTo);
}
if (producerProperties.isPartitioned()) {
messageToSend.put(PARTITION_HEADER,
this.partitionHandler.determinePartition(message));
}
this.delegate.handleMessage(messageToSend.toMessage(getMessageBuilderFactory()));
}
@Override
public void start() {
if (this.delegate instanceof Lifecycle) {
((Lifecycle) this.delegate).start();
}
}
@Override
public void stop() {
if (this.delegate instanceof Lifecycle) {
((Lifecycle) this.delegate).stop();
}
}
@Override
public boolean isRunning() {
if (this.delegate instanceof Lifecycle) {
return ((Lifecycle) this.delegate).isRunning();
}
else {
return true;
}
}
}
private final class ReceivingHandler extends AbstractReplyProducingMessageHandler {
private ReceivingHandler() {
super();
this.setBeanFactory(RabbitMessageChannelBinder.this.getBeanFactory());
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
return deserializePayloadIfNecessary(requestMessage).toMessage(getMessageBuilderFactory());
}
@Override
protected boolean shouldCopyRequestHeaders() {
/*
* we've already copied the headers so no need for the ARPMH to do it, and we don't want the content-type
* restored if absent.
*/
return false;
}
}
}

View File

@@ -0,0 +1,142 @@
/*
* Copyright 2016 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.stream.binder.rabbit;
import javax.validation.constraints.Min;
import org.springframework.amqp.core.MessageDeliveryMode;
/**
* @author Marius Bogoevici
* @author Gary Russell
*/
public class RabbitProducerProperties {
private String prefix = "";
private String[] requestHeaderPatterns = new String[] {"STANDARD_REQUEST_HEADERS", "*"};
private boolean autoBindDlq;
private boolean compress;
private boolean batchingEnabled;
private int batchSize = 100;
private int batchBufferLimit = 10000;
private int batchTimeout = 5000;
private boolean transacted;
private MessageDeliveryMode deliveryMode = MessageDeliveryMode.PERSISTENT;
private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"};
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public void setRequestHeaderPatterns(String[] requestHeaderPatterns) {
this.requestHeaderPatterns = requestHeaderPatterns;
}
public String[] getRequestHeaderPatterns() {
return requestHeaderPatterns;
}
public void setAutoBindDlq(boolean autoBindDlq) {
this.autoBindDlq = autoBindDlq;
}
public boolean isAutoBindDlq() {
return autoBindDlq;
}
public void setCompress(boolean compress) {
this.compress = compress;
}
public boolean isCompress() {
return compress;
}
public void setDeliveryMode(MessageDeliveryMode deliveryMode) {
this.deliveryMode = deliveryMode;
}
public MessageDeliveryMode getDeliveryMode() {
return deliveryMode;
}
public String[] getReplyHeaderPatterns() {
return replyHeaderPatterns;
}
public void setReplyHeaderPatterns(String[] replyHeaderPatterns) {
this.replyHeaderPatterns = replyHeaderPatterns;
}
public boolean isBatchingEnabled() {
return batchingEnabled;
}
public void setBatchingEnabled(boolean batchingEnabled) {
this.batchingEnabled = batchingEnabled;
}
@Min(value = 1, message = "Batch Size should be greater than zero.")
public int getBatchSize() {
return batchSize;
}
public void setBatchSize(int batchSize) {
this.batchSize = batchSize;
}
@Min(value = 1, message = "Batch Buffer Limit should be greater than zero.")
public int getBatchBufferLimit() {
return batchBufferLimit;
}
public void setBatchBufferLimit(int batchBufferLimit) {
this.batchBufferLimit = batchBufferLimit;
}
@Min(value = 1, message = "Batch Timeout should be greater than zero.")
public int getBatchTimeout() {
return batchTimeout;
}
public void setBatchTimeout(int batchTimeout) {
this.batchTimeout = batchTimeout;
}
public boolean isTransacted() {
return this.transacted;
}
public void setTransacted(boolean transacted) {
this.transacted = transacted;
}
}

View File

@@ -0,0 +1,56 @@
/*
* Copyright 2015 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.stream.binder.rabbit.config;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* @author David Turanski
*/
@ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder")
class RabbitBinderConfigurationProperties {
private String[] adminAdresses = new String[0];
private String[] nodes = new String[0];
private int compressionLevel;
public String[] getAdminAdresses() {
return adminAdresses;
}
public void setAdminAdresses(String[] adminAdresses) {
this.adminAdresses = adminAdresses;
}
public String[] getNodes() {
return nodes;
}
public void setNodes(String[] nodes) {
this.nodes = nodes;
}
public int getCompressionLevel() {
return compressionLevel;
}
public void setCompressionLevel(int compressionLevel) {
this.compressionLevel = compressionLevel;
}
}

View File

@@ -0,0 +1,86 @@
/*
* Copyright 2015 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.stream.binder.rabbit.config;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitExtendedBindingProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.codec.Codec;
/**
* Configuration class for RabbitMQ message channel binder.
*
* @author David Turanski
*/
@Configuration
@Import({PropertyPlaceholderAutoConfiguration.class, KryoCodecAutoConfiguration.class})
@EnableConfigurationProperties({RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class})
public class RabbitMessageChannelBinderConfiguration {
@Autowired
private Codec codec;
@Autowired
private ConnectionFactory rabbitConnectionFactory;
@Autowired
private RabbitProperties rabbitProperties;
@Autowired
private RabbitBinderConfigurationProperties rabbitBinderConfigurationProperties;
@Autowired
private RabbitExtendedBindingProperties rabbitExtendedBindingProperties;
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder() {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties);
binder.setCodec(codec);
binder.setAdminAddresses(rabbitBinderConfigurationProperties.getAdminAdresses());
binder.setCompressingPostProcessor(gZipPostProcessor());
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
binder.setNodes(rabbitBinderConfigurationProperties.getNodes());
binder.setExtendedBindingProperties(rabbitExtendedBindingProperties);
return binder;
}
@Bean
MessagePostProcessor deCompressingPostProcessor() {
return new DelegatingDecompressingPostProcessor();
}
@Bean
MessagePostProcessor gZipPostProcessor() {
GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
gZipPostProcessor.setLevel(rabbitBinderConfigurationProperties.getCompressionLevel());
return gZipPostProcessor;
}
}

View File

@@ -0,0 +1,78 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit.config;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.actuate.health.RabbitHealthIndicator;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
import org.springframework.boot.autoconfigure.cloud.CloudAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.Cloud;
import org.springframework.cloud.CloudFactory;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Profile;
/**
* Bind to services, either locally or in a cloud environment.
*
* @author Mark Fisher
* @author Dave Syer
* @author Glenn Renfro
* @author David Turanski
* @author Eric Bottard
*/
@Configuration
@ConditionalOnMissingBean(Binder.class)
@Import(RabbitMessageChannelBinderConfiguration.class)
@AutoConfigureBefore({CloudAutoConfiguration.class, RabbitAutoConfiguration.class})
public class RabbitServiceAutoConfiguration {
@Bean
public HealthIndicator binderHealthIndicator(RabbitTemplate rabbitTemplate) {
return new RabbitHealthIndicator(rabbitTemplate);
}
@Configuration
@Profile("cloud")
@ConditionalOnClass(Cloud.class)
protected static class CloudConfig {
@Bean
public Cloud cloud() {
return new CloudFactory().getCloud();
}
@Bean
@ConditionalOnMissingBean(ConnectionFactory.class)
ConnectionFactory rabbitConnectionFactory(Cloud cloud) {
return cloud.getSingletonServiceConnector(ConnectionFactory.class, null);
}
}
@Configuration
@Profile("!cloud")
@Import(RabbitAutoConfiguration.class)
protected static class NoCloudConfig {
}
}

View File

@@ -0,0 +1,21 @@
/*
* Copyright 2015-2016 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.
*/
/**
* This package contains an implementation of the {@link org.springframework.cloud.stream.binder.Binder} for RabbitMQ.
*/
package org.springframework.cloud.stream.binder.rabbit;

View File

@@ -0,0 +1,2 @@
rabbit:\
org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration

View File

@@ -0,0 +1,73 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit;
import java.util.UUID;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Gary Russell
*/
public class LocalizedQueueConnectionFactoryIntegrationTests {
@ClassRule
public static RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
private LocalizedQueueConnectionFactory lqcf;
@Before
public void setup() {
ConnectionFactory defaultConnectionFactory = rabbitAvailableRule.getResource();
String[] addresses = new String[] { "localhost:9999", "localhost:5672" };
String[] adminAddresses = new String[] { "http://localhost:15672", "http://localhost:15672" };
String[] nodes = new String[] { "foo@bar", "rabbit@localhost" };
String vhost = "/";
String username = "guest";
String password = "guest";
this.lqcf = new LocalizedQueueConnectionFactory(defaultConnectionFactory, addresses,
adminAddresses, nodes, vhost, username, password, false, null, null, null, null);
}
@Test
public void testConnect() {
RabbitAdmin admin = new RabbitAdmin(this.lqcf);
Queue queue = new Queue(UUID.randomUUID().toString(), false, false, true);
admin.declareQueue(queue);
ConnectionFactory targetConnectionFactory = this.lqcf.getTargetConnectionFactory("[" + queue.getName() + "]");
RabbitTemplate template = new RabbitTemplate(targetConnectionFactory);
template.convertAndSend("", queue.getName(), "foo");
assertThat(template.receiveAndConvert(queue.getName())).isEqualTo("foo");
((CachingConnectionFactory) targetConnectionFactory).destroy();
admin.deleteQueue(queue.getName());
}
}

View File

@@ -0,0 +1,210 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit;
import java.net.URI;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.cloud.stream.binder.AbstractBinder;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.fail;
/**
* @author Gary Russell
* @since 1.2
*/
public class RabbitBinderCleanerTests {
private static final String BINDER_PREFIX = "binder.";
@Rule
public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true);
@Test
public void testCleanStream() {
final RabbitBindingCleaner cleaner = new RabbitBindingCleaner();
final RestTemplate template = RabbitManagementUtils.buildRestTemplate("http://localhost:15672", "guest",
"guest");
final String stream1 = UUID.randomUUID().toString();
String stream2 = stream1 + "-1";
String firstQueue = null;
CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
for (int i = 0; i < 5; i++) {
String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i);
String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i);
if (firstQueue == null) {
firstQueue = queue1Name;
}
URI uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", queue1Name)
.encode().toUri();
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", queue2Name)
.encode().toUri();
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name)).encode().toUri();
template.put(uri, new AmqpQueue(false, true));
TopicExchange exchange = new TopicExchange(queue1Name);
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue1Name)).to(exchange).with(queue1Name));
exchange = new TopicExchange(queue2Name);
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name)).to(exchange).with(queue2Name));
}
final TopicExchange topic1 = new TopicExchange(
AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar"));
rabbitAdmin.declareExchange(topic1);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#"));
String foreignQueue = UUID.randomUUID().toString();
rabbitAdmin.declareQueue(new Queue(foreignQueue));
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#"));
final TopicExchange topic2 = new TopicExchange(
AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar"));
rabbitAdmin.declareExchange(topic2);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#"));
new RabbitTemplate(connectionFactory).execute(new ChannelCallback<Void>() {
@Override
public Void doInRabbit(Channel channel) throws Exception {
String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4);
String consumerTag = channel.basicConsume(queueName, new DefaultConsumer(channel));
try {
waitForConsumerStateNot(queueName, 0);
cleaner.clean(stream1, false);
fail("Expected exception");
}
catch (RabbitAdminException e) {
assertThat(e).hasMessageContaining("Queue " + queueName + " is in use");
}
channel.basicCancel(consumerTag);
waitForConsumerStateNot(queueName, 1);
try {
cleaner.clean(stream1, false);
fail("Expected exception");
}
catch (RabbitAdminException e) {
assertThat(e).hasMessageContaining("Cannot delete exchange ");
assertThat(e).hasMessageContaining("; it has bindings:");
}
return null;
}
private void waitForConsumerStateNot(String queueName, int state) throws InterruptedException {
int n = 0;
URI uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues").pathSegment(
"{vhost}", "{queue}")
.buildAndExpand("/", queueName).encode().toUri();
while (n++ < 100) {
@SuppressWarnings("unchecked")
Map<String, Object> queueInfo = template.getForObject(uri, Map.class);
if (!queueInfo.get("consumers").equals(Integer.valueOf(state))) {
break;
}
Thread.sleep(100);
}
assertThat(n < 100).withFailMessage("Consumer state remained at " + state + " after 10 seconds");
}
});
rabbitAdmin.deleteExchange(topic1.getName()); // easier than deleting the binding
rabbitAdmin.declareExchange(topic1);
rabbitAdmin.deleteQueue(foreignQueue);
connectionFactory.destroy();
Map<String, List<String>> cleanedMap = cleaner.clean(stream1, false);
assertThat(cleanedMap).hasSize(2);
List<String> cleanedQueues = cleanedMap.get("queues");
// should *not* clean stream2
assertThat(cleanedQueues).hasSize(10);
for (int i = 0; i < 5; i++) {
assertThat(cleanedQueues.get(i * 2)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i);
assertThat(cleanedQueues.get(i * 2 + 1)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i + ".dlq");
}
List<String> cleanedExchanges = cleanedMap.get("exchanges");
assertThat(cleanedExchanges).hasSize(6);
// wild card *should* clean stream2
cleanedMap = cleaner.clean(stream1 + "*", false);
assertThat(cleanedMap).hasSize(2);
cleanedQueues = cleanedMap.get("queues");
assertThat(cleanedQueues).hasSize(5);
for (int i = 0; i < 5; i++) {
assertThat(cleanedQueues.get(i)).isEqualTo(BINDER_PREFIX + stream2 + ".default." + i);
}
cleanedExchanges = cleanedMap.get("exchanges");
assertThat(cleanedExchanges).hasSize(6);
}
public static class AmqpQueue {
private boolean autoDelete;
private boolean durable;
public AmqpQueue(boolean autoDelete, boolean durable) {
this.autoDelete = autoDelete;
this.durable = durable;
}
@JsonProperty("auto_delete")
protected boolean isAutoDelete() {
return autoDelete;
}
protected void setAutoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
}
protected boolean isDurable() {
return durable;
}
protected void setDurable(boolean durable) {
this.durable = durable;
}
}
}

View File

@@ -0,0 +1,843 @@
/*
* Copyright 2013-2016 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.stream.binder.rabbit;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.zip.Deflater;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.junit.Rule;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
import org.springframework.cloud.stream.binder.PartitionTestSupport;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.context.ApplicationContext;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* @author Mark Fisher
* @author Gary Russell
* @author David Turanski
*/
public class RabbitBinderTests extends
PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
public static final String TEST_PREFIX = "bindertest.";
@Rule
public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport();
@Override
protected RabbitTestBinder getBinder() {
if (testBinder == null) {
testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(), new RabbitProperties());
}
return testBinder;
}
@Override
protected ExtendedConsumerProperties<RabbitConsumerProperties> createConsumerProperties() {
return new ExtendedConsumerProperties<>(new RabbitConsumerProperties());
}
@Override
protected ExtendedProducerProperties<RabbitProducerProperties> createProducerProperties() {
return new ExtendedProducerProperties<>(new RabbitProducerProperties());
}
@Override
protected boolean usesExplicitRouting() {
return true;
}
@Test
public void testSendAndReceiveBad() throws Exception {
RabbitTestBinder binder = getBinder();
DirectChannel moduleOutputChannel = new DirectChannel();
DirectChannel moduleInputChannel = new DirectChannel();
Binding<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel,
createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel,
createConsumerProperties());
Message<?> message = MessageBuilder.withPayload("bad").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar")
.build();
final CountDownLatch latch = new CountDownLatch(3);
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
throw new RuntimeException("bad");
}
});
moduleOutputChannel.send(message);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
producerBinding.unbind();
consumerBinding.unbind();
}
@Test
public void testConsumerProperties() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setTransacted(true);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties);
AbstractEndpoint endpoint = extractEndpoint(consumerBinding);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.AUTO);
assertThat(container.getQueueNames()[0]).startsWith(properties.getExtension().getPrefix());
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isNull();
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(1);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(1);
Advice retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(1000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(10000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(2.0);
consumerBinding.unbind();
assertThat(endpoint.isRunning()).isFalse();
properties = createConsumerProperties();
properties.getExtension().setAcknowledgeMode(AcknowledgeMode.NONE);
properties.setBackOffInitialInterval(2000);
properties.setBackOffMaxInterval(20000);
properties.setBackOffMultiplier(5.0);
properties.setConcurrency(2);
properties.setMaxAttempts(23);
properties.getExtension().setMaxConcurrency(3);
properties.getExtension().setPrefix("foo.");
properties.getExtension().setPrefetch(20);
properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
properties.getExtension().setRequeueRejected(false);
properties.getExtension().setTxSize(10);
properties.setInstanceIndex(0);
consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties);
endpoint = extractEndpoint(consumerBinding);
container = verifyContainer(endpoint);
assertThat(container.getQueueNames()[0]).isEqualTo("foo.props.0.test");
consumerBinding.unbind();
assertThat(endpoint.isRunning()).isFalse();
}
@Test
public void testProducerProperties() throws Exception {
RabbitTestBinder binder = getBinder();
Binding<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(),
createProducerProperties());
@SuppressWarnings("unchecked")
AbstractEndpoint endpoint = extractEndpoint(producerBinding);
MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode",
MessageDeliveryMode.class);
assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT);
List<?> requestHeaders = TestUtils.getPropertyValue(endpoint,
"handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class);
assertThat(requestHeaders).hasSize(2);
producerBinding.unbind();
assertThat(endpoint.isRunning()).isFalse();
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class))
.isFalse();
ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setPrefix("foo.");
properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" });
properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'"));
properties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class);
properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0"));
properties.setPartitionSelectorClass(TestPartitionSelectorClass.class);
properties.setPartitionCount(1);
properties.getExtension().setTransacted(true);
producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties);
endpoint = extractEndpoint(producerBinding);
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class)
.getExpressionString()).isEqualTo("'props.0-' + headers['partition']");
mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", MessageDeliveryMode.class);
assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT);
assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class))
.isTrue();
verifyFooRequestProducer(endpoint);
producerBinding.unbind();
assertThat(endpoint.isRunning()).isFalse();
}
@Test
public void testDurablePubSubWithAutoBindDLQ() throws Exception {
RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix(TEST_PREFIX);
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setDurableSubscription(true);
properties.setMaxAttempts(1); // disable retry
properties.getExtension().setRequeueRejected(false);
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("durableTest");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("foo");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("durabletest.0", "tgroup", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo");
int n = 0;
while (n++ < 100) {
Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "durabletest.0.tgroup.dlq");
if (deadLetter != null) {
assertThat(deadLetter).isEqualTo("foo");
break;
}
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
consumerBinding.unbind();
assertThat(admin.getQueueProperties(TEST_PREFIX + "durabletest.0.tgroup.dlq")).isNotNull();
}
@Test
public void testNonDurablePubSubWithAutoBindDLQ() throws Exception {
RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix(TEST_PREFIX);
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setDurableSubscription(false);
properties.setMaxAttempts(1); // disable retry
properties.getExtension().setRequeueRejected(false);
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("nondurabletest");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("foo");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel,
properties);
consumerBinding.unbind();
assertThat(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq")).isNull();
}
@Test
public void testAutoBindDLQ() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix(TEST_PREFIX);
properties.getExtension().setAutoBindDlq(true);
properties.setMaxAttempts(1); // disable retry
properties.getExtension().setRequeueRejected(false);
properties.getExtension().setDurableSubscription(true);
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("dlqTest");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("foo");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
int n = 0;
while (n++ < 100) {
Object deadLetter = template.receiveAndConvert(TEST_PREFIX + "dlqtest.default.dlq");
if (deadLetter != null) {
assertThat(deadLetter).isEqualTo("foo");
break;
}
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
consumerBinding.unbind();
ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext",
ApplicationContext.class);
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.binding")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq.binding")).isFalse();
assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.dlq")).isFalse();
}
@Test
public void testAutoBindDLQPartionedConsumerFirst() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix("bindertest.");
properties.getExtension().setAutoBindDlq(true);
properties.setMaxAttempts(1); // disable retry
properties.getExtension().setRequeueRejected(false);
properties.setPartitioned(true);
properties.setInstanceIndex(0);
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.0", "default",
new QueueChannel(), properties);
properties.setInstanceIndex(1);
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default",
new QueueChannel(), properties);
ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setPrefix("bindertest.");
producerProperties.getExtension().setAutoBindDlq(true);
producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
producerProperties.setPartitionCount(2);
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partDLQ.0", output, producerProperties);
final CountDownLatch latch0 = new CountDownLatch(1);
input0.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch0.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch0.countDown();
}
});
final CountDownLatch latch1 = new CountDownLatch(1);
input1.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch1.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch1.countDown();
}
});
output.send(new GenericMessage<>(1));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<>(0));
assertThat(latch0.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<>(1));
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.setReceiveTimeout(10000);
String streamDLQName = "bindertest.partDLQ.0.dlqPartGrp.dlq";
org.springframework.amqp.core.Message received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 1);
output.send(new GenericMessage<>(0));
received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 0);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
defaultConsumerBinding2.unbind();
outputBinding.unbind();
}
@Test
public void testAutoBindDLQPartitionedProducerFirst() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setPrefix("bindertest.");
properties.getExtension().setAutoBindDlq(true);
properties.setRequiredGroups("dlqPartGrp");
properties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
properties.setPartitionSelectorClass(PartitionTestSupport.class);
properties.setPartitionCount(2);
DirectChannel output = new DirectChannel();
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partDLQ.1", output, properties);
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.getExtension().setPrefix("bindertest.");
consumerProperties.getExtension().setAutoBindDlq(true);
consumerProperties.setMaxAttempts(1); // disable retry
consumerProperties.getExtension().setRequeueRejected(false);
consumerProperties.setPartitioned(true);
consumerProperties.setInstanceIndex(0);
DirectChannel input0 = new DirectChannel();
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0,
consumerProperties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer",
new QueueChannel(), consumerProperties);
consumerProperties.setInstanceIndex(1);
DirectChannel input1 = new DirectChannel();
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1,
consumerProperties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partDLQ.1", "defaultConsumer",
new QueueChannel(), consumerProperties);
final CountDownLatch latch0 = new CountDownLatch(1);
input0.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch0.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch0.countDown();
}
});
final CountDownLatch latch1 = new CountDownLatch(1);
input1.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch1.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch1.countDown();
}
});
output.send(new GenericMessage<Integer>(1));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(0));
assertThat(latch0.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<Integer>(1));
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.setReceiveTimeout(10000);
String streamDLQName = "bindertest.partDLQ.1.dlqPartGrp.dlq";
org.springframework.amqp.core.Message received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 1);
output.send(new GenericMessage<Integer>(0));
received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 0);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
defaultConsumerBinding2.unbind();
outputBinding.unbind();
}
@Test
public void testAutoBindDLQwithRepublish() throws Exception {
// pre-declare the queue with dead-lettering, users can also use a policy
RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-dead-letter-exchange", TEST_PREFIX + "DLX");
args.put("x-dead-letter-routing-key", TEST_PREFIX + "dlqpubtest.default");
Queue queue = new Queue(TEST_PREFIX + "dlqpubtest.default", true, false, false, args);
admin.declareQueue(queue);
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix(TEST_PREFIX);
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setRepublishToDlq(true);
properties.setMaxAttempts(1); // disable retry
properties.getExtension().setRequeueRejected(false);
properties.getExtension().setDurableSubscription(true);
DirectChannel moduleInputChannel = new DirectChannel();
moduleInputChannel.setBeanName("dlqPubTest");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("foo");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel,
properties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "dlqpubtest.default", "foo");
int n = 0;
while (n++ < 100) {
org.springframework.amqp.core.Message deadLetter = template.receive(TEST_PREFIX + "dlqpubtest.default.dlq");
if (deadLetter != null) {
assertThat(new String(deadLetter.getBody())).isEqualTo("foo");
assertThat(deadLetter.getMessageProperties().getHeaders()).containsKey(("x-exception-stacktrace"));
break;
}
Thread.sleep(100);
}
assertThat(n).isLessThan(100);
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@Test
public void testBatchingAndCompression() throws Exception {
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
RabbitTestBinder binder = getBinder();
ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
properties.getExtension().setBatchingEnabled(true);
properties.getExtension().setBatchSize(2);
properties.getExtension().setBatchBufferLimit(100000);
properties.getExtension().setBatchTimeout(30000);
properties.getExtension().setCompress(true);
properties.setRequiredGroups("default");
DirectChannel output = new DirectChannel();
output.setBeanName("batchingProducer");
Binding<MessageChannel> producerBinding = binder.bindProducer("batching.0", output, properties);
Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class));
new DirectFieldAccessor(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor"))
.setPropertyValue("logger", logger);
when(logger.isTraceEnabled()).thenReturn(true);
assertThat(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.level"))
.isEqualTo(Deflater.BEST_SPEED);
output.send(new GenericMessage<>("foo".getBytes()));
output.send(new GenericMessage<>("bar".getBytes()));
Object out = spyOn("batching.0.default").receive(false);
assertThat(out).isInstanceOf(byte[].class);
assertThat(new String((byte[]) out)).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
verify(logger).trace(captor.capture());
assertThat(captor.getValue().toString()).contains(("Compressed 14 to "));
QueueChannel input = new QueueChannel();
input.setBeanName("batchingConsumer");
Binding<MessageChannel> consumerBinding = binder.bindConsumer("batching.0", "test", input,
createConsumerProperties());
output.send(new GenericMessage<>("foo".getBytes()));
output.send(new GenericMessage<>("bar".getBytes()));
Message<byte[]> in = (Message<byte[]>) input.receive(10000);
assertThat(in).isNotNull();
assertThat(new String(in.getPayload())).isEqualTo("foo");
in = (Message<byte[]>) input.receive(10000);
assertThat(in).isNotNull();
assertThat(new String(in.getPayload())).isEqualTo("bar");
assertThat(in.getHeaders().get(AmqpHeaders.DELIVERY_MODE)).isNull();
producerBinding.unbind();
consumerBinding.unbind();
}
/*
* Test late binding due to broker down; queues with and without DLQs, and partitioned
* queues.
*/
@Test
public void testLateBinding() throws Exception {
RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy();
CachingConnectionFactory cf = new CachingConnectionFactory("localhost", proxy.getPort());
RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties());
RabbitTestBinder binder = new RabbitTestBinder(cf, rabbitBinder);
ExtendedProducerProperties<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setPrefix("latebinder.");
properties.getExtension().setAutoBindDlq(true);
MessageChannel moduleOutputChannel = new DirectChannel();
Binding<MessageChannel> late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, properties);
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> rabbitConsumerProperties = createConsumerProperties();
rabbitConsumerProperties.getExtension().setPrefix("latebinder.");
Binding<MessageChannel> late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel,
rabbitConsumerProperties);
properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1"));
properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
properties.setPartitionCount(2);
MessageChannel partOutputChannel = new DirectChannel();
Binding<MessageChannel> partlate0ProducerBinding = binder.bindProducer("partlate.0", partOutputChannel,
properties);
QueueChannel partInputChannel0 = new QueueChannel();
QueueChannel partInputChannel1 = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> partLateConsumerProperties = createConsumerProperties();
partLateConsumerProperties.getExtension().setPrefix("latebinder.");
partLateConsumerProperties.setPartitioned(true);
partLateConsumerProperties.setInstanceIndex(0);
Binding<MessageChannel> partlate0Consumer0Binding = binder.bindConsumer("partlate.0", "test", partInputChannel0,
partLateConsumerProperties);
partLateConsumerProperties.setInstanceIndex(1);
Binding<MessageChannel> partlate0Consumer1Binding = binder.bindConsumer("partlate.0", "test", partInputChannel1,
partLateConsumerProperties);
ExtendedProducerProperties<RabbitProducerProperties> noDlqProducerProperties = createProducerProperties();
noDlqProducerProperties.getExtension().setPrefix("latebinder.");
MessageChannel noDLQOutputChannel = new DirectChannel();
Binding<MessageChannel> noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel,
noDlqProducerProperties);
QueueChannel noDLQInputChannel = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> noDlqConsumerProperties = createConsumerProperties();
noDlqConsumerProperties.getExtension().setPrefix("latebinder.");
Binding<MessageChannel> noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel,
noDlqConsumerProperties);
MessageChannel outputChannel = new DirectChannel();
Binding<MessageChannel> pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel,
noDlqProducerProperties);
QueueChannel pubSubInputChannel = new QueueChannel();
noDlqConsumerProperties.getExtension().setDurableSubscription(false);
Binding<MessageChannel> nonDurableConsumerBinding = binder.bindConsumer("latePubSub", "lategroup",
pubSubInputChannel, noDlqConsumerProperties);
QueueChannel durablePubSubInputChannel = new QueueChannel();
noDlqConsumerProperties.getExtension().setDurableSubscription(true);
Binding<MessageChannel> durableConsumerBinding = binder.bindConsumer("latePubSub", "lateDurableGroup",
durablePubSubInputChannel, noDlqConsumerProperties);
proxy.start();
moduleOutputChannel.send(new GenericMessage<>("foo"));
Message<?> message = moduleInputChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isNotNull();
noDLQOutputChannel.send(new GenericMessage<>("bar"));
message = noDLQInputChannel.receive(10000);
assertThat(message);
assertThat(message.getPayload()).isEqualTo("bar");
outputChannel.send(new GenericMessage<>("baz"));
message = pubSubInputChannel.receive(10000);
assertThat(message);
assertThat(message.getPayload()).isEqualTo("baz");
message = durablePubSubInputChannel.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("baz");
partOutputChannel.send(new GenericMessage<>("0"));
partOutputChannel.send(new GenericMessage<>("1"));
message = partInputChannel0.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("0");
message = partInputChannel1.receive(10000);
assertThat(message).isNotNull();
assertThat(message.getPayload()).isEqualTo("1");
late0ProducerBinding.unbind();
late0ConsumerBinding.unbind();
partlate0ProducerBinding.unbind();
partlate0Consumer0Binding.unbind();
partlate0Consumer1Binding.unbind();
noDlqProducerBinding.unbind();
noDlqConsumerBinding.unbind();
pubSubProducerBinding.unbind();
nonDurableConsumerBinding.unbind();
durableConsumerBinding.unbind();
binder.cleanup();
proxy.stop();
cf.destroy();
this.rabbitAvailableRule.getResource().destroy();
}
private SimpleMessageListenerContainer verifyContainer(AbstractEndpoint endpoint) {
SimpleMessageListenerContainer container;
Advice retry;
container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer",
SimpleMessageListenerContainer.class);
assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.NONE);
assertThat(container.getQueueNames()[0]).startsWith("foo.props.0");
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(container, "concurrentConsumers")).isEqualTo(2);
assertThat(TestUtils.getPropertyValue(container, "maxConcurrentConsumers")).isEqualTo(3);
assertThat(TestUtils.getPropertyValue(container, "defaultRequeueRejected", Boolean.class)).isFalse();
assertThat(TestUtils.getPropertyValue(container, "prefetchCount")).isEqualTo(20);
assertThat(TestUtils.getPropertyValue(container, "txSize")).isEqualTo(10);
retry = TestUtils.getPropertyValue(container, "adviceChain", Advice[].class)[0];
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.retryPolicy.maxAttempts")).isEqualTo(23);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.initialInterval")).isEqualTo(2000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.maxInterval")).isEqualTo(20000L);
assertThat(TestUtils.getPropertyValue(retry, "retryOperations.backOffPolicy.multiplier")).isEqualTo(5.0);
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint, "headerMapper.requestHeaderMatcher.matchers",
List.class);
assertThat(requestMatchers).hasSize(1);
assertThat(TestUtils.getPropertyValue(requestMatchers.get(0), "pattern")).isEqualTo("foo");
return container;
}
private void verifyFooRequestProducer(AbstractEndpoint endpoint) {
List<?> requestMatchers = TestUtils.getPropertyValue(endpoint,
"handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class);
assertThat(requestMatchers).hasSize(1);
assertThat(TestUtils.getPropertyValue(requestMatchers.get(0), "pattern")).isEqualTo("foo");
}
@Override
protected String getEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class)
.getExpressionString();
}
@Override
protected String getExpectedRoutingBaseDestination(String name, String group) {
return name;
}
@Override
protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) {
return TestUtils.getPropertyValue(endpoint, "handler.delegate.exchangeNameExpression", SpelExpression.class)
.getExpressionString();
}
@Override
protected String getClassUnderTestName() {
return CLASS_UNDER_TEST_NAME;
}
@Override
public Spy spyOn(final String queue) {
final RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.setAfterReceivePostProcessors(new DelegatingDecompressingPostProcessor());
return new Spy() {
@Override
public Object receive(boolean expectNull) throws Exception {
if (expectNull) {
Thread.sleep(50);
return template.receiveAndConvert(new RabbitConsumerProperties().getPrefix() + queue);
}
Object bar = null;
int n = 0;
while (n++ < 100 && bar == null) {
bar = template.receiveAndConvert(new RabbitConsumerProperties().getPrefix() + queue);
Thread.sleep(100);
}
assertThat(n).isLessThan(100).withFailMessage("Message did not arrive in RabbitMQ");
return bar;
}
};
}
private static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy {
@Override
public Object extractKey(Message<?> message) {
return null;
}
}
private static class TestPartitionSelectorClass implements PartitionSelectorStrategy {
@Override
public int selectPartition(Object key, int partitionCount) {
return 0;
}
}
}

View File

@@ -0,0 +1,107 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit;
import java.util.HashSet;
import java.util.Set;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.AbstractTestBinder;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.codec.kryo.PojoCodec;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* Test support class for {@link RabbitMessageChannelBinder}.
*
* @author Ilayaperumal Gopinathan
* @author Gary Russell
* @author David Turanski
* @author Mark Fisher
*/
public class RabbitTestBinder extends AbstractTestBinder<RabbitMessageChannelBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private final RabbitAdmin rabbitAdmin;
private final Set<String> prefixes = new HashSet<>();
private final Set<String> queues = new HashSet<String>();
private final Set<String> exchanges = new HashSet<String>();
public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties));
}
public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {
GenericApplicationContext context = new GenericApplicationContext();
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(1);
scheduler.afterPropertiesSet();
context.getBeanFactory().registerSingleton(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);
context.refresh();
binder.setApplicationContext(context);
binder.setCodec(new PojoCodec());
this.setBinder(binder);
this.rabbitAdmin = new RabbitAdmin(connectionFactory);
}
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
if (group != null) {
this.queues.add(properties.getExtension().getPrefix() + name + ("." + group));
}
this.exchanges.add(properties.getExtension().getPrefix() + name);
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel,
ExtendedProducerProperties<RabbitProducerProperties> properties) {
this.queues.add(properties.getExtension().getPrefix() + name + ".default");
this.exchanges.add(properties.getExtension().getPrefix() + name);
return super.bindProducer(name, moduleOutputChannel, properties);
}
@Override
public void cleanup() {
for (String queue : this.queues) {
this.rabbitAdmin.deleteQueue(queue);
this.rabbitAdmin.deleteQueue(queue + ".dlq");
// delete any partitioned queues
for (int i = 0; i < 10; i++) {
this.rabbitAdmin.deleteQueue(queue + "-" + i);
this.rabbitAdmin.deleteQueue(queue + "-" + i + ".dlq");
}
}
for (String exchange : this.exchanges) {
this.rabbitAdmin.deleteExchange(exchange);
}
for (String prefix : this.prefixes) {
this.rabbitAdmin.deleteExchange(prefix + "DLX");
}
}
}

View File

@@ -0,0 +1,210 @@
/*
* Copyright 2015-2016 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.stream.binder.rabbit.integration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.junit.After;
import org.junit.ClassRule;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.utils.test.TestUtils;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.actuate.health.CompositeHealthIndicator;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.actuate.health.Status;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.MessageChannel;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Gary Russell
*/
public class RabbitBinderModuleTests {
@ClassRule
public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
private ConfigurableApplicationContext context;
public static final ConnectionFactory MOCK_CONNECTION_FACTORY = Mockito.mock(ConnectionFactory.class,
Mockito.RETURNS_MOCKS);
@After
public void tearDown() {
if (context != null) {
context.close();
context = null;
}
RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
admin.deleteQueue("binder.input.default");
admin.deleteQueue("binder.output.default");
admin.deleteExchange("binder.input");
admin.deleteExchange("binder.output");
}
@Test
public void testParentConnectionFactoryInheritedByDefault() {
context = SpringApplication.run(SimpleProcessor.class, "--server.port=0");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators).containsKey(("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo((Status.UP));
}
@Test
@SuppressWarnings("unchecked")
public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() {
context = SpringApplication.run(SimpleProcessor.class, "--server.port=0",
"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
ChannelBindingService channelBindingService = context.getBean(ChannelBindingService.class);
DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(channelBindingService);
Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>) channelBindingServiceAccessor
.getPropertyValue("consumerBindings");
Binding<MessageChannel> inputBinding = consumerBindings.get("input").get(0);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding,
"endpoint.messageListenerContainer", SimpleMessageListenerContainer.class);
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
Map<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) TestUtils
.getPropertyValue(channelBindingService, "producerBindings");
Binding<MessageChannel> outputBinding = producerBindings.get("output");
assertThat(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional",
Boolean.class)).isTrue();
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators).containsKey("rabbit");
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.UP);
}
@Test
public void testParentConnectionFactoryInheritedIfOverridden() {
context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class)
.run("--server.port=0");
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory).isSameAs(MOCK_CONNECTION_FACTORY);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertThat(bindersHealthIndicator).isNotNull();
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators).containsKey("rabbit");
// mock connection factory behaves as if down
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.DOWN);
}
@Test
public void testParentConnectionFactoryNotInheritedByCustomizedBinders() {
List<String> params = new ArrayList<>();
params.add("--spring.cloud.stream.input.binder=custom");
params.add("--spring.cloud.stream.output.binder=custom");
params.add("--spring.cloud.stream.binders.custom.type=rabbit");
params.add("--spring.cloud.stream.binders.custom.environment.foo=bar");
params.add("--server.port=0");
context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[params.size()]));
BinderFactory<?> binderFactory = context.getBean(BinderFactory.class);
Binder binder = binderFactory.getBinder(null);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
assertThat(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("indicators");
assertThat(healthIndicators).containsKey("custom");
assertThat(healthIndicators.get("custom").health().getStatus()).isEqualTo(Status.UP);
}
@EnableBinding(Processor.class)
@SpringBootApplication
public static class SimpleProcessor {
}
public static class ConnectionFactoryConfiguration {
@Bean
public ConnectionFactory connectionFactory() {
return MOCK_CONNECTION_FACTORY;
}
}
}

View File

@@ -0,0 +1,8 @@
log4j.rootCategory=DEBUG, stdout
# standard logging including calling site
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %40.40c:%4L - %m%n
log4j.category.org.springframework.cloud.binder.rabbit=DEBUG