Author: davsclaus
Date: Fri Jul 17 05:49:42 2009
New Revision: 794965
URL: http://svn.apache.org/viewvc?rev=794965&view=rev
Log:
CAMEL-1838: ftp consumer tries to better recover from lost connection.
Added:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
(with props)
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
(with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerPollStrategy.java
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerPollStrategy.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerPollStrategy.java?rev=794965&r1=794964&r2=794965&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerPollStrategy.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultPollingConsumerPollStrategy.java
Fri Jul 17 05:49:42 2009
@@ -29,7 +29,7 @@
*/
public class DefaultPollingConsumerPollStrategy implements
PollingConsumerPollStrategy {
- private static final transient Log LOG =
LogFactory.getLog(DefaultPollingConsumerPollStrategy.class);
+ protected final transient Log log = LogFactory.getLog(getClass());
public void begin(Consumer consumer, Endpoint endpoint) {
// noop
@@ -40,7 +40,7 @@
}
public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) throws Exception {
- LOG.warn("Consumer " + consumer + " could not poll endpoint: " +
endpoint.getEndpointUri() + " caused by: " + e.getMessage(), e);
+ log.warn("Consumer " + consumer + " could not poll endpoint: " +
endpoint.getEndpointUri() + " caused by: " + e.getMessage(), e);
// we do not want to retry
return false;
}
Modified:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java?rev=794965&r1=794964&r2=794965&view=diff
==============================================================================
---
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
(original)
+++
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/FtpOperations.java
Fri Jul 17 05:49:42 2009
@@ -38,6 +38,7 @@
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
/**
* FTP remote file operations
@@ -88,10 +89,24 @@
LOG.trace("Reconnect attempt #" + attempt + " connecting
to + " + configuration.remoteServerInformation());
}
client.connect(host, port);
- // yes we could connect
- connected = true;
+ // must check reply code if we are connected
+ int reply = client.getReplyCode();
+
+ if (FTPReply.isPositiveCompletion(reply)) {
+ // yes we could connect
+ connected = true;
+ } else {
+ // throw an exception to force the retry logic in the
catch exception block
+ throw new
GenericFileOperationFailedException(client.getReplyCode(),
client.getReplyString(), "Server refused connection");
+ }
} catch (Exception e) {
- GenericFileOperationFailedException failed = new
GenericFileOperationFailedException(client.getReplyCode(),
client.getReplyString(), e.getMessage(), e);
+ GenericFileOperationFailedException failed;
+ if (e instanceof GenericFileOperationFailedException) {
+ failed = (GenericFileOperationFailedException) e;
+ } else {
+ failed = new
GenericFileOperationFailedException(client.getReplyCode(),
client.getReplyString(), e.getMessage(), e);
+ }
+
if (LOG.isTraceEnabled()) {
LOG.trace("Could not connect due: " + failed.getMessage());
}
@@ -102,7 +117,7 @@
if (endpoint.getReconnectDelay() > 0) {
try {
Thread.sleep(endpoint.getReconnectDelay());
- } catch (InterruptedException e1) {
+ } catch (InterruptedException ie) {
// ignore
}
}
@@ -145,10 +160,17 @@
}
public void disconnect() throws GenericFileOperationFailedException {
+ // logout before disconnecting
try {
- client.disconnect();
+ client.logout();
} catch (IOException e) {
throw new
GenericFileOperationFailedException(client.getReplyCode(),
client.getReplyString(), e.getMessage(), e);
+ } finally {
+ try {
+ client.disconnect();
+ } catch (IOException e) {
+ throw new
GenericFileOperationFailedException(client.getReplyCode(),
client.getReplyString(), e.getMessage(), e);
+ }
}
}
Modified:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java?rev=794965&r1=794964&r2=794965&view=diff
==============================================================================
---
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
(original)
+++
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFileConsumer.java
Fri Jul 17 05:49:42 2009
@@ -30,6 +30,7 @@
public RemoteFileConsumer(RemoteFileEndpoint<T> endpoint, Processor
processor, RemoteFileOperations<T> operations) {
super(endpoint, processor, operations);
+ this.setPollStrategy(new RemoteFilePollingConsumerPollStrategy());
}
protected boolean prePollCheck() throws Exception {
@@ -45,12 +46,17 @@
@Override
protected void doStop() throws Exception {
super.doStop();
+ disconnect();
+ }
+ protected void disconnect() {
// disconnect when stopping
try {
if (((RemoteFileOperations) operations).isConnected()) {
loggedIn = false;
- log.debug("Disconnecting from: " + remoteServer());
+ if (log.isDebugEnabled()) {
+ log.debug("Disconnecting from: " + remoteServer());
+ }
((RemoteFileOperations) operations).disconnect();
}
} catch (GenericFileOperationFailedException e) {
Added:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java?rev=794965&view=auto
==============================================================================
---
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
(added)
+++
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
Fri Jul 17 05:49:42 2009
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.file.remote;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultPollingConsumerPollStrategy;
+
+/**
+ * Remote file consumer polling strategy that attempts to help recovering from
lost connections.
+ *
+ * @version $Revision$
+ */
+public class RemoteFilePollingConsumerPollStrategy extends
DefaultPollingConsumerPollStrategy {
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) throws Exception {
+ // disconnect from the server to force it to re login at next poll to
recover
+ RemoteFileConsumer rfc = (RemoteFileConsumer) consumer;
+ log.warn("Trying to recover by disconnecting from remote server
forcing a re-connect at next poll: " + rfc.remoteServer());
+ rfc.disconnect();
+ return super.rollback(consumer, endpoint, retryCounter, e);
+ }
+
+}
Propchange:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/components/camel-ftp/src/main/java/org/apache/camel/component/file/remote/RemoteFilePollingConsumerPollStrategy.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added:
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java?rev=794965&view=auto
==============================================================================
---
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
(added)
+++
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
Fri Jul 17 05:49:42 2009
@@ -0,0 +1,93 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.file.remote;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.junit.Test;
+
+/**
+ * Simulate network issues by using a custom poll startegy to force exceptions
+ * occuring during poll.
+ *
+ * @version $Revision$
+ */
+public class FromFtpSimulateNetworkIssueRecoverTest extends
FtpServerTestSupport {
+
+ private static int counter;
+ private static int rollback;
+
+ private String getFtpUrl() {
+ return "ftp://ad...@localhost:"; + getPort() +
"/recover?password=admin&pollStrategy=#myPoll";
+ }
+
+ @Override
+ public void setUp() throws Exception {
+ deleteDirectory("./res/home/recover");
+ super.setUp();
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry jndi = super.createRegistry();
+ jndi.bind("myPoll", new MyPollStrategy());
+ return jndi;
+ }
+
+ @Test
+ public void testFtpRecover() throws Exception {
+ // should be able to download the file after recovering
+ MockEndpoint resultEndpoint = getMockEndpoint("mock:result");
+ resultEndpoint.expectedMinimumMessageCount(3);
+
+ template.sendBody(getFtpUrl(), "Hello World");
+
+ resultEndpoint.assertIsSatisfied();
+
+ assertEquals(3, counter);
+ assertEquals(2, rollback);
+ }
+
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() throws Exception {
+ from(getFtpUrl()).to("mock:result");
+ }
+ };
+ }
+
+ public class MyPollStrategy extends RemoteFilePollingConsumerPollStrategy {
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint) {
+ counter++;
+ if (counter < 3) {
+ throw new IllegalArgumentException("Forced by unit test");
+ }
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) throws Exception {
+ rollback++;
+ return super.rollback(consumer, endpoint, retryCounter, e);
+ }
+ }
+
+}
Propchange:
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/components/camel-ftp/src/test/java/org/apache/camel/component/file/remote/FromFtpSimulateNetworkIssueRecoverTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
