You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2009/04/03 11:14:04 UTC
svn commit: r761579 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/ main/java/org/apache/camel/component/dataset/
main/java/org/apache/camel/component/file/ main/java/org/apache/camel/impl/
test/java/org/apache/camel/component/file/
Author: davsclaus
Date: Fri Apr 3 09:14:03 2009
New Revision: 761579
URL: http://svn.apache.org/viewvc?rev=761579&view=rev
Log:
CAMEL-1514: Introduced PollingConsumerAware for eg file exchange to prepare itself for being polled and thus load the file content, so its not lost when the file is deleted/moved when the file consumer ends.
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java (with props)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContextAware.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetEndpoint.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileBinding.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileBinding.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConverter.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileDefaultBinding.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileExchange.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContextAware.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContextAware.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContextAware.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContextAware.java Fri Apr 3 09:14:03 2009
@@ -27,7 +27,7 @@
/**
* Injects the {@link CamelContext}
*
- * @param camelContext
+ * @param camelContext the Camel context
*/
void setCamelContext(CamelContext camelContext);
}
Added: camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java?rev=761579&view=auto
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java (added)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java Fri Apr 3 09:14:03 2009
@@ -0,0 +1,35 @@
+/**
+ * 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;
+
+/**
+ * A callback for {@link org.apache.camel.Exchange} when they are polled by a {@link org.apache.camel.PollingConsumer}.
+ * <p/>
+ * Is used if the {@link org.apache.camel.Exchange} needs to prepare, eg loading content that it otherwisw would have
+ * loaded lazy on demand.
+ *
+ * @version $Revision$
+ */
+public interface PollingConsumerAware {
+
+ /**
+ * Callback when an {@link org.apache.camel.Exchange} has been pooled.
+ *
+ * @param exchange the exchange
+ */
+ void exchangePolled(Exchange exchange);
+}
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/camel-core/src/main/java/org/apache/camel/PollingConsumerAware.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetEndpoint.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetEndpoint.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/dataset/DataSetEndpoint.java Fri Apr 3 09:14:03 2009
@@ -66,11 +66,6 @@
}
@Override
- public PollingConsumer createPollingConsumer() throws Exception {
- return new EventDrivenPollingConsumer(this);
- }
-
- @Override
public Consumer createConsumer(Processor processor) throws Exception {
return new DataSetConsumer(this, processor);
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileBinding.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileBinding.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileBinding.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/FileBinding.java Fri Apr 3 09:14:03 2009
@@ -17,16 +17,27 @@
package org.apache.camel.component.file;
import java.io.File;
+import java.io.IOException;
import java.io.Serializable;
+import org.apache.camel.Exchange;
+import org.apache.camel.NoTypeConversionAvailableException;
+import org.apache.camel.util.IOHelper;
+
/**
* File binding with the {@link java.io.File} type.
*/
public class FileBinding implements GenericFileBinding<File>, Serializable {
private File body;
+ private byte[] content;
public Object getBody(GenericFile<File> file) {
+ // if file content has been loaded then return it
+ if (content != null) {
+ return content;
+ }
+
// as we use java.io.File itself as the body (not loading its content into a OutputStream etc.)
// we just store a java.io.File handle to the actual file denoted by the
// file.getAbsoluteFilePath. We must do this as the original file consumed can be renamed before
@@ -43,4 +54,12 @@
// noop
}
+ public void loadContent(Exchange exchange, GenericFile<File> file) throws IOException {
+ try {
+ content = exchange.getContext().getTypeConverter().mandatoryConvertTo(byte[].class, file.getFile());
+ } catch (NoTypeConversionAvailableException e) {
+ throw IOHelper.createIOException("Cannot load file content: " + file.getAbsoluteFilePath(), e);
+ }
+ }
+
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileBinding.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileBinding.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileBinding.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileBinding.java Fri Apr 3 09:14:03 2009
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.file;
+import java.io.IOException;
+
+import org.apache.camel.Exchange;
+
/**
* Binding between the generic file and the body content.
*/
@@ -36,4 +40,13 @@
* @param body the body
*/
void setBody(GenericFile<T> file, Object body);
+
+ /**
+ * Loads the content from the file into memory
+ *
+ * @param exchange the current exchange
+ * @param file the file
+ * @throws java.io.IOException is thrown if the content could not be loaded
+ */
+ void loadContent(Exchange exchange, GenericFile<T> file) throws IOException;
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConverter.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConverter.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConverter.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileConverter.java Fri Apr 3 09:14:03 2009
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.file;
+import java.io.FileNotFoundException;
+
import org.apache.camel.Converter;
import org.apache.camel.Exchange;
import org.apache.camel.FallbackConverter;
@@ -37,6 +39,7 @@
public static Object convertTo(Class<?> type, Exchange exchange, Object value, TypeConverterRegistry registry) {
// use a fallback type converter so we can convert the embedded body if the value is GenericFile
if (GenericFile.class.isAssignableFrom(value.getClass())) {
+
GenericFile file = (GenericFile) value;
Class from = file.getBody().getClass();
@@ -44,6 +47,7 @@
if (from.isAssignableFrom(type)) {
return file.getBody();
}
+
// no then try to lookup a type converter
TypeConverter tc = registry.lookup(type, from);
if (tc != null) {
@@ -55,4 +59,9 @@
return null;
}
+ @Converter
+ public static String convertToString(GenericFile file, Exchange exchange) {
+ return exchange.getContext().getTypeConverter().convertTo(String.class, file.getBody());
+ }
+
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileDefaultBinding.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileDefaultBinding.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileDefaultBinding.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileDefaultBinding.java Fri Apr 3 09:14:03 2009
@@ -16,6 +16,10 @@
*/
package org.apache.camel.component.file;
+import java.io.IOException;
+
+import org.apache.camel.Exchange;
+
/**
* Default binding for generic file.
*/
@@ -30,4 +34,9 @@
public void setBody(GenericFile<T> file, Object body) {
this.body = body;
}
+
+ public void loadContent(Exchange exchange, GenericFile<T> file) throws IOException {
+ // noop as the body is already loaded
+ }
+
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileExchange.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileExchange.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileExchange.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/component/file/GenericFileExchange.java Fri Apr 3 09:14:03 2009
@@ -17,15 +17,18 @@
package org.apache.camel.component.file;
import java.io.File;
+import java.io.IOException;
import java.util.Date;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
+import org.apache.camel.PollingConsumerAware;
+import org.apache.camel.RuntimeExchangeException;
import org.apache.camel.impl.DefaultExchange;
-public class GenericFileExchange<T> extends DefaultExchange {
+public class GenericFileExchange<T> extends DefaultExchange implements PollingConsumerAware {
private GenericFile<T> file;
@@ -100,4 +103,12 @@
return new GenericFileExchange<T>(this, file);
}
+ public void exchangePolled(Exchange exchange) {
+ try {
+ // load content into memory
+ file.getBinding().loadContent(exchange, file);
+ } catch (IOException e) {
+ throw new RuntimeExchangeException("Cannot load content of file: " + file.getAbsoluteFilePath(), exchange, e);
+ }
+ }
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/EventDrivenPollingConsumer.java Fri Apr 3 09:14:03 2009
@@ -23,6 +23,7 @@
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
+import org.apache.camel.PollingConsumerAware;
import org.apache.camel.Processor;
import org.apache.camel.processor.Logger;
import org.apache.camel.spi.ExceptionHandler;
@@ -79,6 +80,10 @@
}
public void process(Exchange exchange) throws Exception {
+ if (exchange instanceof PollingConsumerAware) {
+ // callback that this exchange has been polled
+ ((PollingConsumerAware)exchange).exchangePolled(exchange);
+ }
queue.offer(exchange);
}
Modified: camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java?rev=761579&r1=761578&r2=761579&view=diff
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java (original)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeTemplateTest.java Fri Apr 3 09:14:03 2009
@@ -28,14 +28,22 @@
protected void setUp() throws Exception {
deleteDirectory("target/template");
super.setUp();
- template.sendBodyAndHeader("file://target/template/", "Hello World", Exchange.FILE_NAME, "hello.txt");
+ template.sendBodyAndHeader("file://target/template", "Hello World", Exchange.FILE_NAME, "hello.txt");
template.sendBodyAndHeader("file://target/template/", "Bye World", Exchange.FILE_NAME, "bye.txt");
}
public void testConsumeFileWithTemplate() throws Exception {
- Exchange out = consumer.receive("file://target/template?fileName=hello.txt");
+ Exchange out = consumer.receive("file://target/template?sortBy=file:name");
assertNotNull(out);
- assertEquals("Hello World", out.getIn().getBody(String.class));
+
+ Exchange out2 = consumer.receive("file://target/template?sortBy=file:name");
+ assertNotNull(out2);
+
+ String body = out.getIn().getBody(String.class);
+ String body2 = out2.getIn().getBody(String.class);
+
+ assertEquals("Bye World", body);
+ assertEquals("Hello World", body2);
}
}
\ No newline at end of file