Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-361 Fix Javadocs for JDK8

Browse Source 
JIRA: https://jira.springsource.org/browse/AMQP-361
This commit is contained in:
Gary Russell
2014-01-15 15:01:10 -05:00
parent cbfc7f1292
commit 2fcea29bc4
42 changed files with 383 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-amqp/src/main/java/org/springframework/amqp/core/Address.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -82,9 +82,9 @@ public class Address {
* Create an Address given the exchange type, exchange name and routing key. This will set the exchange type, name
* and the routing key explicitly.
*
* @param exchangeType
* @param exchangeName
* @param routingKey
* @param exchangeType The exchange type.
* @param exchangeName The exchange name.
* @param routingKey The routing key.
*/
public Address(String exchangeType, String exchangeName, String routingKey) {
this.exchangeType = exchangeType;

6
spring-amqp/src/main/java/org/springframework/amqp/core/AmqpAdmin.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -17,7 +17,7 @@ import java.util.Properties;
/**
* Specifies a basic set of portable AMQP administrative operations for AMQP > 0.8
* Specifies a basic set of portable AMQP administrative operations for AMQP > 0.8
*
* @author Mark Pollack
* @author Dave Syer
@@ -44,6 +44,8 @@ public interface AmqpAdmin {
/**
* Declare a queue whose name is automatically named. It is created with exclusive = true, autoDelete=true, and
* durable = false.
*
* @return The queue.
*/
Queue declareQueue();

14
spring-amqp/src/main/java/org/springframework/amqp/core/AmqpTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -170,6 +170,8 @@ public interface AmqpTemplate {
*
* @param callback a user-provided {@link ReceiveAndReplyCallback} implementation to process received message
* and return a reply message.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received
* @throws AmqpException if there is a problem
*/
@@ -185,6 +187,8 @@ public interface AmqpTemplate {
* @param queueName the queue name to receive a message.
* @param callback a user-provided {@link ReceiveAndReplyCallback} implementation to process received message
* and return a reply message.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received.
* @throws AmqpException if there is a problem.
*/
@@ -199,6 +203,8 @@ public interface AmqpTemplate {
* and return a reply message.
* @param replyExchange the exchange name to send reply message.
* @param replyRoutingKey the routing key to send reply message.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received.
* @throws AmqpException if there is a problem.
*/
@@ -216,6 +222,8 @@ public interface AmqpTemplate {
* and return a reply message.
* @param replyExchange the exchange name to send reply message.
* @param replyRoutingKey the routing key to send reply message.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received
* @throws AmqpException if there is a problem
*/
@@ -231,6 +239,8 @@ public interface AmqpTemplate {
* @param callback a user-provided {@link ReceiveAndReplyCallback} implementation to process received message
* and return a reply message.
* @param replyToAddressCallback the callback to determine replyTo address at runtime.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received.
* @throws AmqpException if there is a problem.
*/
@@ -247,6 +257,8 @@ public interface AmqpTemplate {
* @param callback a user-provided {@link ReceiveAndReplyCallback} implementation to process received message
* and return a reply message.
* @param replyToAddressCallback the callback to determine replyTo address at runtime.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
* @return {@code true}, if message was received
* @throws AmqpException if there is a problem
*/

5
spring-amqp/src/main/java/org/springframework/amqp/core/ReceiveAndReplyCallback.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2014 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
@@ -18,6 +18,9 @@ package org.springframework.amqp.core;
* as processor for inbound object and producer for outbound object.
*
* <p>This often as an anonymous class within a method implementation.
* @param <R> The type of the request after conversion from the {@link Message}.
* @param <S> The type of the response.
*
* @author Artem Bilan
* @since 1.3

5
spring-amqp/src/main/java/org/springframework/amqp/core/UniquelyNamedQueue.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -35,7 +35,8 @@ public class UniquelyNamedQueue extends Queue {
/**
* Create a uniquely named queue with a fixed prefix. The constructor argument specifies
* the prefix which is prepended to a random UUID.
* @param prefix
*
* @param prefix The prefix.
*/
public UniquelyNamedQueue(String prefix) {
super(prefix + "-" + UUID.randomUUID());

9
spring-amqp/src/main/java/org/springframework/amqp/remoting/client/AmqpClientInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -15,6 +15,7 @@ package org.springframework.amqp.remoting.client;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter;
import org.springframework.remoting.RemoteProxyFailureException;
@@ -77,6 +78,8 @@ public class AmqpClientInterceptor extends RemoteAccessor implements MethodInter
* sending messages as described <a href=
* "http://static.springsource.org/spring-amqp/reference/html/amqp.html#request-reply" >in the Spring-AMQP
* documentation</a>.
*
* @param amqpTemplate The amqp template.
*/
public void setAmqpTemplate(AmqpTemplate amqpTemplate) {
this.amqpTemplate = amqpTemplate;
@@ -91,6 +94,8 @@ public class AmqpClientInterceptor extends RemoteAccessor implements MethodInter
* broker. If not set, the {@link AmqpTemplate}'s default routing key will be used.
* <p>
* This property is useful if you want to use the same AmqpTemplate to talk to multiple services.
*
* @param routingKey The routing key.
*/
public void setRoutingKey(String routingKey) {
this.routingKey = routingKey;
@@ -104,6 +109,8 @@ public class AmqpClientInterceptor extends RemoteAccessor implements MethodInter
* Set the RemoteInvocationFactory to use for this accessor. Default is a {@link DefaultRemoteInvocationFactory}.
* <p>
* A custom invocation factory can add further context information to the invocation, for example user credentials.
*
* @param remoteInvocationFactory The remote invocation factory.
*/
public void setRemoteInvocationFactory(RemoteInvocationFactory remoteInvocationFactory) {
this.remoteInvocationFactory = remoteInvocationFactory;

7
spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -92,6 +92,9 @@ public class AmqpInvokerServiceExporter extends RemoteInvocationBasedExporter im
* of those the respective parameters from the original message's <code>returnAddress</code> are being used.
* <p>
* Also, the template's {@link MessageConverter} is not used for the reply.
*
* @param amqpTemplate The amqp template.
*
* @see #setMessageConverter(MessageConverter)
*/
public void setAmqpTemplate(AmqpTemplate amqpTemplate) {
@@ -111,6 +114,8 @@ public class AmqpInvokerServiceExporter extends RemoteInvocationBasedExporter im
* <p>
* Note that this class never uses the message converter of the underlying {@link AmqpTemplate}!
*
* @param messageConverter The message converter.
*
* @see org.springframework.amqp.support.converter.SimpleMessageConverter
*/
public void setMessageConverter(MessageConverter messageConverter) {

4
spring-amqp/src/main/java/org/springframework/amqp/support/converter/AbstractJsonMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License");
* Copyright 2002-2014 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,
@@ -36,6 +36,8 @@ public abstract class AbstractJsonMessageConverter extends AbstractMessageConver
/**
* Specify the default charset to use when converting to or from text-based
* Message body content. If not specified, the charset will be "UTF-8".
*
* @param defaultCharset The default charset.
*/
public void setDefaultCharset(String defaultCharset) {
this.defaultCharset = (defaultCharset != null) ? defaultCharset

9
spring-amqp/src/main/java/org/springframework/amqp/support/converter/MarshallingMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -98,6 +98,8 @@ public class MarshallingMessageConverter extends AbstractMessageConverter implem
/**
* Set the contentType to be used by this message converter.
*
* @param contentType The content type.
*/
public void setContentType(String contentType) {
this.contentType = contentType;
@@ -105,6 +107,8 @@ public class MarshallingMessageConverter extends AbstractMessageConverter implem
/**
* Set the {@link Marshaller} to be used by this message converter.
*
* @param marshaller The marshaller.
*/
public void setMarshaller(Marshaller marshaller) {
Assert.notNull(marshaller, "marshaller must not be null");
@@ -113,12 +117,15 @@ public class MarshallingMessageConverter extends AbstractMessageConverter implem
/**
* Set the {@link Unmarshaller} to be used by this message converter.
*
* @param unmarshaller The unmarshaller.
*/
public void setUnmarshaller(Unmarshaller unmarshaller) {
Assert.notNull(unmarshaller, "unmarshaller must not be null");
this.unmarshaller = unmarshaller;
}
@Override
public void afterPropertiesSet() {
Assert.notNull(this.marshaller, "Property 'marshaller' is required");
Assert.notNull(this.unmarshaller, "Property 'unmarshaller' is required");

3
spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -28,6 +28,7 @@ public interface MessageConverter {
/**
* Convert a Java object to a Message.
* @param object the object to convert
* @param messageProperties The message properties.
* @return the Message
* @throws MessageConversionException in case of conversion failure
*/

2
spring-amqp/src/main/java/org/springframework/amqp/support/converter/SerializerMessageConverter.java
View File

@@ -58,6 +58,8 @@ public class SerializerMessageConverter extends AbstractMessageConverter {
/**
* Specify the default charset to use when converting to or from text-based Message body content. If not specified,
* the charset will be "UTF-8".
*
* @param defaultCharset The default charset.
*/
public void setDefaultCharset(String defaultCharset) {
this.defaultCharset = (defaultCharset != null) ? defaultCharset : DEFAULT_CHARSET;

4
spring-amqp/src/main/java/org/springframework/amqp/support/converter/SimpleMessageConverter.java
View File

@@ -60,6 +60,8 @@ public class SimpleMessageConverter extends AbstractMessageConverter implements
* <p>
* Follows RMI's codebase conventions for dynamic class download.
*
* @param codebaseUrl The codebase URL.
*
* @see org.springframework.remoting.rmi.CodebaseAwareObjectInputStream
* @see java.rmi.server.RMIClassLoader
*/
@@ -70,6 +72,8 @@ public class SimpleMessageConverter extends AbstractMessageConverter implements
/**
* Specify the default charset to use when converting to or from text-based
* Message body content. If not specified, the charset will be "UTF-8".
*
* @param defaultCharset The default charset.
*/
public void setDefaultCharset(String defaultCharset) {
this.defaultCharset = (defaultCharset != null) ? defaultCharset : DEFAULT_CHARSET;

5
spring-erlang/src/main/java/org/springframework/erlang/connection/ConnectionFactoryUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -30,6 +30,9 @@ public class ConnectionFactoryUtils {
/**
* Release the given Connection by closing it.
*
* @param con The connection.
* @param cf The connection factory.
*/
public static void releaseConnection(Connection con, ConnectionFactory cf) {
if (con == null) {

14
spring-erlang/src/main/java/org/springframework/erlang/connection/SingleConnectionFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -28,6 +28,7 @@ import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.erlang.OtpIOException;
@@ -86,11 +87,11 @@ public class SingleConnectionFactory implements ConnectionFactory,
private boolean uniqueSelfNodeName = true;
private String selfNodeName;
private final String selfNodeName;
private String cookie;
private String peerNodeName;
private final String peerNodeName;
private OtpSelf otpSelf;
@@ -125,6 +126,7 @@ public class SingleConnectionFactory implements ConnectionFactory,
this.uniqueSelfNodeName = uniqueSelfNodeName;
}
@Override
public Connection createConnection() throws UnknownHostException,
OtpAuthException {
synchronized (this.connectionMonitor) {
@@ -163,6 +165,7 @@ public class SingleConnectionFactory implements ConnectionFactory,
* <p>As this bean implements DisposableBean, a bean factory will
* automatically invoke this on destruction of its cached singletons.
*/
@Override
public void destroy() {
resetConnection();
}
@@ -204,7 +207,8 @@ public class SingleConnectionFactory implements ConnectionFactory,
* Create a JInterface Connection via this class's ConnectionFactory.
*
* @return the new Otp Connection
* @throws OtpAuthException
* @throws OtpAuthException Any.
* @throws IOException Any.
*/
protected Connection doCreateConnection() throws IOException,
OtpAuthException {
@@ -239,6 +243,7 @@ public class SingleConnectionFactory implements ConnectionFactory,
* @see
* org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() {
Assert.isTrue(this.selfNodeName != null || this.peerNodeName != null,
"'selfNodeName' or 'peerNodeName' is required");
@@ -268,6 +273,7 @@ public class SingleConnectionFactory implements ConnectionFactory,
this.target = target;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
if (method.getName().equals("equals")) {

6
spring-erlang/src/main/java/org/springframework/erlang/core/ConnectionCallback.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -28,7 +28,9 @@ public interface ConnectionCallback<T> {
* Execute any number of operations against the supplied OTP connection,
* possibly returning a result.
*
* @throws Exception We are not sure everything it throws
* @param connection The connection.
* @return The result.
* @throws Exception We are not sure everything it throws.
*/
T doInConnection(Connection connection) throws Exception;

10
spring-erlang/src/main/java/org/springframework/util/exec/Background.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -27,7 +27,10 @@ import java.util.ArrayList;
public class Background {
/**
* Execute the command (and its args, ala Runtime.exec), sending the
* output && error streams to the void.
* output and error streams to the void.
*
* @param cmd The command and args.
* @throws IOException Any.
*/
public static void exec(String[] cmd)
throws IOException
@@ -48,12 +51,15 @@ public class Background {
/**
* Execute a command (and its args, ala Runtime.exec)
*
* @param cmd The command and args.
* @param outFile File to send standard out from the process to
* @param appendOut If true, append the file with standard out,
* else truncate or create a new file
* @param errFile File to send standard err from the process to
* @param appendErr If true, append the file with standard error,
* else truncate or create a new file
*
* @throws IOException Any.
*/
public static void exec(String[] cmd,
File outFile, boolean appendOut,

7
spring-erlang/src/main/java/org/springframework/util/exec/Escape.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -28,6 +28,9 @@ public class Escape {
/**
* Escape a string by quoting the magical elements
* (such as whitespace, quotes, slashes, etc.)
*
* @param in The string to escape.
* @return The escaped string.
*/
public static String escape(String in){
char[] inChars, outChars;
@@ -42,7 +45,7 @@ public class Escape {
if(outChars.length - numOut < 5){
outChars = enlargeArray(outChars);
}
if(Character.isWhitespace(inChars[i]) ||
inChars[i] == '\\' ||
inChars[i] == '\'' ||

10
spring-erlang/src/main/java/org/springframework/util/exec/Execute.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -43,8 +43,8 @@ public class Execute {
private String[] cmdl = null;
private String[] env = null;
private int exitValue = INVALID;
private ExecuteStreamHandler streamHandler;
private ExecuteWatchdog watchdog;
private final ExecuteStreamHandler streamHandler;
private final ExecuteWatchdog watchdog;
private File workingDirectory = null;
private boolean newEnvironment = false;
private Process process;
@@ -53,6 +53,8 @@ public class Execute {
/**
* Find the list of environment variables for this process.
*
* @return The environment.
*/
public static synchronized Vector<String> getProcEnvironment() {
@@ -360,6 +362,8 @@ public class Execute {
* @param cmd a vector of the commands to execute
* @param baseDir the base directory to run from (optional)
* @param timeToWait milliseconds to wait for completion
*
* @return The result.
*/
public static int execute(Vector<String> envVars, Vector<String> cmd, File baseDir, int timeToWait) {
try {

11
spring-erlang/src/main/java/org/springframework/util/exec/ExecuteStreamHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -15,11 +15,11 @@
*/
package org.springframework.util.exec;
import java.io.InputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Used by <code>Execute</code> to handle input and output stream of
* subprocesses.
@@ -33,6 +33,7 @@ public interface ExecuteStreamHandler {
*
* @param os output stream to write to the standard input stream of the
* subprocess
* @throws IOException Any.
*/
void setProcessInputStream(OutputStream os) throws IOException;
@@ -40,6 +41,7 @@ public interface ExecuteStreamHandler {
* Install a handler for the error stream of the subprocess.
*
* @param is input stream to read from the error stream from the subprocess
* @throws IOException Any.
*/
void setProcessErrorStream(InputStream is) throws IOException;
@@ -47,11 +49,14 @@ public interface ExecuteStreamHandler {
* Install a handler for the output stream of the subprocess.
*
* @param is input stream to read from the error stream from the subprocess
* @throws IOException Any.
*/
void setProcessOutputStream(InputStream is) throws IOException;
/**
* Start handling of the streams.
*
* @throws IOException Any.
*/
void start() throws IOException;

15
spring-erlang/src/main/java/org/springframework/util/exec/ExecuteWatchdog.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -21,12 +21,14 @@ package org.springframework.util.exec;
* Destroys a process running for too long.
* For example:
* <pre class="code">
* ExecuteWatchdog watchdog = new ExecuteWatchdog(30000);
* Execute exec = new Execute(myloghandler, watchdog);
* exec.setCommandLine(mycmdline);
* int exitvalue = exec.execute();
* if (exitvalue != SUCCESS && watchdog.killedProcess()){
* {@code
* ExecuteWatchdog watchdog = new ExecuteWatchdog(30000);
* Execute exec = new Execute(myloghandler, watchdog);
* exec.setCommandLine(mycmdline);
* int exitvalue = exec.execute();
* if (exitvalue != SUCCESS && watchdog.killedProcess()){
* // it was killed on purpose by the watchdog
* }
* }
* </pre>
@@ -74,6 +76,7 @@ public class ExecuteWatchdog implements Runnable {
* Watches the given process and terminates it, if it runs for too long.
* All information from the previous run are reset.
* @param process the process to monitor. It cannot be <tt>null</tt>
* @param execThread The thread.
* @throws IllegalStateException thrown if a process is still being monitored.
*/
public synchronized void start(Process process, Thread execThread) {

14
spring-erlang/src/main/java/org/springframework/util/exec/Os.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -37,14 +37,15 @@ public class Os {
* Determines if the OS on which Ant is executing matches the
* given OS family.
*
* @param family The OS family type desired<br />
* Possible values:<br />
* @param family The OS family type desired<br>
* Possible values:<br>
* <ul><li>dos</li>
* <li>mac</li>
* <li>netware</li>
* <li>os/2</li>
* <li>unix</li>
* <li>windows</li></ul>
* @return true if the OS is in the family.
* @since 1.5
*/
public static boolean isFamily(String family) {
@@ -55,6 +56,8 @@ public class Os {
* Determines if the OS on which Ant is executing matches the
* given OS name.
*
* @param name The name.
* @return The result.
* @since 1.7
*/
public static boolean isName(String name) {
@@ -65,6 +68,8 @@ public class Os {
* Determines if the OS on which Ant is executing matches the
* given OS architecture.
*
* @param arch The arch.
* @return The result.
* @since 1.7
*/
public static boolean isArch(String arch) {
@@ -75,6 +80,8 @@ public class Os {
* Determines if the OS on which Ant is executing matches the
* given OS version.
*
* @param version The version.
* @return The result.
* @since 1.7
*/
public static boolean isVersion(String version) {
@@ -90,6 +97,7 @@ public class Os {
* @param arch The OS architecture
* @param version The OS version
*
* @return The result.
* @since 1.7
*/
public static boolean isOs(String family, String name, String arch,

25
spring-erlang/src/main/java/org/springframework/util/exec/PumpStreamHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -15,8 +15,8 @@
*/
package org.springframework.util.exec;
import java.io.InputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
@@ -32,7 +32,7 @@ public class PumpStreamHandler implements ExecuteStreamHandler {
private Thread inputThread;
private Thread errorThread;
private OutputStream out, err;
private final OutputStream out, err;
boolean running=false;
public PumpStreamHandler(OutputStream out, OutputStream err) {
@@ -48,28 +48,33 @@ public class PumpStreamHandler implements ExecuteStreamHandler {
this(System.out, System.err);
}
public void setProcessOutputStream(InputStream is) {
@Override
public void setProcessOutputStream(InputStream is) {
createProcessOutputPump(is, out);
}
public void setProcessErrorStream(InputStream is) {
@Override
public void setProcessErrorStream(InputStream is) {
createProcessErrorPump(is, err);
}
public void setProcessInputStream(OutputStream os) {
@Override
public void setProcessInputStream(OutputStream os) {
}
public void start() {
@Override
public void start() {
inputThread.start();
errorThread.start();
running=true;
}
public void stop() {
@Override
public void stop() {
if( !running ) {
return;
}
@@ -107,6 +112,10 @@ public class PumpStreamHandler implements ExecuteStreamHandler {
/**
* Creates a stream pumper to copy the given input stream to the given output stream.
*
* @param is The input stream.
* @param os The output stream.
* @return The thread.
*/
protected Thread createPump(InputStream is, OutputStream os) {
final Thread result = new Thread(new StreamPumper(is, os));

22
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -55,6 +55,8 @@ public abstract class NamespaceUtils {
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to populate the property
* @param propertyName the name of the property to be populated
*
* @return true if defined.
*/
public static boolean setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName) {
@@ -79,6 +81,8 @@ public abstract class NamespaceUtils {
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be set on the property
*
* @return true if defined.
*/
public static boolean setValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
@@ -91,6 +95,8 @@ public abstract class NamespaceUtils {
*
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used as a constructor argument
*
* @return true if defined.
*/
public static boolean isAttributeDefined(Element element, String attributeName) {
String value = element.getAttribute(attributeName);
@@ -104,6 +110,8 @@ public abstract class NamespaceUtils {
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used as a constructor argument
*
* @return true if defined.
*/
public static boolean addConstructorArgValueIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
@@ -141,6 +149,8 @@ public abstract class NamespaceUtils {
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to set the reference
*
* @return true if defined.
*/
public static boolean addConstructorArgRefIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
@@ -159,6 +169,8 @@ public abstract class NamespaceUtils {
* @param builder the bean definition builder to be configured
* @param element the XML element where the attribute should be defined
* @param attributeName the name of the attribute whose value will be used to set the reference
*
* @return true if defined.
*/
public static boolean addConstructorArgParentRefIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
@@ -208,6 +220,8 @@ public abstract class NamespaceUtils {
* @param attributeName the name of the attribute whose value will be used as a bean reference to populate the
* property
*
* @return true if defined.
*
* @see Conventions#attributeNameToPropertyName(String)
*/
public static boolean setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
@@ -219,6 +233,9 @@ public abstract class NamespaceUtils {
/**
* Provides a user friendly description of an element based on its node name and, if available, its "id" attribute
* value. This is useful for creating error messages from within bean definition parsers.
*
* @param element The element.
* @return The description.
*/
public static String createElementDescription(Element element) {
String elementId = "'" + element.getNodeName() + "'";
@@ -255,6 +272,9 @@ public abstract class NamespaceUtils {
/**
* Parses 'auto-declare' and 'declared-by' attributes.
*
* @param element The element.
* @param builder The builder.
*/
public static void parseDeclarationControls(Element element, BeanDefinitionBuilder builder) {
NamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-declare", "shouldDeclare");

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractRoutingConnectionFactory.java
View File

@@ -40,6 +40,8 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
* <p>The key can be of arbitrary type; this class implements the
* generic lookup process only. The concrete key representation will
* be handled by {@link #determineCurrentLookupKey()}.
*
* @param targetConnectionFactories The target connection factories and lookup keys.
*/
public void setTargetConnectionFactories(Map<Object, ConnectionFactory> targetConnectionFactories) {
Assert.notNull(targetConnectionFactories, "'targetConnectionFactories' must not be null.");
@@ -52,6 +54,8 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
* <p>This {@link ConnectionFactory} will be used as target if none of the keyed
* {@link #targetConnectionFactories} match the
* {@link #determineCurrentLookupKey()} current lookup key.
*
* @param defaultTargetConnectionFactory The default target connection factory.
*/
public void setDefaultTargetConnectionFactory(ConnectionFactory defaultTargetConnectionFactory) {
this.defaultTargetConnectionFactory = defaultTargetConnectionFactory;
@@ -67,6 +71,8 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
* if the lookup key was {@code null}. Lookup keys without a {@link ConnectionFactory}
* entry will then lead to an {@link IllegalStateException}.
*
* @param lenientFallback true to fall back to the default, if specified.
*
* @see #setTargetConnectionFactories
* @see #setDefaultTargetConnectionFactory
* @see #determineCurrentLookupKey()
@@ -91,6 +97,9 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
* a lookup in the {@link #targetConnectionFactories} map,
* falls back to the specified
* {@link #defaultTargetConnectionFactory} if necessary.
*
* @return The connection factory.
*
* @see #determineCurrentLookupKey()
*/
protected ConnectionFactory determineTargetConnectionFactory() {
@@ -133,6 +142,8 @@ public abstract class AbstractRoutingConnectionFactory implements ConnectionFact
/**
* Determine the current lookup key. This will typically be implemented to check a thread-bound context.
*
* @return The lookup key.
*/
protected abstract Object determineCurrentLookupKey();

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java
View File

@@ -52,7 +52,7 @@ import com.rabbitmq.client.Channel;
* channels, such as those used in listener container threads. In those cases, the channel must be closed
* anyway in order to re-queue any un-acked messages.
* <p>
* <b>{@link CacheMode#CONNECTION} is not compatible with a Rabbit Admin that auto-declares queues etc.
* <b>{@link CacheMode#CONNECTION} is not compatible with a Rabbit Admin that auto-declares queues etc.</b>
* <p>
* <b>NOTE: This ConnectionFactory requires explicit closing of all Channels obtained form its shared Connection.</b>
* This is the usual recommendation for native Rabbit access code anyway. However, with this ConnectionFactory, its use

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -65,22 +65,27 @@ public class ConnectionFactoryUtils {
final boolean synchedLocalTransactionAllowed) {
RabbitResourceHolder holder = doGetTransactionalResourceHolder(connectionFactory, new ResourceFactory() {
@Override
public Channel getChannel(RabbitResourceHolder holder) {
return holder.getChannel();
}
@Override
public Connection getConnection(RabbitResourceHolder holder) {
return holder.getConnection();
}
@Override
public Connection createConnection() throws IOException {
return connectionFactory.createConnection();
}
@Override
public Channel createChannel(Connection con) throws IOException {
return con.createChannel(synchedLocalTransactionAllowed);
}
@Override
public boolean isSynchedLocalTransactionAllowed() {
return synchedLocalTransactionAllowed;
}
@@ -166,9 +171,6 @@ public class ConnectionFactoryUtils {
}
}
/**
*
*/
public static void registerDeliveryTag(ConnectionFactory connectionFactory, Channel channel, Long tag)
throws IOException {

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConsumerChannelRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -44,7 +44,9 @@ public class ConsumerChannelRegistry {
* an external transaction manager because local transactions work the same in that
* the channel is bound to the thread. This is for the case when a user happens
* to wire in a RabbitTransactionManager.
* @param channel
*
* @param channel The channel to register.
* @param connectionFactory The connection factory.
*/
public static void registerConsumerChannel(Channel channel, ConnectionFactory connectionFactory) {
if (logger.isDebugEnabled()) {
@@ -68,6 +70,8 @@ public class ConsumerChannelRegistry {
/**
* See registerConsumerChannel. This method is called to retrieve the
* channel for this consumer.
*
* @return The channel.
*/
public static Channel getConsumerChannel() {
ChannelHolder channelHolder = consumerChannel.get();
@@ -81,7 +85,9 @@ public class ConsumerChannelRegistry {
/**
* See registerConsumerChannel. This method is called to retrieve the
* channel for this consumer if the connection factory matches.
*
* @param connectionFactory The connection factory.
* @return The channel.
*/
public static Channel getConsumerChannel(ConnectionFactory connectionFactory) {
ChannelHolder channelHolder = consumerChannel.get();

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitAccessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -17,6 +17,7 @@ import java.io.IOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
@@ -52,18 +53,21 @@ public abstract class RabbitAccessor implements InitializingBean {
/**
* Set the ConnectionFactory to use for obtaining RabbitMQ {@link Connection Connections}.
*
* @param connectionFactory The connection factory.
*/
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.connectionFactory = connectionFactory;
}
/**
* Return the ConnectionFactory that this accessor uses for obtaining RabbitMQ {@link Connection Connections}.
* @return The ConnectionFactory that this accessor uses for obtaining RabbitMQ {@link Connection Connections}.
*/
public ConnectionFactory getConnectionFactory() {
return this.connectionFactory;
}
@Override
public void afterPropertiesSet() {
Assert.notNull(this.connectionFactory, "ConnectionFactory is required");
}

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitResourceHolder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -21,6 +21,7 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.AmqpIOException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -50,7 +51,7 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
private static final Log logger = LogFactory.getLog(RabbitResourceHolder.class);
private boolean frozen = false;
private final boolean frozen = false;
private final List<Connection> connections = new LinkedList<Connection>();
@@ -72,6 +73,7 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
/**
* @param channel a channel to add
* @param releaseAfterCompletion true if the channel should be released after completion.
*/
public RabbitResourceHolder(Channel channel, boolean releaseAfterCompletion) {
this();
@@ -87,6 +89,8 @@ public class RabbitResourceHolder extends ResourceHolderSupport {
* Whether the resources should be released after transaction completion.
* Default true. Listener containers set to false because the listener continues
* to use the channel.
*
* @return true if the resources should be released.
*/
public boolean isReleaseAfterCompletion() {
return releaseAfterCompletion;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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
@@ -129,7 +129,7 @@ public abstract class RabbitUtils {
/**
* Sets a ThreadLocal indicating the channel MUST be physically closed.
* @param b
* @param b true if the channel must be closed.
*/
public static void setPhysicalCloseRequired(boolean b) {
physicalCloseRequired.set(b);

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelAwareMessageListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 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.
@@ -33,7 +33,7 @@ public interface ChannelAwareMessageListener {
* typically sending reply messages through the given Session.
* @param message the received AMQP message (never <code>null</code>)
* @param channel the underlying Rabbit Channel (never <code>null</code>)
* @throws Exception
* @throws Exception Any.
*/
void onMessage(Message message, Channel channel) throws Exception;

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/ChannelCallback.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -28,6 +28,8 @@ public interface ChannelCallback<T> {
* Execute any number of operations against the supplied RabbitMQ
* {@link Channel}, possibly returning a result.
*
* @param channel The channel.
* @return The result.
* @throws Exception Not sure what else Rabbit Throws
*/
T doInRabbit(Channel channel) throws Exception;

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitAdmin.java
View File

@@ -42,7 +42,7 @@ import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.Channel;
/**
* RabbitMQ implementation of portable AMQP administrative operations for AMQP >= 0.9.1
* RabbitMQ implementation of portable AMQP administrative operations for AMQP &gt;= 0.9.1
*
* @author Mark Pollack
* @author Mark Fisher

51
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -249,6 +249,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
* The default converter is a SimpleMessageConverter, which is able to handle byte arrays, Strings, and Serializable
* Objects depending on the message content type header.
*
* @param messageConverter The message converter.
*
* @see #convertAndSend
* @see #receiveAndConvert
* @see org.springframework.amqp.support.converter.SimpleMessageConverter
@@ -262,6 +264,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
* content in the message headers and plain Java objects. In particular there are limitations when dealing with very
* long string headers, which hopefully are rare in practice, but if you need to use long headers you might need to
* inject a special converter here.
*
* @param messagePropertiesConverter The message properties converter.
*/
public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
Assert.notNull(messagePropertiesConverter, "messagePropertiesConverter must not be null");
@@ -271,6 +275,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
/**
* Return the message converter for this template. Useful for clients that want to take advantage of the converter
* in {@link ChannelCallback} implementations.
*
* @return The message converter.
*/
public MessageConverter getMessageConverter() {
return this.messageConverter;
@@ -333,14 +339,17 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
return unconfirmed.size() > 0 ? unconfirmed : null;
}
@Override
public void send(Message message) throws AmqpException {
send(this.exchange, this.routingKey, message);
}
@Override
public void send(String routingKey, Message message) throws AmqpException {
send(this.exchange, routingKey, message);
}
@Override
public void send(final String exchange, final String routingKey, final Message message) throws AmqpException {
this.send(exchange, routingKey, message, null);
}
@@ -350,6 +359,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
throws AmqpException {
execute(new ChannelCallback<Object>() {
@Override
public Object doInRabbit(Channel channel) throws Exception {
doSend(channel, exchange, routingKey, message, correlationData);
return null;
@@ -357,6 +367,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
});
}
@Override
public void convertAndSend(Object object) throws AmqpException {
convertAndSend(this.exchange, this.routingKey, object, (CorrelationData) null);
}
@@ -365,6 +376,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
convertAndSend(this.exchange, this.routingKey, object, correlationData);
}
@Override
public void convertAndSend(String routingKey, final Object object) throws AmqpException {
convertAndSend(this.exchange, routingKey, object, (CorrelationData) null);
}
@@ -373,6 +385,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
convertAndSend(this.exchange, routingKey, object, correlationData);
}
@Override
public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException {
convertAndSend(exchange, routingKey, object, (CorrelationData) null);
}
@@ -381,10 +394,12 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
send(exchange, routingKey, convertMessageIfNecessary(object), corrationData);
}
@Override
public void convertAndSend(Object message, MessagePostProcessor messagePostProcessor) throws AmqpException {
convertAndSend(this.exchange, this.routingKey, message, messagePostProcessor);
}
@Override
public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor)
throws AmqpException {
convertAndSend(this.exchange, routingKey, message, messagePostProcessor, null);
@@ -396,6 +411,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
convertAndSend(this.exchange, routingKey, message, messagePostProcessor, correlationData);
}
@Override
public void convertAndSend(String exchange, String routingKey, final Object message,
final MessagePostProcessor messagePostProcessor) throws AmqpException {
convertAndSend(exchange, routingKey, message, messagePostProcessor, null);
@@ -408,14 +424,17 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
send(exchange, routingKey, messageToSend, correlationData);
}
@Override
public Message receive() throws AmqpException {
String queue = this.getRequiredQueue();
return this.receive(queue);
}
@Override
public Message receive(final String queueName) {
return execute(new ChannelCallback<Message>() {
@Override
public Message doInRabbit(Channel channel) throws IOException {
GetResponse response = channel.basicGet(queueName, !isChannelTransacted());
// Response can be null is the case that there is no message on the queue.
@@ -438,10 +457,12 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
});
}
@Override
public Object receiveAndConvert() throws AmqpException {
return receiveAndConvert(this.getRequiredQueue());
}
@Override
public Object receiveAndConvert(String queueName) throws AmqpException {
Message response = receive(queueName);
if (response != null) {
@@ -578,41 +599,50 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
});
}
@Override
public Message sendAndReceive(final Message message) throws AmqpException {
return this.doSendAndReceive(this.exchange, this.routingKey, message);
}
@Override
public Message sendAndReceive(final String routingKey, final Message message) throws AmqpException {
return this.doSendAndReceive(this.exchange, routingKey, message);
}
@Override
public Message sendAndReceive(final String exchange, final String routingKey, final Message message)
throws AmqpException {
return this.doSendAndReceive(exchange, routingKey, message);
}
@Override
public Object convertSendAndReceive(final Object message) throws AmqpException {
return this.convertSendAndReceive(this.exchange, this.routingKey, message, null);
}
@Override
public Object convertSendAndReceive(final String routingKey, final Object message) throws AmqpException {
return this.convertSendAndReceive(this.exchange, routingKey, message, null);
}
@Override
public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message)
throws AmqpException {
return this.convertSendAndReceive(exchange, routingKey, message, null);
}
@Override
public Object convertSendAndReceive(final Object message, final MessagePostProcessor messagePostProcessor) throws AmqpException {
return this.convertSendAndReceive(this.exchange, this.routingKey, message, messagePostProcessor);
}
@Override
public Object convertSendAndReceive(final String routingKey, final Object message, final MessagePostProcessor messagePostProcessor)
throws AmqpException {
return this.convertSendAndReceive(this.exchange, routingKey, message, messagePostProcessor);
}
@Override
public Object convertSendAndReceive(final String exchange, final String routingKey, final Object message,
final MessagePostProcessor messagePostProcessor) throws AmqpException {
Message requestMessage = convertMessageIfNecessary(message);
@@ -653,6 +683,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
protected Message doSendAndReceiveWithTemporary(final String exchange, final String routingKey, final Message message) {
Message replyMessage = this.execute(new ChannelCallback<Message>() {
@Override
public Message doInRabbit(Channel channel) throws Exception {
final ArrayBlockingQueue<Message> replyHandoff = new ArrayBlockingQueue<Message>(1);
@@ -699,6 +730,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
protected Message doSendAndReceiveWithFixed(final String exchange, final String routingKey, final Message message) {
Message replyMessage = this.execute(new ChannelCallback<Message>() {
@Override
public Message doInRabbit(Channel channel) throws Exception {
final PendingReply pendingReply = new PendingReply();
String messageTag = UUID.randomUUID().toString();
@@ -748,6 +780,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
return replyMessage;
}
@Override
public <T> T execute(ChannelCallback<T> action) {
Assert.notNull(action, "Callback object must not be null");
RabbitResourceHolder resourceHolder = getTransactionalResourceHolder();
@@ -773,11 +806,12 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
/**
* Send the given message to the specified exchange.
*
* @param channel the RabbitMQ Channel to operate within
* @param exchange the name of the RabbitMQ exchange to send to
* @param routingKey the routing key
* @param message the Message to send
* @throws IOException if thrown by RabbitMQ API methods
* @param channel The RabbitMQ Channel to operate within.
* @param exchange The name of the RabbitMQ exchange to send to.
* @param routingKey The routing key.
* @param message The Message to send.
* @param correlationData The correlation data.
* @throws IOException If thrown by RabbitMQ API methods
*/
protected void doSend(Channel channel, String exchange, String routingKey, Message message,
CorrelationData correlationData) throws Exception {
@@ -895,6 +929,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
}
}
@Override
public void handleConfirm(PendingConfirm pendingConfirm, boolean ack) {
if (this.confirmCallback != null) {
this.confirmCallback.confirm(pendingConfirm.getCorrelationData(), ack);
@@ -906,6 +941,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
}
}
@Override
public void handleReturn(int replyCode,
String replyText,
String exchange,
@@ -929,14 +965,17 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
}
}
@Override
public boolean isConfirmListener() {
return this.confirmCallback != null;
}
@Override
public boolean isReturnListener() {
return this.returnCallback != null;
}
@Override
public void removePendingConfirmsReference(Channel channel,
SortedMap<Long, PendingConfirm> unconfirmed) {
this.pendingConfirms.remove(channel);
@@ -945,10 +984,12 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations,
}
}
@Override
public String getUUID() {
return this.uuid;
}
@Override
public void onMessage(Message message) {
try {
String messageTag;

14
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/support/RabbitGatewaySupport.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.core.support;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.BeanInitializationException;
@@ -50,7 +51,8 @@ public class RabbitGatewaySupport implements InitializingBean {
* Will automatically create a RabbitTemplate for the given ConnectionFactory.
* @see #createRabbitTemplate
* @see #setConnectionFactory(org.springframework.amqp.rabbit.connection.ConnectionFactory)
* @param connectionFactory
*
* @param connectionFactory The connection factory.
*/
public final void setConnectionFactory(ConnectionFactory connectionFactory) {
this.rabbitTemplate = createRabbitTemplate(connectionFactory);
@@ -59,6 +61,7 @@ public class RabbitGatewaySupport implements InitializingBean {
/**
* Create a RabbitTemplate for the given ConnectionFactory.
* Only invoked if populating the gateway with a ConnectionFactory reference.
*
* @param connectionFactory the Rabbit ConnectionFactory to create a RabbitTemplate for
* @return the new RabbitTemplate instance
* @see #setConnectionFactory
@@ -68,7 +71,7 @@ public class RabbitGatewaySupport implements InitializingBean {
}
/**
* Return the Rabbit ConnectionFactory used by the gateway.
* @return The Rabbit ConnectionFactory used by the gateway.
*/
public final ConnectionFactory getConnectionFactory() {
return (this.rabbitTemplate != null ? this.rabbitTemplate.getConnectionFactory() : null);
@@ -76,7 +79,7 @@ public class RabbitGatewaySupport implements InitializingBean {
/**
* Set the RabbitTemplate for the gateway.
* @param rabbitTemplate
* @param rabbitTemplate The Rabbit template.
* @see #setConnectionFactory(org.springframework.amqp.rabbit.connection.ConnectionFactory)
*/
public final void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
@@ -84,12 +87,13 @@ public class RabbitGatewaySupport implements InitializingBean {
}
/**
* Return the RabbitTemplate for the gateway.
* @return The RabbitTemplate for the gateway.
*/
public final RabbitTemplate getRabbitTemplate() {
return this.rabbitTemplate;
}
@Override
public final void afterPropertiesSet() throws IllegalArgumentException, BeanInitializationException {
if (this.rabbitTemplate == null) {
throw new IllegalArgumentException("'connectionFactory' or 'rabbitTemplate' is required");

50
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
@@ -118,7 +118,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
}
/**
* Return the name of the queue to receive messages from.
* @return the name of the queues to receive messages from.
*/
public String[] getQueueNames() {
return this.queueNames;
@@ -131,7 +131,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
}
/**
* Return whether to expose the listener {@link Channel} to a registered {@link ChannelAwareMessageListener}.
* @return whether to expose the listener {@link Channel} to a registered {@link ChannelAwareMessageListener}.
*/
public boolean isExposeListenerChannel() {
return this.exposeListenerChannel;
@@ -147,6 +147,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Note that Channels managed by an external transaction manager will always get exposed to
* {@link org.springframework.amqp.rabbit.core.RabbitTemplate} calls. So in terms of RabbitTemplate exposure, this
* setting only affects locally transacted Channels.
*
* @param exposeListenerChannel true to expose the channel.
*
* @see ChannelAwareMessageListener
*/
public void setExposeListenerChannel(boolean exposeListenerChannel) {
@@ -156,6 +159,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
/**
* Set the message listener implementation to register. This can be either a Spring {@link MessageListener} object
* or a Spring {@link ChannelAwareMessageListener} object.
*
* @param messageListener The listener.
* @throws IllegalArgumentException if the supplied listener is not a {@link MessageListener} or a
* {@link ChannelAwareMessageListener}
* @see MessageListener
@@ -184,7 +189,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
}
/**
* Return the message listener object to register.
* @return The message listener object to register.
*/
public Object getMessageListener() {
return this.messageListener;
@@ -193,6 +198,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
/**
* Set an ErrorHandler to be invoked in case of any uncaught exceptions thrown while processing a Message. By
* default there will be <b>no</b> ErrorHandler so that error-level logging is the only result.
*
* @param errorHandler The error handler.
*/
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
@@ -202,11 +209,14 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Set whether to automatically start the container after initialization.
* <p>
* Default is "true"; set this to "false" to allow for manual startup through the {@link #start()} method.
*
* @param autoStartup true for auto startup.
*/
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
@Override
public boolean isAutoStartup() {
return this.autoStartup;
}
@@ -215,24 +225,28 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Specify the phase in which this container should be started and stopped. The startup order proceeds from lowest
* to highest, and the shutdown order is the reverse of that. By default this value is Integer.MAX_VALUE meaning
* that this container starts as late as possible and stops as soon as possible.
*
* @param phase The phase.
*/
public void setPhase(int phase) {
this.phase = phase;
}
/**
* Return the phase in which this container will be started and stopped.
* @return The phase in which this container will be started and stopped.
*/
@Override
public int getPhase() {
return this.phase;
}
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
}
/**
* Return the bean name that this listener container has been assigned in its containing bean factory, if any.
* @return The bean name that this listener container has been assigned in its containing bean factory, if any.
*/
protected final String getBeanName() {
return this.beanName;
@@ -268,6 +282,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Calls {@link #shutdown()} when the BeanFactory destroys the container instance.
* @see #shutdown()
*/
@Override
public void destroy() {
shutdown();
}
@@ -319,8 +334,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Register any invokers within this container.
* <p>
* Subclasses need to implement this method for their specific invoker management process.
* <p>
* @throws Exception
*
* @throws Exception Any Exception.
*/
protected abstract void doInitialize() throws Exception;
@@ -335,7 +350,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
protected abstract void doShutdown();
/**
* Return whether this container is currently active, that is, whether it has been set up but not shut down yet.
* @return Whether this container is currently active, that is, whether it has been set up but not shut down yet.
*/
public final boolean isActive() {
synchronized (this.lifecycleMonitor) {
@@ -347,6 +362,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Start this container.
* @see #doStart
*/
@Override
public void start() {
if (!initialized) {
synchronized (this.lifecycleMonitor) {
@@ -383,6 +399,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Stop this container.
* @see #doStop
*/
@Override
public void stop() {
try {
doStop();
@@ -396,6 +413,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
}
}
@Override
public void stop(Runnable callback) {
this.stop();
callback.run();
@@ -413,6 +431,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* @see #start()
* @see #stop()
*/
@Override
public final boolean isRunning() {
synchronized (this.lifecycleMonitor) {
return (this.running);
@@ -438,8 +457,11 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
/**
* Execute the specified listener, committing or rolling back the transaction afterwards (if necessary).
*
* @param channel the Rabbit Channel to operate on
* @param message the received Rabbit Message
* @throws Throwable Any Throwable.
*
* @see #invokeListener
* @see #handleListenerException
*/
@@ -462,7 +484,6 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Invoke the specified listener: either as standard MessageListener or (preferably) as SessionAwareMessageListener.
* @param channel the Rabbit Channel to operate on
* @param message the received Rabbit Message
* @throws Exception
* @throws Exception if thrown by Rabbit API methods
* @see #setMessageListener
*/
@@ -502,7 +523,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* @param channel the Rabbit Channel to operate on
* @param message the received Rabbit Message
* @throws Exception if thrown by Rabbit API methods or listener itself.
* <p/>
* <p>
* Exception thrown from listener will be wrapped to {@link ListenerExecutionFailedException}.
* @see ChannelAwareMessageListener
* @see #setExposeListenerChannel(boolean)
@@ -573,10 +594,13 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
* Invoke the specified listener as Spring Rabbit MessageListener.
* <p>
* Default implementation performs a plain invocation of the <code>onMessage</code> method.
* <p/>
* <p>
* Exception thrown from listener will be wrapped to {@link ListenerExecutionFailedException}.
*
* @param listener the Rabbit MessageListener to invoke
* @param message the received Rabbit Message
* @throws Exception Any Exception.
*
* @see org.springframework.amqp.core.MessageListener#onMessage
*/
protected void doInvokeListener(MessageListener listener, Message message) throws Exception {
@@ -637,7 +661,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor im
}
/**
* @param e
* @param e The Exception.
* @return If 'e' is of type {@link ListenerExecutionFailedException} - return 'e' as it is, otherwise wrap it to
* {@link ListenerExecutionFailedException} and return.
*/

33
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -103,6 +103,14 @@ public class BlockingQueueConsumer {
/**
* Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker
* until it is started. RequeueRejected defaults to true.
*
* @param connectionFactory The connection factory.
* @param messagePropertiesConverter The properties converter.
* @param activeObjectCounter The active object counter; used during shutdown.
* @param acknowledgeMode The acknowledgemode.
* @param transactional Whether the channel is transactional.
* @param prefetchCount The prefetch count.
* @param queues The queues.
*/
public BlockingQueueConsumer(ConnectionFactory connectionFactory,
MessagePropertiesConverter messagePropertiesConverter,
@@ -115,6 +123,15 @@ public class BlockingQueueConsumer {
/**
* Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker
* until it is started.
*
* @param connectionFactory The connection factory.
* @param messagePropertiesConverter The properties converter.
* @param activeObjectCounter The active object counter; used during shutdown.
* @param acknowledgeMode The acknowledge mode.
* @param transactional Whether the channel is transactional.
* @param prefetchCount The prefetch count.
* @param defaultRequeueRejected true to reject requeued messages.
* @param queues The queues.
*/
public BlockingQueueConsumer(ConnectionFactory connectionFactory,
MessagePropertiesConverter messagePropertiesConverter,
@@ -127,6 +144,16 @@ public class BlockingQueueConsumer {
/**
* Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker
* until it is started.
*
* @param connectionFactory The connection factory.
* @param messagePropertiesConverter The properties converter.
* @param activeObjectCounter The active object counter; used during shutdown.
* @param acknowledgeMode The acknowledge mode.
* @param transactional Whether the channel is transactional.
* @param prefetchCount The prefetch count.
* @param defaultRequeueRejected true to reject requeued messages.
* @param consumerArgs The consumer arguments (e.g. x-priority).
* @param queues The queues.
*/
public BlockingQueueConsumer(ConnectionFactory connectionFactory,
MessagePropertiesConverter messagePropertiesConverter,
@@ -464,8 +491,10 @@ public class BlockingQueueConsumer {
/**
* Perform a commit or message acknowledgement, as appropriate.
* @param locallyTransacted
* @throws IOException
*
* @param locallyTransacted Whether the channel is locally transacted.
* @throws IOException Any IOException.
* @return true if at least one delivery tag exists.
*/
public boolean commitIfNecessary(boolean locallyTransacted) throws IOException {

21
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -182,6 +182,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
/**
* Specify the interval between recovery attempts, in <b>milliseconds</b>. The default is 5000 ms, that is, 5
* seconds.
*
* @param recoveryInterval The recovery interval.
*/
public void setRecoveryInterval(long recoveryInterval) {
this.recoveryInterval = recoveryInterval;
@@ -256,7 +258,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* the minimum time (milliseconds) between starting new consumers on demand. Default is 10000
* (10 seconds).
*
* @param startConsumerMinInterval
* @param startConsumerMinInterval The minimum interval between new consumer starts.
*
* @see #setMaxConcurrentConsumers(int)
* @see #setStartConsumerMinInterval(long)
@@ -272,7 +274,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* minimum time (milliseconds) between stopping idle consumers. Default is 60000
* (1 minute).
*
* @param stopConsumerMinInterval
* @param stopConsumerMinInterval The minimum interval between consumer stops.
*
* @see #setMaxConcurrentConsumers(int)
* @see #setStopConsumerMinInterval(long)
@@ -290,7 +292,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* This is impacted by the {@link #txSize}.
* Default is 10 consecutive messages.
*
* @param consecutiveActiveTrigger
* @param consecutiveActiveTrigger The number of consecutive receives to trigger a new consumer.
*
* @see #setMaxConcurrentConsumers(int)
* @see #setStartConsumerMinInterval(long)
@@ -311,7 +313,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
*
* Default is 10 consecutive idles.
*
* @param consecutiveIdleTrigger
* @param consecutiveIdleTrigger The number of consecutive timeouts to trigger stopping a consumer.
*
* @see #setMaxConcurrentConsumers(int)
* @see #setStopConsumerMinInterval(long)
@@ -387,6 +389,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
/**
* Set the {@link MessagePropertiesConverter} for this listener container.
*
* @param messagePropertiesConverter The properties converter.
*/
public void setMessagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
Assert.notNull(messagePropertiesConverter, "messagePropertiesConverter must not be null");
@@ -400,7 +404,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* to be sent to the dead letter exchange. Setting to false causes all rejections to not
* be requeued. When true, the default can be overridden by the listener throwing an
* {@link AmqpRejectAndDontRequeueException}. Default true.
* @param defaultRequeueRejected
*
* @param defaultRequeueRejected true to reject by default.
*/
public void setDefaultRequeueRejected(boolean defaultRequeueRejected) {
this.defaultRequeueRejected = defaultRequeueRejected;
@@ -458,6 +463,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
/**
* Always use a shared Rabbit Connection.
*
* @return true
*/
protected final boolean sharedConnectionEnabled() {
return true;
@@ -467,7 +474,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* Creates the specified number of concurrent consumers, in the form of a Rabbit Channel plus associated
* MessageConsumer.
*
* @throws Exception
* @throws Exception Any Exception.
*/
@Override
protected void doInitialize() throws Exception {
@@ -486,7 +493,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* Re-initializes this container's Rabbit message consumers, if not initialized already. Then submits each consumer
* to this container's task executor.
*
* @throws Exception
* @throws Exception Any Exception.
*/
@Override
protected void doStart() throws Exception {

24
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java
View File

@@ -205,6 +205,8 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* <p>
* If no explicit delegate object has been specified, listener methods are expected to present on this adapter
* instance, that is, on a custom subclass of this adapter, defining listener methods.
*
* @param delegate The delegate listener or POJO.
*/
public void setDelegate(Object delegate) {
Assert.notNull(delegate, "Delegate must not be null");
@@ -212,7 +214,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
}
/**
* Return the target object to delegate message listening to.
* @return The target object to delegate message listening to.
*/
protected Object getDelegate() {
return this.delegate;
@@ -230,6 +232,9 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
/**
* Specify the name of the default listener method to delegate to, for the case where no specific listener method
* has been determined. Out-of-the-box value is {@link #ORIGINAL_DEFAULT_LISTENER_METHOD "handleMessage"}.
*
* @param defaultListenerMethod The name of the default listener method.
*
* @see #getListenerMethodName
*/
public void setDefaultListenerMethod(String defaultListenerMethod) {
@@ -237,7 +242,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
}
/**
* Return the name of the default listener method to delegate to.
* @return The name of the default listener method to delegate to.
*/
protected String getDefaultListenerMethod() {
return this.defaultListenerMethod;
@@ -249,6 +254,8 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* <p>
* Response destinations are only relevant for listener methods that return result objects, which will be wrapped in
* a response message and sent to a response destination.
*
* @param responseRoutingKey The routing key.
*/
public void setResponseRoutingKey(String responseRoutingKey) {
this.responseRoutingKey = responseRoutingKey;
@@ -260,7 +267,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* <p>
* Response destinations are only relevant for listener methods that return result objects, which will be wrapped in
* a response message and sent to a response destination.
* @param responseExchange
* @param responseExchange The exchange.
*/
public void setResponseExchange(String responseExchange) {
this.responseExchange = responseExchange;
@@ -271,6 +278,8 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* from listener methods back to Rabbit messages.
* <p>
* The default converter is a {@link SimpleMessageConverter}, which is able to handle "text" content-types.
*
* @param messageConverter The message converter.
*/
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
@@ -279,6 +288,8 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
/**
* Return the converter that will convert incoming Rabbit messages to listener method arguments, and objects
* returned from listener methods back to Rabbit messages.
*
* @return The message converter.
*/
protected MessageConverter getMessageConverter() {
return this.messageConverter;
@@ -289,6 +300,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
}
/**
* @param immediatePublish No longer supported.
* @deprecated 'immediate' no longer support by RabbitMQ.
*/
@Deprecated
@@ -309,6 +321,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* @see #handleListenerException
* @see #onMessage(Message, Channel)
*/
@Override
public void onMessage(Message message) {
try {
onMessage(message, null);
@@ -326,6 +339,7 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
* @param channel the Rabbit channel to operate on
* @throws Exception if thrown by Rabbit API methods
*/
@Override
public void onMessage(Message message, Channel channel) throws Exception {
// Check whether the delegate is a MessageListener impl itself.
// In that case, the adapter will simply act as a pass-through.
@@ -595,9 +609,11 @@ public class MessageListenerAdapter implements MessageListener, ChannelAwareMess
}
/**
* Post-process the given message producer before using it to send the response.
* Post-process the given message before sending the response.
* <p>
* The default implementation is empty.
*
* @param channel The channel.
* @param response the outgoing Rabbit message about to be sent
* @throws Exception if thrown by Rabbit API methods
*/

9
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) 2011-2013 by the original author(s).
* Copyright (c) 2011-2014 by the original author(s).
*
* 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
@@ -54,9 +54,6 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
* A Log4J appender that publishes logging events to an AMQP Exchange.
* <p>
* A fully-configured AmqpAppender, with every option set to their defaults, would look like this:
* </p>
* <p/>
*
* <pre class="code">
* log4j.appender.amqp=org.springframework.amqp.log4j.AmqpAppender
* #-------------------------------
@@ -103,10 +100,10 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
* ## Standard Log4J stuff
* #-------------------------------
* log4j.appender.amqp.layout=org.apache.log4j.PatternLayout
* log4j.appender.amqp.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
* log4j.appender.amqp.layout.ConversionPattern=%d %p %t [%c] - &lt;%m&gt;%n
* </pre>
*
* @author Jon Brisbin <jbrisbin@vmware.com>
* @author Jon Brisbin
* @author Gary Russell
*/
public class AmqpAppender extends AppenderSkeleton {

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PendingConfirm.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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.
@@ -31,8 +31,8 @@ public class PendingConfirm {
private final long timestamp;
/**
* @param correlationData
* @param timestamp
* @param correlationData The correlation data.
* @param timestamp The timestamp.
*/
public PendingConfirm(CorrelationData correlationData, long timestamp) {
this.correlationData = correlationData;

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -41,6 +41,7 @@ public interface PublisherCallbackChannel extends Channel {
* The client must <b>NOT</b> modify the contents of
* this array, and must synchronize on it when
* iterating over its collections.
*
* @param listener The Listener.
* @return A reference to pending confirms for the listener
*/
@@ -48,12 +49,16 @@ public interface PublisherCallbackChannel extends Channel {
/**
* Gets a reference to the current listener, or null.
* @return the Listener.
*
* @param listener the Listener.
* @return true if the listener was present.
*/
boolean removeListener(Listener listener);
/**
* Adds a pending confirmation to this channel's map.
*
* @param listener The listener.
* @param seq The key to the map.
* @param pendingConfirm The PendingConfirm object.
*/
@@ -86,6 +91,8 @@ public interface PublisherCallbackChannel extends Channel {
/**
* When called, this listener must remove all references to the
* pending confirm map.
*
* @param channel The channel.
* @param unconfirmed The pending confirm map.
*/
void removePendingConfirmsReference(Channel channel, SortedMap<Long, PendingConfirm> unconfirmed);