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 <[email protected]> Authored: Sun Sep 21 09:49:45 2014 +0200 Committer: Claus Ibsen <[email protected]> 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:[email protected]";>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:[email protected]";>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:[email protected]";>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:[email protected]";>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:[email protected]";>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:[email protected]";>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:[email protected]";>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:[email protected]";>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() {
