You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2014/09/21 09:56:27 UTC
[7/8] git commit: Fixed CS
Fixed CS
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d6361062
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d6361062
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d6361062
Branch: refs/heads/master
Commit: d6361062ea58dc5f3c1b7e0684064d164bdc2538
Parents: e12a631
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Sep 21 09:49:45 2014 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Sun Sep 21 09:49:45 2014 +0200
----------------------------------------------------------------------
components/camel-beanstalk/pom.xml | 13 +-
components/camel-beanstalk/src/etc/header.txt | 13 --
.../component/beanstalk/BeanstalkComponent.java | 46 ++++---
.../component/beanstalk/BeanstalkConsumer.java | 131 ++++++++++---------
.../component/beanstalk/BeanstalkEndpoint.java | 82 ++++++------
.../beanstalk/BeanstalkExchangeHelper.java | 17 +--
.../component/beanstalk/BeanstalkProducer.java | 40 +++---
.../component/beanstalk/ConnectionSettings.java | 28 ++--
.../beanstalk/ConnectionSettingsFactory.java | 17 +--
.../camel/component/beanstalk/Headers.java | 37 +++---
.../beanstalk/processors/BuryCommand.java | 20 +--
.../component/beanstalk/processors/Command.java | 3 +-
.../beanstalk/processors/DefaultCommand.java | 2 +-
.../beanstalk/processors/DeleteCommand.java | 19 +--
.../beanstalk/processors/KickCommand.java | 9 +-
.../beanstalk/processors/PutCommand.java | 13 +-
.../beanstalk/processors/ReleaseCommand.java | 19 +--
.../beanstalk/processors/TouchCommand.java | 17 +--
.../beanstalk/AwaitingConsumerTest.java | 23 ++--
.../beanstalk/BeanstalkMockTestSupport.java | 6 +-
.../beanstalk/ConnectionSettingsTest.java | 25 ++--
.../beanstalk/ConsumerCompletionTest.java | 20 ++-
.../camel/component/beanstalk/EndpointTest.java | 16 ++-
.../camel/component/beanstalk/Helper.java | 7 +-
.../beanstalk/ImmediateConsumerTest.java | 22 ++--
.../camel/component/beanstalk/ProducerTest.java | 69 ++++++----
.../integration/BeanstalkCamelTestSupport.java | 11 +-
.../BuryProducerIntegrationTest.java | 10 +-
.../integration/ConsumerIntegrationTest.java | 17 +--
.../DeleteProducerIntegrationTest.java | 10 +-
.../integration/PutProducerIntegrationTest.java | 25 ++--
.../ReleaseProducerIntegrationTest.java | 10 +-
.../TouchProducerIntegrationTest.java | 10 +-
.../src/test/resources/log4j.properties | 35 +++++
34 files changed, 471 insertions(+), 371 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/pom.xml b/components/camel-beanstalk/pom.xml
index 1716fc0..f44256a 100644
--- a/components/camel-beanstalk/pom.xml
+++ b/components/camel-beanstalk/pom.xml
@@ -17,7 +17,8 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -29,7 +30,6 @@
<artifactId>camel-beanstalk</artifactId>
<packaging>bundle</packaging>
<name>Camel :: Beanstalk</name>
-
<description>Camel Beanstalk component</description>
<properties>
@@ -66,6 +66,11 @@
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<profiles>
@@ -76,13 +81,13 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
- <version>2.14</version>
<configuration>
- <excludes />
+ <excludes/>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
+
</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/etc/header.txt
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/etc/header.txt b/components/camel-beanstalk/src/etc/header.txt
deleted file mode 100644
index 4091cb9..0000000
--- a/components/camel-beanstalk/src/etc/header.txt
+++ /dev/null
@@ -1,13 +0,0 @@
-Copyright (C) ${year} ${author} <${email}>
-
-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.
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkComponent.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkComponent.java
index 0040697..ff59ed5 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkComponent.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkComponent.java
@@ -17,15 +17,16 @@
package org.apache.camel.component.beanstalk;
import java.util.Map;
+
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.DefaultComponent;
/**
* Beanstalk Camel component.
- *
+ * <p/>
* URI is <code>beanstalk://[host[:port]][/tube]?query</code>
- * <p>
+ * <p/>
* Parameters:<ul>
* <li><code>command</code> - one of "put", "release", "bury", "touch", "delete", "kick".
* "put" is the default for Producers.</li>
@@ -36,25 +37,24 @@ import org.apache.camel.impl.DefaultComponent;
* <li><code>consumer.awaitJob</code></li>
* </ul>
*
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
* @see BeanstalkEndpoint
* @see ConnectionSettingsFactory
*/
public class BeanstalkComponent extends DefaultComponent {
- public static final String DEFAULT_TUBE = "default";
+ public static final String DEFAULT_TUBE = "default";
- public final static String COMMAND_BURY = "bury";
- public final static String COMMAND_RELEASE = "release";
- public final static String COMMAND_PUT = "put";
- public final static String COMMAND_TOUCH = "touch";
- public final static String COMMAND_DELETE = "delete";
- public final static String COMMAND_KICK = "kick";
+ public static final String COMMAND_BURY = "bury";
+ public static final String COMMAND_RELEASE = "release";
+ public static final String COMMAND_PUT = "put";
+ public static final String COMMAND_TOUCH = "touch";
+ public static final String COMMAND_DELETE = "delete";
+ public static final String COMMAND_KICK = "kick";
- public final static long DEFAULT_PRIORITY = 1000; // 0 is highest
- public final static int DEFAULT_DELAY = 0;
- public final static int DEFAULT_TIME_TO_RUN = 60; // if 0 the daemon sets 1.
+ public static final long DEFAULT_PRIORITY = 1000; // 0 is highest
+ public static final int DEFAULT_DELAY = 0;
+ public static final int DEFAULT_TIME_TO_RUN = 60; // if 0 the daemon sets 1.
- static ConnectionSettingsFactory connFactory = ConnectionSettingsFactory.DEFAULT;
+ private static ConnectionSettingsFactory connectionSettingsFactory = ConnectionSettingsFactory.DEFAULT;
public BeanstalkComponent() {
}
@@ -69,20 +69,24 @@ public class BeanstalkComponent extends DefaultComponent {
}
@Override
- protected Endpoint createEndpoint(final String uri, final String remaining, final Map<String,Object> parameters) throws Exception {
- return new BeanstalkEndpoint(uri, this, connFactory.parseUri(remaining));
+ protected Endpoint createEndpoint(final String uri, final String remaining, final Map<String, Object> parameters) throws Exception {
+ return new BeanstalkEndpoint(uri, this, connectionSettingsFactory.parseUri(remaining));
}
/**
- * Custom ConnectionSettingsFactory.
- * <p>
+ * Custom {@link ConnectionSettingsFactory}.
+ * <p/>
* Specify which {@link ConnectionSettingsFactory} to use to make connections to Beanstalkd. Especially
* useful for unit testing without beanstalkd daemon (you can mock {@link ConnectionSettings})
- *
- * @param connFactory
+ *
+ * @param connFactory the connection factory
* @see ConnectionSettingsFactory
*/
public static void setConnectionSettingsFactory(ConnectionSettingsFactory connFactory) {
- BeanstalkComponent.connFactory = connFactory;
+ BeanstalkComponent.connectionSettingsFactory = connFactory;
+ }
+
+ public static ConnectionSettingsFactory getConnectionSettingsFactory() {
+ return connectionSettingsFactory;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkConsumer.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkConsumer.java
index 9c16f7d..ac0e058 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkConsumer.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkConsumer.java
@@ -16,109 +16,116 @@
*/
package org.apache.camel.component.beanstalk;
-import org.apache.camel.component.beanstalk.processors.*;
-import com.surftools.BeanstalkClient.BeanstalkException;
-import com.surftools.BeanstalkClient.Client;
-import com.surftools.BeanstalkClient.Job;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
+
+import com.surftools.BeanstalkClient.BeanstalkException;
+import com.surftools.BeanstalkClient.Client;
+import com.surftools.BeanstalkClient.Job;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Processor;
-import org.apache.camel.spi.Synchronization;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.beanstalk.processors.BuryCommand;
+import org.apache.camel.component.beanstalk.processors.Command;
+import org.apache.camel.component.beanstalk.processors.DeleteCommand;
+import org.apache.camel.component.beanstalk.processors.ReleaseCommand;
import org.apache.camel.impl.ScheduledPollConsumer;
+import org.apache.camel.spi.Synchronization;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* PollingConsumer to read Beanstalk jobs.
- *
+ * <p/>
* The consumer may delete the job immediately or based on successful {@link Exchange}
* completion. The behavior is configurable by <code>consumer.awaitJob</code>
* flag (by default <code>true</code>)
- *
+ * <p/>
* This consumer will add a {@link Synchronization} object to every {@link Exchange}
* object it creates in order to react on successful exchange completion or failure.
- *
+ * <p/>
* In the case of successful completion, Beanstalk's <code>delete</code> method is
* called upon the job. In the case of failure the default reaction is to call
* <code>bury</code>.
- *
+ * <p/>
* The reaction on failures is configurable: possible variants are "bury", "release" or "delete"
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
*/
public class BeanstalkConsumer extends ScheduledPollConsumer {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
-
- String onFailure = BeanstalkComponent.COMMAND_BURY;
- boolean useBlockIO = true;
- boolean deleteImmediately = false;
+ private static final Logger LOG = LoggerFactory.getLogger(BeanstalkConsumer.class);
+ private static final String[] STATS_KEY_STR = new String[]{"tube", "state"};
+ private static final String[] STATS_KEY_INT = new String[]{"age", "time-left", "timeouts", "releases", "buries", "kicks"};
- private Client client = null;
- private ExecutorService executor = null;
- private Synchronization sync = null;
-
- private static String[] statsKeysStr = new String[] {"tube", "state"};
- private static String[] statsKeysInt = new String[] {"age", "time-left", "timeouts", "releases", "buries", "kicks"};
+ private String onFailure = BeanstalkComponent.COMMAND_BURY;
+ private boolean useBlockIO = true;
+ private boolean deleteImmediately;
+ private Client client;
+ private ExecutorService executor;
+ private Synchronization sync;
private final Runnable initTask = new Runnable() {
- @Override
- public void run() {
- client = getEndpoint().getConnection().newReadingClient(useBlockIO);
- }
- };
+ @Override
+ public void run() {
+ client = getEndpoint().getConnection().newReadingClient(useBlockIO);
+ }
+ };
+
private final Callable<Exchange> pollTask = new Callable<Exchange>() {
- final Integer NO_WAIT = Integer.valueOf(0);
-
+ final Integer noWait = 0;
+
@Override
public Exchange call() throws Exception {
- if (client == null)
+ if (client == null) {
throw new RuntimeCamelException("Beanstalk client not initialized");
+ }
try {
- final Job job = client.reserve(NO_WAIT);
- if (job == null)
+ final Job job = client.reserve(noWait);
+ if (job == null) {
return null;
+ }
- if (log.isDebugEnabled())
- log.debug(String.format("Received job ID %d (data length %d)", job.getJobId(), job.getData().length));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Received job ID %d (data length %d)", job.getJobId(), job.getData().length));
+ }
final Exchange exchange = getEndpoint().createExchange(ExchangePattern.InOnly);
exchange.setProperty(Headers.JOB_ID, job.getJobId());
exchange.getIn().setBody(job.getData(), byte[].class);
- Map<String,String> jobStats = client.statsJob(job.getJobId());
+ Map<String, String> jobStats = client.statsJob(job.getJobId());
if (jobStats != null) {
- for (String key : statsKeysStr) {
- if (jobStats.containsKey(key))
- exchange.setProperty(Headers.PREFIX+key, jobStats.get(key).trim());
+ for (String key : STATS_KEY_STR) {
+ if (jobStats.containsKey(key)) {
+ exchange.setProperty(Headers.PREFIX + key, jobStats.get(key).trim());
+ }
}
- if (jobStats.containsKey("pri"))
+ if (jobStats.containsKey("pri")) {
exchange.setProperty(Headers.PRIORITY, Long.parseLong(jobStats.get("pri").trim()));
+ }
- for (String key : statsKeysInt) {
- if (jobStats.containsKey(key))
- exchange.setProperty(Headers.PREFIX+key, Integer.parseInt(jobStats.get(key).trim()));
+ for (String key : STATS_KEY_INT) {
+ if (jobStats.containsKey(key)) {
+ exchange.setProperty(Headers.PREFIX + key, Integer.parseInt(jobStats.get(key).trim()));
+ }
}
}
- if (deleteImmediately)
+ if (deleteImmediately) {
client.delete(job.getJobId());
- else
+ } else {
exchange.addOnCompletion(sync);
+ }
return exchange;
} catch (BeanstalkException e) {
- log.error("Beanstalk client error", e);
+ getExceptionHandler().handleException("Beanstalk client error", e);
resetClient();
return null;
}
}
-
};
public BeanstalkConsumer(final BeanstalkEndpoint endpoint, final Processor processor) {
@@ -130,8 +137,9 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
int messagesPolled = 0;
while (isPollAllowed()) {
final Exchange exchange = executor.submit(pollTask).get();
- if (exchange == null)
+ if (exchange == null) {
break;
+ }
++messagesPolled;
getProcessor().process(exchange);
@@ -179,13 +187,15 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
@Override
protected void doStop() throws Exception {
super.doStop();
- if (executor != null)
- executor.shutdown();
+ if (executor != null) {
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(executor);
+ }
}
protected void resetClient() {
- if (client != null)
+ if (client != null) {
client.close();
+ }
initTask.run();
}
@@ -196,14 +206,15 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
public Sync() {
successCommand = new DeleteCommand(getEndpoint());
- if (BeanstalkComponent.COMMAND_BURY.equals(onFailure))
+ if (BeanstalkComponent.COMMAND_BURY.equals(onFailure)) {
failureCommand = new BuryCommand(getEndpoint());
- else if (BeanstalkComponent.COMMAND_RELEASE.equals(onFailure))
+ } else if (BeanstalkComponent.COMMAND_RELEASE.equals(onFailure)) {
failureCommand = new ReleaseCommand(getEndpoint());
- else if (BeanstalkComponent.COMMAND_DELETE.equals(onFailure))
+ } else if (BeanstalkComponent.COMMAND_DELETE.equals(onFailure)) {
failureCommand = new DeleteCommand(getEndpoint());
- else
+ } else {
throw new IllegalArgumentException(String.format("Unknown failure command: %s", onFailure));
+ }
}
@Override
@@ -211,8 +222,7 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
try {
executor.submit(new RunCommand(successCommand, exchange)).get();
} catch (Exception e) {
- if (log.isErrorEnabled())
- log.error(String.format("Could not run completion of exchange %s", exchange), e);
+ LOG.error(String.format("Could not run completion of exchange %s", exchange), e);
}
}
@@ -221,8 +231,7 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
try {
executor.submit(new RunCommand(failureCommand, exchange)).get();
} catch (Exception e) {
- if (log.isErrorEnabled())
- log.error(String.format("%s could not run failure of exchange %s", failureCommand.getClass().getName(), exchange), e);
+ LOG.error(String.format("%s could not run failure of exchange %s", failureCommand.getClass().getName(), exchange), e);
}
}
@@ -241,14 +250,12 @@ public class BeanstalkConsumer extends ScheduledPollConsumer {
try {
command.act(client, exchange);
} catch (BeanstalkException e) {
- if (log.isWarnEnabled())
- log.warn(String.format("Post-processing %s of exchange %s failed, retrying.", command.getClass().getName(), exchange), e);
+ LOG.warn(String.format("Post-processing %s of exchange %s failed, retrying.", command.getClass().getName(), exchange), e);
resetClient();
command.act(client, exchange);
}
} catch (final Exception e) {
- if (log.isErrorEnabled())
- log.error(String.format("%s could not post-process exchange %s", command.getClass().getName(), exchange), e);
+ LOG.error(String.format("%s could not post-process exchange %s", command.getClass().getName(), exchange), e);
exchange.setException(e);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
index 62c6809..7437204 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkEndpoint.java
@@ -18,28 +18,28 @@ package org.apache.camel.component.beanstalk;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Component;
-import org.apache.camel.Producer;
-import org.apache.camel.component.beanstalk.processors.*;
import org.apache.camel.Consumer;
import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.component.beanstalk.processors.BuryCommand;
+import org.apache.camel.component.beanstalk.processors.Command;
+import org.apache.camel.component.beanstalk.processors.DeleteCommand;
+import org.apache.camel.component.beanstalk.processors.KickCommand;
+import org.apache.camel.component.beanstalk.processors.PutCommand;
+import org.apache.camel.component.beanstalk.processors.ReleaseCommand;
+import org.apache.camel.component.beanstalk.processors.TouchCommand;
import org.apache.camel.impl.ScheduledPollEndpoint;
-/**
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
- * @see BeanstalkConsumer
- * @see org.apache.camel.component.beanstalk.processors.PutCommand
- */
public class BeanstalkEndpoint extends ScheduledPollEndpoint {
final ConnectionSettings conn;
- String command = BeanstalkComponent.COMMAND_PUT;
- long priority = BeanstalkComponent.DEFAULT_PRIORITY;
- int delay = BeanstalkComponent.DEFAULT_DELAY;
- int timeToRun = BeanstalkComponent.DEFAULT_TIME_TO_RUN;
+ private String command = BeanstalkComponent.COMMAND_PUT;
+ private long jobPriority = BeanstalkComponent.DEFAULT_PRIORITY;
+ private int jobDelay = BeanstalkComponent.DEFAULT_DELAY;
+ private int jobTimeToRun = BeanstalkComponent.DEFAULT_TIME_TO_RUN;
- BeanstalkEndpoint(final String uri, final Component component, final ConnectionSettings conn) {
+ public BeanstalkEndpoint(final String uri, final Component component, final ConnectionSettings conn) {
super(uri, component);
-
this.conn = conn;
}
@@ -47,66 +47,70 @@ public class BeanstalkEndpoint extends ScheduledPollEndpoint {
return conn;
}
- /**
- * The command {@link Producer} must execute
- *
- * @param command
- */
- public void setCommand(final String command) {
- this.command = command;
+ public ConnectionSettings getConn() {
+ return conn;
}
- public void setJobPriority(final long priority) {
- this.priority = priority;
+ public String getCommand() {
+ return command;
+ }
+
+ public void setCommand(String command) {
+ this.command = command;
}
public long getJobPriority() {
- return priority;
+ return jobPriority;
}
- public void setJobDelay(final int delay) {
- this.delay = delay;
+ public void setJobPriority(long jobPriority) {
+ this.jobPriority = jobPriority;
}
public int getJobDelay() {
- return delay;
+ return jobDelay;
}
- public void setJobTimeToRun(final int timeToRun) {
- this.timeToRun = timeToRun;
+ public void setJobDelay(int jobDelay) {
+ this.jobDelay = jobDelay;
}
public int getJobTimeToRun() {
- return timeToRun;
+ return jobTimeToRun;
+ }
+
+ public void setJobTimeToRun(int jobTimeToRun) {
+ this.jobTimeToRun = jobTimeToRun;
}
/**
* Creates Camel producer.
- * <p>
+ * <p/>
* Depending on the command parameter (see {@link BeanstalkComponent} URI) it
* will create one of the producer implementations.
*
* @return {@link Producer} instance
* @throws IllegalArgumentException when {@link ConnectionSettings} cannot
- * create a writable {@link Client}
+ * create a writable {@link Client}
*/
@Override
public Producer createProducer() throws Exception {
- Command cmd = null;
- if (BeanstalkComponent.COMMAND_PUT.equals(command))
+ Command cmd;
+ if (BeanstalkComponent.COMMAND_PUT.equals(command)) {
cmd = new PutCommand(this);
- else if (BeanstalkComponent.COMMAND_RELEASE.equals(command))
+ } else if (BeanstalkComponent.COMMAND_RELEASE.equals(command)) {
cmd = new ReleaseCommand(this);
- else if (BeanstalkComponent.COMMAND_BURY.equals(command))
+ } else if (BeanstalkComponent.COMMAND_BURY.equals(command)) {
cmd = new BuryCommand(this);
- else if (BeanstalkComponent.COMMAND_TOUCH.equals(command))
+ } else if (BeanstalkComponent.COMMAND_TOUCH.equals(command)) {
cmd = new TouchCommand(this);
- else if (BeanstalkComponent.COMMAND_DELETE.equals(command))
+ } else if (BeanstalkComponent.COMMAND_DELETE.equals(command)) {
cmd = new DeleteCommand(this);
- else if (BeanstalkComponent.COMMAND_KICK.equals(command))
+ } else if (BeanstalkComponent.COMMAND_KICK.equals(command)) {
cmd = new KickCommand(this);
- else
+ } else {
throw new IllegalArgumentException(String.format("Unknown command for Beanstalk endpoint: %s", command));
+ }
return new BeanstalkProducer(this, cmd);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkExchangeHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkExchangeHelper.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkExchangeHelper.java
index 19a298f..9e39369 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkExchangeHelper.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkExchangeHelper.java
@@ -21,27 +21,28 @@ import org.apache.camel.Message;
import org.apache.camel.NoSuchHeaderException;
import org.apache.camel.util.ExchangeHelper;
-/**
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
- */
public final class BeanstalkExchangeHelper {
+
+ private BeanstalkExchangeHelper() {
+ }
+
public static long getPriority(final BeanstalkEndpoint endpoint, final Message in) {
- return in.getHeader(Headers.PRIORITY, Long.valueOf(endpoint.getJobPriority()), Long.class).longValue();
+ return in.getHeader(Headers.PRIORITY, endpoint.getJobPriority(), Long.class);
}
public static int getDelay(final BeanstalkEndpoint endpoint, final Message in) {
- return in.getHeader(Headers.DELAY, Integer.valueOf(endpoint.getJobDelay()), Integer.class).intValue();
+ return in.getHeader(Headers.DELAY, endpoint.getJobDelay(), Integer.class);
}
public static int getTimeToRun(final BeanstalkEndpoint endpoint, final Message in) {
- return in.getHeader(Headers.TIME_TO_RUN, Integer.valueOf(endpoint.getJobTimeToRun()), Integer.class).intValue();
+ return in.getHeader(Headers.TIME_TO_RUN, endpoint.getJobTimeToRun(), Integer.class);
}
public static long getJobID(final Exchange exchange) throws NoSuchHeaderException {
Long jobId = exchange.getProperty(Headers.JOB_ID, Long.class);
- if (jobId != null)
+ if (jobId != null) {
return jobId;
+ }
return ExchangeHelper.getMandatoryHeader(exchange, Headers.JOB_ID, Long.class);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkProducer.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkProducer.java
index 83cfa98..6ff0082 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkProducer.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/BeanstalkProducer.java
@@ -17,30 +17,30 @@
package org.apache.camel.component.beanstalk;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import org.apache.camel.component.beanstalk.processors.Command;
+import java.util.concurrent.Future;
+
import com.surftools.BeanstalkClient.BeanstalkException;
import com.surftools.BeanstalkClient.Client;
-import org.apache.camel.Exchange;
-import org.apache.camel.AsyncProcessor;
import org.apache.camel.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.beanstalk.processors.Command;
import org.apache.camel.impl.DefaultProducer;
-/**
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
- */
public class BeanstalkProducer extends DefaultProducer implements AsyncProcessor {
- private ExecutorService executor = null;
-
- Client client = null;
- final Command command;
+ private ExecutorService executor;
+ private Client client;
+ private final Command command;
public BeanstalkProducer(BeanstalkEndpoint endpoint, final Command command) throws Exception {
super(endpoint);
this.command = command;
}
+ public Command getCommand() {
+ return command;
+ }
+
@Override
public void process(final Exchange exchange) throws Exception {
Future f = executor.submit(new RunCommand(exchange));
@@ -65,8 +65,9 @@ public class BeanstalkProducer extends DefaultProducer implements AsyncProcessor
}
protected void closeClient() {
- if (client != null)
+ if (client != null) {
client.close();
+ }
}
protected void initClient() {
@@ -78,15 +79,15 @@ public class BeanstalkProducer extends DefaultProducer implements AsyncProcessor
super.doStart();
executor = getEndpoint().getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(this, "Beanstalk-Producer");
executor.execute(new Runnable() {
- public void run() {
- initClient();
- }
- });
+ public void run() {
+ initClient();
+ }
+ });
}
@Override
protected void doStop() throws Exception {
- executor.shutdown();
+ getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(executor);
closeClient();
super.doStop();
}
@@ -122,8 +123,9 @@ public class BeanstalkProducer extends DefaultProducer implements AsyncProcessor
} catch (Throwable t) {
exchange.setException(t);
} finally {
- if (callback != null)
+ if (callback != null) {
callback.done(false);
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettings.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettings.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettings.java
index 35359b1..d79db06 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettings.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettings.java
@@ -16,20 +16,19 @@
*/
package org.apache.camel.component.beanstalk;
-import java.util.Arrays;
+import java.io.UnsupportedEncodingException;
+import java.net.URLDecoder;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Scanner;
-import java.net.URLDecoder;
+
import com.surftools.BeanstalkClient.Client;
import com.surftools.BeanstalkClientImpl.ClientImpl;
-import java.io.UnsupportedEncodingException;
/**
* Represents the connection to Beanstalk.
- * <p>
+ * <p/>
* Along with the list of tubes it may watch.
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
*/
public class ConnectionSettings {
final String host;
@@ -54,24 +53,24 @@ public class ConnectionSettings {
while (scanner.hasNext()) {
final String tubeRaw = scanner.next();
try {
- buffer.add( URLDecoder.decode(tubeRaw, "UTF-8") );
+ buffer.add(URLDecoder.decode(tubeRaw, "UTF-8"));
} catch (UnsupportedEncodingException e) {
buffer.add(tubeRaw);
}
}
- this.tubes = buffer.toArray(new String[0]);
+ this.tubes = buffer.toArray(new String[buffer.size()]);
scanner.close();
}
/**
* Returns the {@link Client} instance ready for writing
* operations, e.g. "put".
- * <p>
+ * <p/>
* <code>use(tube)</code> is applied during this call.
*
* @return {@link Client} instance
* @throws IllegalArgumentException the exception is raised when this ConnectionSettings
- * has more than one tube.
+ * has more than one tube.
*/
public Client newWritingClient() throws IllegalArgumentException {
if (tubes.length > 1) {
@@ -93,7 +92,7 @@ public class ConnectionSettings {
/**
* Returns the {@link Client} instance for reading operations with all
* the tubes aleady watched
- * <p>
+ * <p/>
* <code>watch(tube)</code> is applied for every tube during this call.
*
* @param useBlockIO configuration param to {@link Client}
@@ -106,8 +105,9 @@ public class ConnectionSettings {
when using uniqueConnectionPerThread=false. The symptom is that ProtocolHandler
breaks the protocol, reading incomplete messages. To be investigated. */
//client.setUniqueConnectionPerThread(false);
- for (String tube : tubes)
+ for (String tube : tubes) {
client.watch(tube);
+ }
return client;
}
@@ -122,11 +122,11 @@ public class ConnectionSettings {
@Override
public int hashCode() {
- return 41*(41*(41+host.hashCode())+port)+Arrays.hashCode(tubes);
+ return 41 * (41 * (41 + host.hashCode()) + port) + Arrays.hashCode(tubes);
}
@Override
public String toString() {
- return "beanstalk://"+host+":"+port+"/"+Arrays.toString(tubes);
+ return "beanstalk://" + host + ":" + port + "/" + Arrays.toString(tubes);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettingsFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettingsFactory.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettingsFactory.java
index 7949cf2..915cc4a 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettingsFactory.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/ConnectionSettingsFactory.java
@@ -16,23 +16,24 @@
*/
package org.apache.camel.component.beanstalk;
-import com.surftools.BeanstalkClient.Client;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-/**
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
- */
+import com.surftools.BeanstalkClient.Client;
+
public class ConnectionSettingsFactory {
+
public static final ConnectionSettingsFactory DEFAULT = new ConnectionSettingsFactory();
+ private static final Pattern HOST_PORT_TUBE_RE = Pattern.compile("^(([\\w.-]+)(:([\\d]+))?/)?([\\w%+]*)$");
- final Pattern HostPortTubeRE = Pattern.compile("^(([\\w.-]+)(:([\\d]+))?/)?([\\w%+]*)$");
+ public ConnectionSettingsFactory() {
+ }
public ConnectionSettings parseUri(final String remaining) throws IllegalArgumentException {
- final Matcher m = HostPortTubeRE.matcher(remaining);
- if (!m.matches())
+ final Matcher m = HOST_PORT_TUBE_RE.matcher(remaining);
+ if (!m.matches()) {
throw new IllegalArgumentException(String.format("Invalid path format: %s - should be [<hostName>[:<port>]/][<tubes>]", remaining));
+ }
final String host = m.group(2) != null ? m.group(2) : Client.DEFAULT_HOST;
final int port = m.group(4) != null ? Integer.parseInt(m.group(4)) : Client.DEFAULT_PORT;
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/Headers.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/Headers.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/Headers.java
index b944872..9e8e3ab 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/Headers.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/Headers.java
@@ -16,31 +16,32 @@
*/
package org.apache.camel.component.beanstalk;
-/**
- *
- * @author <a href="mailto:azarov@osinka.com">Alexander Azarov</a>
- */
public final class Headers {
- public static final String PREFIX = "beanstalk.";
+
+ public static final String PREFIX = "beanstalk.";
// in
- public static final String PRIORITY = PREFIX+"priority";
- public static final String DELAY = PREFIX+"delay";
- public static final String TIME_TO_RUN = PREFIX+"timeToRun";
+ public static final String PRIORITY = PREFIX + "priority";
+ public static final String DELAY = PREFIX + "delay";
+ public static final String TIME_TO_RUN = PREFIX + "timeToRun";
// in/out
- public static final String JOB_ID = PREFIX+"jobId";
+ public static final String JOB_ID = PREFIX + "jobId";
// out
- public static final String RESULT = PREFIX+"result";
+ public static final String RESULT = PREFIX + "result";
// other info
- public static final String TUBE = PREFIX+"tube";
- public static final String STATE = PREFIX+"state";
- public static final String AGE = PREFIX+"age";
- public static final String TIME_LEFT = PREFIX+"time-left";
- public static final String TIMEOUTS = PREFIX+"timeouts";
- public static final String RELEASES = PREFIX+"releases";
- public static final String BURIES = PREFIX+"buries";
- public static final String KICKS = PREFIX+"kicks";
+ public static final String TUBE = PREFIX + "tube";
+ public static final String STATE = PREFIX + "state";
+ public static final String AGE = PREFIX + "age";
+ public static final String TIME_LEFT = PREFIX + "time-left";
+ public static final String TIMEOUTS = PREFIX + "timeouts";
+ public static final String RELEASES = PREFIX + "releases";
+ public static final String BURIES = PREFIX + "buries";
+ public static final String KICKS = PREFIX + "kicks";
+
+ private Headers() {
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/BuryCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/BuryCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/BuryCommand.java
index 2df0e60..3a3c06d 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/BuryCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/BuryCommand.java
@@ -16,17 +16,17 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
-import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
-import org.apache.camel.component.beanstalk.Headers;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
+import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
+import org.apache.camel.component.beanstalk.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class BuryCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(BuryCommand.class);
public BuryCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -36,14 +36,14 @@ public class BuryCommand extends DefaultCommand {
public void act(final Client client, final Exchange exchange) throws NoSuchHeaderException {
final Long jobId = BeanstalkExchangeHelper.getJobID(exchange);
final long priority = BeanstalkExchangeHelper.getPriority(endpoint, exchange.getIn());
- final boolean result = client.bury(jobId.longValue(), priority);
+ final boolean result = client.bury(jobId, priority);
- if (!result && log.isWarnEnabled())
- log.warn(String.format("Failed to bury job %d (with priority %d)", jobId, priority));
- else if (log.isDebugEnabled())
- log.debug(String.format("Job %d buried with priority %d. Result is %b", jobId, priority, result));
+ if (!result && LOG.isWarnEnabled()) {
+ LOG.warn(String.format("Failed to bury job %d (with priority %d)", jobId, priority));
+ } else if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Job %d buried with priority %d. Result is %b", jobId, priority, result));
+ }
answerWith(exchange, Headers.RESULT, result);
-
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/Command.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/Command.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/Command.java
index 7088279..e578286 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/Command.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/Command.java
@@ -20,5 +20,6 @@ import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
public interface Command {
- public void act(Client client, Exchange exchange) throws Exception;
+
+ void act(Client client, Exchange exchange) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DefaultCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DefaultCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DefaultCommand.java
index e8e9a45..d81bb10 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DefaultCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DefaultCommand.java
@@ -16,9 +16,9 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
import org.apache.camel.util.ExchangeHelper;
abstract class DefaultCommand implements Command {
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DeleteCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DeleteCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DeleteCommand.java
index f738556..6f5967b 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DeleteCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/DeleteCommand.java
@@ -16,17 +16,17 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
-import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
-import org.apache.camel.component.beanstalk.Headers;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
+import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
+import org.apache.camel.component.beanstalk.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DeleteCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(DeleteCommand.class);
public DeleteCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -35,11 +35,12 @@ public class DeleteCommand extends DefaultCommand {
@Override
public void act(final Client client, final Exchange exchange) throws NoSuchHeaderException {
final Long jobId = BeanstalkExchangeHelper.getJobID(exchange);
- final boolean result = client.delete(jobId.longValue());
- if (!result && log.isWarnEnabled())
- log.warn(String.format("Failed to delete job %d", jobId));
- else if (log.isDebugEnabled())
- log.debug(String.format("Job %d deleted. Result is %b", jobId, result));
+ final boolean result = client.delete(jobId);
+ if (!result && LOG.isWarnEnabled()) {
+ LOG.warn(String.format("Failed to delete job %d", jobId));
+ } else if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Job %d deleted. Result is %b", jobId, result));
+ }
answerWith(exchange, Headers.RESULT, result);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/KickCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/KickCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/KickCommand.java
index 7bad253..fdc709a 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/KickCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/KickCommand.java
@@ -16,17 +16,17 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class KickCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(KickCommand.class);
public KickCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -36,8 +36,9 @@ public class KickCommand extends DefaultCommand {
public void act(final Client client, final Exchange exchange) throws NoSuchHeaderException, InvalidPayloadException {
final Integer jobs = exchange.getIn().getMandatoryBody(Integer.class);
final int result = client.kick(jobs);
- if (log.isDebugEnabled())
- log.debug(String.format("Kick %d jobs. Kicked %d actually.", jobs, result));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Kick %d jobs. Kicked %d actually.", jobs, result));
+ }
final Message answer = getAnswerMessage(exchange);
answer.setBody(result, Integer.class);
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/PutCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/PutCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/PutCommand.java
index c6fa32e..4f5baae 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/PutCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/PutCommand.java
@@ -16,18 +16,18 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
-import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
-import org.apache.camel.component.beanstalk.Headers;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
+import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
+import org.apache.camel.component.beanstalk.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class PutCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(PutCommand.class);
public PutCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -42,8 +42,9 @@ public class PutCommand extends DefaultCommand {
final int timeToRun = BeanstalkExchangeHelper.getTimeToRun(endpoint, in);
final long jobId = client.put(priority, delay, timeToRun, in.getBody(byte[].class));
- if (log.isDebugEnabled())
- log.debug(String.format("Created job %d with priority %d, delay %d seconds and time to run %d", jobId, priority, delay, timeToRun));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Created job %d with priority %d, delay %d seconds and time to run %d", jobId, priority, delay, timeToRun));
+ }
answerWith(exchange, Headers.JOB_ID, jobId);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/ReleaseCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/ReleaseCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/ReleaseCommand.java
index f41e080..762a42d 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/ReleaseCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/ReleaseCommand.java
@@ -16,18 +16,18 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
-import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
-import org.apache.camel.component.beanstalk.Headers;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
+import org.apache.camel.component.beanstalk.BeanstalkExchangeHelper;
+import org.apache.camel.component.beanstalk.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ReleaseCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(ReleaseCommand.class);
public ReleaseCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -41,11 +41,12 @@ public class ReleaseCommand extends DefaultCommand {
final long priority = BeanstalkExchangeHelper.getPriority(endpoint, in);
final int delay = BeanstalkExchangeHelper.getDelay(endpoint, in);
- final boolean result = client.release(jobId.longValue(), priority, delay);
- if (!result && log.isWarnEnabled())
- log.warn(String.format("Failed to release job %d (priority %d, delay %d)", jobId, priority, delay));
- else if (log.isDebugEnabled())
- log.debug(String.format("Job %d released with priority %d, delay %d seconds. Result is %b", jobId, priority, delay, result));
+ final boolean result = client.release(jobId, priority, delay);
+ if (!result && LOG.isWarnEnabled()) {
+ LOG.warn(String.format("Failed to release job %d (priority %d, delay %d)", jobId, priority, delay));
+ } else if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Job %d released with priority %d, delay %d seconds. Result is %b", jobId, priority, delay, result));
+ }
answerWith(exchange, Headers.RESULT, result);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/TouchCommand.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/TouchCommand.java b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/TouchCommand.java
index c43f32c..105cc71 100644
--- a/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/TouchCommand.java
+++ b/components/camel-beanstalk/src/main/java/org/apache/camel/component/beanstalk/processors/TouchCommand.java
@@ -16,17 +16,17 @@
*/
package org.apache.camel.component.beanstalk.processors;
-import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
-import org.apache.camel.component.beanstalk.Headers;
import com.surftools.BeanstalkClient.Client;
import org.apache.camel.Exchange;
import org.apache.camel.NoSuchHeaderException;
+import org.apache.camel.component.beanstalk.BeanstalkEndpoint;
+import org.apache.camel.component.beanstalk.Headers;
import org.apache.camel.util.ExchangeHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TouchCommand extends DefaultCommand {
- private final transient Logger log = LoggerFactory.getLogger(getClass());
+ private static final Logger LOG = LoggerFactory.getLogger(TouchCommand.class);
public TouchCommand(BeanstalkEndpoint endpoint) {
super(endpoint);
@@ -35,11 +35,12 @@ public class TouchCommand extends DefaultCommand {
@Override
public void act(final Client client, final Exchange exchange) throws NoSuchHeaderException {
final Long jobId = ExchangeHelper.getMandatoryHeader(exchange, Headers.JOB_ID, Long.class);
- final boolean result = client.touch(jobId.longValue());
- if (!result && log.isWarnEnabled())
- log.warn(String.format("Failed to touch job %d", jobId));
- else if (log.isDebugEnabled())
- log.debug(String.format("Job %d touched. Result is %b", jobId, result));
+ final boolean result = client.touch(jobId);
+ if (!result && LOG.isWarnEnabled()) {
+ LOG.warn(String.format("Failed to touch job %d", jobId));
+ } else if (LOG.isDebugEnabled()) {
+ LOG.debug(String.format("Job %d touched. Result is %b", jobId, result));
+ }
answerWith(exchange, Headers.RESULT, result);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/AwaitingConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/AwaitingConsumerTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/AwaitingConsumerTest.java
index 164dff2..ebef336 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/AwaitingConsumerTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/AwaitingConsumerTest.java
@@ -16,20 +16,27 @@
*/
package org.apache.camel.component.beanstalk;
-import com.surftools.BeanstalkClient.Job;
import com.surftools.BeanstalkClient.BeanstalkException;
+import com.surftools.BeanstalkClient.Job;
import org.apache.camel.EndpointInject;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-import static org.mockito.Mockito.*;
+
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class AwaitingConsumerTest extends BeanstalkMockTestSupport {
- final String testMessage = "hello, world";
@EndpointInject(uri = "beanstalk:tube")
protected BeanstalkEndpoint endpoint;
+ private String testMessage = "hello, world";
+
@Test
public void testReceive() throws Exception {
final Job jobMock = mock(Job.class);
@@ -39,8 +46,8 @@ public class AwaitingConsumerTest extends BeanstalkMockTestSupport {
when(jobMock.getJobId()).thenReturn(jobId);
when(jobMock.getData()).thenReturn(payload);
when(client.reserve(anyInt()))
- .thenReturn(jobMock)
- .thenReturn(null);
+ .thenReturn(jobMock)
+ .thenReturn(null);
MockEndpoint result = getMockEndpoint("mock:result");
result.expectedMessageCount(1);
@@ -62,8 +69,8 @@ public class AwaitingConsumerTest extends BeanstalkMockTestSupport {
when(jobMock.getJobId()).thenReturn(jobId);
when(jobMock.getData()).thenReturn(payload);
when(client.reserve(anyInt()))
- .thenThrow(new BeanstalkException("test"))
- .thenReturn(jobMock);
+ .thenThrow(new BeanstalkException("test"))
+ .thenReturn(jobMock);
MockEndpoint result = getMockEndpoint("mock:result");
result.expectedMessageCount(1);
@@ -71,7 +78,7 @@ public class AwaitingConsumerTest extends BeanstalkMockTestSupport {
result.expectedPropertyReceived(Headers.JOB_ID, jobId);
result.message(0).header(Headers.JOB_ID).isEqualTo(jobId);
result.assertIsSatisfied(100);
-
+
verify(client, atLeast(1)).reserve(anyInt());
verify(client, times(1)).close();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/BeanstalkMockTestSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/BeanstalkMockTestSupport.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/BeanstalkMockTestSupport.java
index 42b3114..172849c 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/BeanstalkMockTestSupport.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/BeanstalkMockTestSupport.java
@@ -22,10 +22,12 @@ import org.junit.After;
import org.junit.Before;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
-import static org.mockito.Mockito.*;
+
+import static org.mockito.Mockito.reset;
public class BeanstalkMockTestSupport extends CamelTestSupport {
- @Mock Client client;
+ @Mock
+ Client client;
@Before
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConnectionSettingsTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConnectionSettingsTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConnectionSettingsTest.java
index 6b1d81d..ab221b5 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConnectionSettingsTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConnectionSettingsTest.java
@@ -16,16 +16,19 @@
*/
package org.apache.camel.component.beanstalk;
-import org.junit.Test;
-import static org.junit.Assert.*;
-
import com.surftools.BeanstalkClient.Client;
import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
public class ConnectionSettingsTest {
+
@Test
public void parseUriTest() {
- final ConnectionSettingsFactory factory = BeanstalkComponent.connFactory;
+ final ConnectionSettingsFactory factory = BeanstalkComponent.getConnectionSettingsFactory();
assertEquals("Full URI", new ConnectionSettings("host.domain.tld", 11300, "someTube"), factory.parseUri("host.domain.tld:11300/someTube"));
assertEquals("No port", new ConnectionSettings("host.domain.tld", Client.DEFAULT_PORT, "someTube"), factory.parseUri("host.domain.tld/someTube"));
assertEquals("Only tube", new ConnectionSettings(Client.DEFAULT_HOST, Client.DEFAULT_PORT, "someTube"), factory.parseUri("someTube"));
@@ -33,21 +36,21 @@ public class ConnectionSettingsTest {
@Test
public void parseTubesTest() {
- final ConnectionSettingsFactory factory = BeanstalkComponent.connFactory;
- assertArrayEquals("Full URI", new String[] {"tube1", "tube2"}, factory.parseUri("host:90/tube1+tube2").tubes);
- assertArrayEquals("No port", new String[] {"tube1", "tube2"}, factory.parseUri("host/tube1+tube2").tubes);
- assertArrayEquals("Only tubes", new String[] {"tube1", "tube2"}, factory.parseUri("tube1+tube2").tubes);
+ final ConnectionSettingsFactory factory = BeanstalkComponent.getConnectionSettingsFactory();
+ assertArrayEquals("Full URI", new String[]{"tube1", "tube2"}, factory.parseUri("host:90/tube1+tube2").tubes);
+ assertArrayEquals("No port", new String[]{"tube1", "tube2"}, factory.parseUri("host/tube1+tube2").tubes);
+ assertArrayEquals("Only tubes", new String[]{"tube1", "tube2"}, factory.parseUri("tube1+tube2").tubes);
assertArrayEquals("Empty URI", new String[0], factory.parseUri("").tubes);
}
- @Test(expected=IllegalArgumentException.class)
+ @Test(expected = IllegalArgumentException.class)
public void notValidHost() {
- final ConnectionSettingsFactory factory = BeanstalkComponent.connFactory;
+ final ConnectionSettingsFactory factory = BeanstalkComponent.getConnectionSettingsFactory();
fail(String.format("Calling on not valid URI must raise exception, but got result %s", factory.parseUri("not_valid?host/tube?")));
}
@Before
public void setUp() {
- BeanstalkComponent.connFactory = new ConnectionSettingsFactory();
+ BeanstalkComponent.setConnectionSettingsFactory(new ConnectionSettingsFactory());
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConsumerCompletionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConsumerCompletionTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConsumerCompletionTest.java
index 14a0955..4d0a863 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConsumerCompletionTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ConsumerCompletionTest.java
@@ -23,16 +23,24 @@ import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-import static org.mockito.Mockito.*;
-public class ConsumerCompletionTest extends BeanstalkMockTestSupport {
- final String testMessage = "hello, world";
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
- boolean shouldIdie = false;
- final Processor processor = new Processor() {
+public class ConsumerCompletionTest extends BeanstalkMockTestSupport {
+ private String testMessage = "hello, world";
+ private boolean shouldIdie;
+ private Processor processor = new Processor() {
@Override
public void process(Exchange exchange) throws InterruptedException {
- if (shouldIdie) throw new InterruptedException("die");
+ if (shouldIdie) {
+ throw new InterruptedException("die");
+ }
}
};
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/EndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/EndpointTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/EndpointTest.java
index 903f272..d778329 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/EndpointTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/EndpointTest.java
@@ -21,13 +21,15 @@ import org.apache.camel.FailedToCreateProducerException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.junit.After;
-import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
-import org.junit.Ignore;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
public class EndpointTest {
- CamelContext context = null;
+ CamelContext context;
@Before
public void setUp() throws Exception {
@@ -61,19 +63,19 @@ public class EndpointTest {
public void testCommand() {
BeanstalkEndpoint endpoint = context.getEndpoint("beanstalk:default?command=release", BeanstalkEndpoint.class);
assertNotNull("Beanstalk endpoint", endpoint);
- assertEquals("Command", BeanstalkComponent.COMMAND_RELEASE, endpoint.command);
+ assertEquals("Command", BeanstalkComponent.COMMAND_RELEASE, endpoint.getCommand());
}
@Test
public void testTubes() {
BeanstalkEndpoint endpoint = context.getEndpoint("beanstalk:host:11303/tube1+tube%2B+tube%3F?command=kick", BeanstalkEndpoint.class);
assertNotNull("Beanstalk endpoint", endpoint);
- assertEquals("Command", BeanstalkComponent.COMMAND_KICK, endpoint.command);
+ assertEquals("Command", BeanstalkComponent.COMMAND_KICK, endpoint.getCommand());
assertEquals("Host", "host", endpoint.conn.host);
- assertArrayEquals("Tubes", new String[] {"tube1", "tube+", "tube?"}, endpoint.conn.tubes);
+ assertArrayEquals("Tubes", new String[]{"tube1", "tube+", "tube?"}, endpoint.conn.tubes);
}
- @Test(expected=FailedToCreateProducerException.class)
+ @Test(expected = FailedToCreateProducerException.class)
public void testWrongCommand() throws Exception {
context.addRoutes(new RouteBuilder() {
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/Helper.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/Helper.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/Helper.java
index 3ef5cb9..be1ecd6 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/Helper.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/Helper.java
@@ -16,13 +16,18 @@
*/
package org.apache.camel.component.beanstalk;
-import com.surftools.BeanstalkClient.Client;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+
+import com.surftools.BeanstalkClient.Client;
import org.apache.camel.CamelContext;
public final class Helper {
+
+ private Helper() {
+ }
+
public static ConnectionSettings mockConn(final Client client) {
return new MockConnectionSettings(client);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ImmediateConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ImmediateConsumerTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ImmediateConsumerTest.java
index 7a5a296..eb24dd2 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ImmediateConsumerTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ImmediateConsumerTest.java
@@ -17,23 +17,29 @@
package org.apache.camel.component.beanstalk;
import com.surftools.BeanstalkClient.Job;
-import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class ImmediateConsumerTest extends BeanstalkMockTestSupport {
- final String testMessage = "hello, world";
+ String testMessage = "hello, world";
+ boolean shouldIdie;
- boolean shouldIdie = false;
- final Processor processor = new Processor() {
+ Processor processor = new Processor() {
@Override
public void process(Exchange exchange) throws InterruptedException {
- if (shouldIdie) throw new InterruptedException("die");
+ if (shouldIdie) {
+ throw new InterruptedException("die");
+ }
}
};
@@ -46,8 +52,8 @@ public class ImmediateConsumerTest extends BeanstalkMockTestSupport {
when(jobMock.getJobId()).thenReturn(jobId);
when(jobMock.getData()).thenReturn(payload);
when(client.reserve(anyInt()))
- .thenReturn(jobMock)
- .thenReturn(null);
+ .thenReturn(jobMock)
+ .thenReturn(null);
MockEndpoint result = getMockEndpoint("mock:result");
result.expectedMessageCount(1);
http://git-wip-us.apache.org/repos/asf/camel/blob/d6361062/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ProducerTest.java b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ProducerTest.java
index e3949a2..42e809d 100644
--- a/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ProducerTest.java
+++ b/components/camel-beanstalk/src/test/java/org/apache/camel/component/beanstalk/ProducerTest.java
@@ -17,8 +17,6 @@
package org.apache.camel.component.beanstalk;
import com.surftools.BeanstalkClient.BeanstalkException;
-import org.apache.camel.component.beanstalk.processors.*;
-import org.apache.camel.CamelExecutionException;
import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -27,14 +25,23 @@ import org.apache.camel.Produce;
import org.apache.camel.Producer;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.beanstalk.processors.BuryCommand;
+import org.apache.camel.component.beanstalk.processors.DeleteCommand;
+import org.apache.camel.component.beanstalk.processors.PutCommand;
+import org.apache.camel.component.beanstalk.processors.ReleaseCommand;
+import org.apache.camel.component.beanstalk.processors.TouchCommand;
import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Test;
-import static org.junit.Assert.*;
-import static org.hamcrest.CoreMatchers.*;
-import static org.mockito.Mockito.*;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class ProducerTest extends BeanstalkMockTestSupport {
- final String testMessage = "hello, world";
@EndpointInject(uri = "beanstalk:tube")
protected BeanstalkEndpoint endpoint;
@@ -45,6 +52,8 @@ public class ProducerTest extends BeanstalkMockTestSupport {
@Produce(uri = "direct:start")
protected ProducerTemplate direct;
+ private String testMessage = "hello, world";
+
@Test
public void testPut() throws Exception {
final long priority = BeanstalkComponent.DEFAULT_PRIORITY;
@@ -58,7 +67,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
final Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(PutCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(PutCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() { // TODO: SetBodyProcessor(?)
public void process(Exchange exchange) {
@@ -83,7 +92,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(PutCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(PutCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOut, new Processor() { // TODO: SetBodyProcessor(?)
public void process(Exchange exchange) {
@@ -108,7 +117,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(PutCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(PutCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() { // TODO: SetBodyProcessor(?)
public void process(Exchange exchange) {
@@ -132,7 +141,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(BuryCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(BuryCommand.class));
when(client.bury(jobId, priority)).thenReturn(true);
@@ -153,10 +162,11 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(BuryCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(BuryCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() {
- public void process(Exchange exchange) {}
+ public void process(Exchange exchange) {
+ }
});
assertTrue("Exchange failed", exchange.isFailed());
@@ -173,7 +183,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(BuryCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(BuryCommand.class));
when(client.bury(jobId, priority)).thenReturn(true);
@@ -197,7 +207,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(DeleteCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(DeleteCommand.class));
when(client.delete(jobId)).thenReturn(true);
@@ -218,10 +228,11 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(DeleteCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(DeleteCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() {
- public void process(Exchange exchange) {}
+ public void process(Exchange exchange) {
+ }
});
assertTrue("Exchange failed", exchange.isFailed());
@@ -239,7 +250,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(ReleaseCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(ReleaseCommand.class));
when(client.release(jobId, priority, delay)).thenReturn(true);
@@ -260,10 +271,11 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(ReleaseCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(ReleaseCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() {
- public void process(Exchange exchange) {}
+ public void process(Exchange exchange) {
+ }
});
assertTrue("Exchange failed", exchange.isFailed());
@@ -281,7 +293,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(ReleaseCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(ReleaseCommand.class));
when(client.release(jobId, priority, delay)).thenReturn(true);
@@ -306,7 +318,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(TouchCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(TouchCommand.class));
when(client.touch(jobId)).thenReturn(true);
@@ -327,10 +339,11 @@ public class ProducerTest extends BeanstalkMockTestSupport {
Producer producer = endpoint.createProducer();
assertNotNull("Producer", producer);
assertThat("Producer class", producer, instanceOf(BeanstalkProducer.class));
- assertThat("Processor class", ((BeanstalkProducer)producer).command, instanceOf(TouchCommand.class));
+ assertThat("Processor class", ((BeanstalkProducer) producer).getCommand(), instanceOf(TouchCommand.class));
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() {
- public void process(Exchange exchange) {}
+ public void process(Exchange exchange) {
+ }
});
assertTrue("Exchange failed", exchange.isFailed());
@@ -350,7 +363,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
resultEndpoint.expectedMessageCount(1);
resultEndpoint.allMessages().body().isEqualTo(testMessage);
- resultEndpoint.allMessages().header(Headers.JOB_ID).isEqualTo(Long.valueOf(jobId));
+ resultEndpoint.allMessages().header(Headers.JOB_ID).isEqualTo(jobId);
direct.sendBodyAndHeader(testMessage, Headers.TIME_TO_RUN, timeToRun);
resultEndpoint.assertIsSatisfied();
@@ -370,12 +383,12 @@ public class ProducerTest extends BeanstalkMockTestSupport {
final long jobId = 113;
when(client.put(priority, delay, timeToRun, payload))
- .thenThrow(new BeanstalkException("test"))
- .thenReturn(jobId);
+ .thenThrow(new BeanstalkException("test"))
+ .thenReturn(jobId);
resultEndpoint.expectedMessageCount(1);
resultEndpoint.allMessages().body().isEqualTo(testMessage);
- resultEndpoint.allMessages().header(Headers.JOB_ID).isEqualTo(Long.valueOf(jobId));
+ resultEndpoint.allMessages().header(Headers.JOB_ID).isEqualTo(jobId);
direct.sendBodyAndHeader(testMessage, Headers.TIME_TO_RUN, timeToRun);
resultEndpoint.assertIsSatisfied();
@@ -392,7 +405,7 @@ public class ProducerTest extends BeanstalkMockTestSupport {
final long jobId = 111;
when(client.touch(jobId))
- .thenThrow(new BeanstalkException("test"));
+ .thenThrow(new BeanstalkException("test"));
endpoint.setCommand(BeanstalkComponent.COMMAND_TOUCH);
final Exchange exchange = template.send(endpoint, ExchangePattern.InOnly, new Processor() {