You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2017/02/09 17:25:23 UTC
[2/2] camel git commit: CAMEL-10790: gridfs should be renamed to
mongodb-gridfs
CAMEL-10790: gridfs should be renamed to mongodb-gridfs
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e067ae91
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e067ae91
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e067ae91
Branch: refs/heads/master
Commit: e067ae91d7366de6027006bf5885e47c54fb8f8b
Parents: 708e139
Author: lburgazzoli <lb...@gmail.com>
Authored: Thu Feb 9 18:24:55 2017 +0100
Committer: lburgazzoli <lb...@gmail.com>
Committed: Thu Feb 9 18:24:55 2017 +0100
----------------------------------------------------------------------
components/camel-mongodb-gridfs/pom.xml | 2 +-
.../src/main/docs/gridfs-component.adoc | 197 ----------
.../src/main/docs/mongodb-gridfs-component.adoc | 197 ++++++++++
.../camel/component/gridfs/GridFsComponent.java | 45 ---
.../camel/component/gridfs/GridFsConsumer.java | 177 ---------
.../camel/component/gridfs/GridFsEndpoint.java | 369 -------------------
.../camel/component/gridfs/GridFsProducer.java | 147 --------
.../camel/component/gridfs/QueryStrategy.java | 25 --
.../mongodb/gridfs/GridFsComponent.java | 45 +++
.../mongodb/gridfs/GridFsConsumer.java | 177 +++++++++
.../mongodb/gridfs/GridFsEndpoint.java | 369 +++++++++++++++++++
.../mongodb/gridfs/GridFsProducer.java | 147 ++++++++
.../component/mongodb/gridfs/QueryStrategy.java | 25 ++
.../services/org/apache/camel/component/gridfs | 18 -
.../org/apache/camel/component/mongodb-gridfs | 18 +
.../component/gridfs/AbstractMongoDbTest.java | 62 ----
.../gridfs/EmbedMongoConfiguration.java | 58 ---
.../component/gridfs/GridFsConsumerTest.java | 99 -----
.../gridfs/GridFsProducerOperationsTest.java | 74 ----
.../mongodb/gridfs/AbstractMongoDbTest.java | 62 ++++
.../mongodb/gridfs/EmbedMongoConfiguration.java | 65 ++++
.../mongodb/gridfs/GridFsConsumerTest.java | 99 +++++
.../gridfs/GridFsProducerOperationsTest.java | 74 ++++
components/readme.adoc | 6 +-
docs/user-manual/en/SUMMARY.md | 2 +-
parent/pom.xml | 1 +
.../GridFsComponentAutoConfiguration.java | 80 ----
.../GridFsComponentAutoConfiguration.java | 80 ++++
...dditional-spring-configuration-metadata.json | 6 +-
29 files changed, 1367 insertions(+), 1359 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/pom.xml b/components/camel-mongodb-gridfs/pom.xml
index e9d7bc9..a5610ca 100644
--- a/components/camel-mongodb-gridfs/pom.xml
+++ b/components/camel-mongodb-gridfs/pom.xml
@@ -50,7 +50,7 @@
<dependency>
<groupId>org.mongodb</groupId>
<artifactId>mongo-java-driver</artifactId>
- <version>3.2.2</version>
+ <version>${mongo-java-driver32-version}</version>
</dependency>
<!-- test -->
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/docs/gridfs-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/docs/gridfs-component.adoc b/components/camel-mongodb-gridfs/src/main/docs/gridfs-component.adoc
deleted file mode 100644
index 5a5f244..0000000
--- a/components/camel-mongodb-gridfs/src/main/docs/gridfs-component.adoc
+++ /dev/null
@@ -1,197 +0,0 @@
-## MongoDB GridFS Component
-
-*Available as of Camel version 2.18.0*
-
-Maven users will need to add the following dependency to their `pom.xml`
-for this component:
-
-[source,xml]
-------------------------------------------------------------
-<dependency>
- <groupId>org.apache.camel</groupId>
- <artifactId>camel-mongodb-gridfs</artifactId>
- <version>x.y.z</version>
- <!-- use the same version as your Camel core version -->
-</dependency>
-------------------------------------------------------------
-
-### URI format
-
-[source,java]
-------------------------------------------------------------------------------
-gridfs:connectionBean?database=databaseName&bucket=bucketName[&moreOptions...]
-------------------------------------------------------------------------------
-
-### MongoDB GridFS options
-
-
-// component options: START
-The MongoDB GridFS component has no options.
-// component options: END
-
-
-
-
-// endpoint options: START
-The MongoDB GridFS component supports 18 endpoint options which are listed below:
-
-{% raw %}
-[width="100%",cols="2,1,1m,1m,5",options="header"]
-|=======================================================================
-| Name | Group | Default | Java Type | Description
-| connectionBean | common | | String | *Required* Name of com.mongodb.Mongo to use.
-| bucket | common | fs | String | Sets the name of the GridFS bucket within the database. Default is fs.
-| database | common | | String | *Required* Sets the name of the MongoDB database to target
-| readPreference | common | | ReadPreference | Sets a MongoDB ReadPreference on the Mongo connection. Read preferences set directly on the connection will be overridden by this setting. The link com.mongodb.ReadPreferencevalueOf(String) utility method is used to resolve the passed readPreference value. Some examples for the possible values are nearest primary or secondary etc.
-| writeConcern | common | | WriteConcern | Set the WriteConcern for write operations on MongoDB using the standard ones. Resolved from the fields of the WriteConcern class by calling the link WriteConcernvalueOf(String) method.
-| writeConcernRef | common | | WriteConcern | Set the WriteConcern for write operations on MongoDB passing in the bean ref to a custom WriteConcern which exists in the Registry. You can also use standard WriteConcerns by passing in their key. See the link setWriteConcern(String) setWriteConcern method.
-| bridgeErrorHandler | consumer | false | boolean | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored.
-| delay | consumer | 500 | long | Sets the delay between polls within the Consumer. Default is 500ms
-| fileAttributeName | consumer | camel-processed | String | If the QueryType uses a FileAttribute this sets the name of the attribute that is used. Default is camel-processed.
-| initialDelay | consumer | 1000 | long | Sets the initialDelay before the consumer will start polling. Default is 1000ms
-| persistentTSCollection | consumer | camel-timestamps | String | If the QueryType uses a persistent timestamp this sets the name of the collection within the DB to store the timestamp.
-| persistentTSObject | consumer | camel-timestamp | String | If the QueryType uses a persistent timestamp this is the ID of the object in the collection to store the timestamp.
-| query | consumer | | String | Additional query parameters (in JSON) that are used to configure the query used for finding files in the GridFsConsumer
-| queryStrategy | consumer | TimeStamp | QueryStrategy | Sets the QueryStrategy that is used for polling for new files. Default is Timestamp
-| exceptionHandler | consumer (advanced) | | ExceptionHandler | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored.
-| exchangePattern | consumer (advanced) | | ExchangePattern | Sets the exchange pattern when the consumer creates an exchange.
-| operation | producer | | String | Sets the operation this endpoint will execute against GridRS.
-| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
-|=======================================================================
-{% endraw %}
-// endpoint options: END
-
-
-
-### Configuration of database in Spring XML
-
-The following Spring XML creates a bean defining the connection to a
-MongoDB instance.
-
-[source,xml]
-----------------------------------------------------------------------------------------------------------------------------------
-<beans xmlns="http://www.springframework.org/schema/beans"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
- <bean id="mongoBean" class="com.mongodb.Mongo">
- <constructor-arg name="host" value="${mongodb.host}" />
- <constructor-arg name="port" value="${mongodb.port}" />
- </bean>
-</beans>
-----------------------------------------------------------------------------------------------------------------------------------
-
-### Sample route
-
-The following route defined in Spring XML executes the operation
-link:mongodb-gridfs.html[*findOne*] on a collection.
-
-*Get a file from GridFS*
-
-[source,xml]
-----------------------------------------------------------------------------------
-<route>
- <from uri="direct:start" />
- <!-- using bean 'mongoBean' defined above -->
- <to uri="gridfs:mongoBean?database=${mongodb.database}&operation=findOne" />
- <to uri="direct:result" />
-</route>
-----------------------------------------------------------------------------------
-
-�
-
-### GridFS operations - producer endpoint
-
-#### count
-
-Returns the total number of file in the collection, returning an Integer
-as the OUT message body. +
- +
-
-[source,java]
----------------------------------------------------------------------------------
-// from("direct:count").to("gridfs?database=tickets&operation=count");
-Integer result = template.requestBodyAndHeader("direct:count", "irrelevantBody");
-assertTrue("Result is not of type Long", result instanceof Integer);
----------------------------------------------------------------------------------
-
-You can provide a filename header to provide a count of files matching
-that filename.
-
-[source,java]
--------------------------------------------------------------------------------
-Map<String, Object> headers = new HashMap<String, Object>();
-headers.put(Exchange.FILE_NAME, "filename.txt");
-Integer count = template.requestBodyAndHeaders("direct:count", query, headers);
--------------------------------------------------------------------------------
-
-#### listAll
-
-Returns an Reader that lists all the filenames and their IDs in a tab
-separated stream.
-
-[source,java]
-----------------------------------------------------------------------------------
-// from("direct:listAll").to("gridfs?database=tickets&operation=listAll");
-Reader result = template.requestBodyAndHeader("direct:listAll", "irrelevantBody");
-
-filename1.txt 1252314321
-filename2.txt 2897651254
-----------------------------------------------------------------------------------
-
-�
-
-#### *findOne*
-
-Finds a file in the GridFS system and sets the body to an InputStream of
-the content. � Also provides the metadata has headers. �It uses
-Exchange.FILE_NAME from the incoming headers to determine the file to
-find.
-
-[source,java]
--------------------------------------------------------------------------------------------------
-// from("direct:findOne").to("gridfs?database=tickets&operation=findOne");
-Map<String, Object> headers = new HashMap<String, Object>();
-headers.put(Exchange.FILE_NAME, "filename.txt");
-InputStream result = template.requestBodyAndHeaders("direct:findOne", "irrelevantBody", headers);
--------------------------------------------------------------------------------------------------
-
-�
-
-#### create
-
-Creates a new file in the GridFs database. It uses the
-Exchange.FILE_NAME from the incoming headers for the name and the body
-contents (as an InputStream) as the content.
-
-[source,java]
-------------------------------------------------------------------------
-// from("direct:create").to("gridfs?database=tickets&operation=create");
-Map<String, Object> headers = new HashMap<String, Object>();
-headers.put(Exchange.FILE_NAME, "filename.txt");
-InputStream stream = ... the data for the file ...
-template.requestBodyAndHeaders("direct:create", stream, headers);
-------------------------------------------------------------------------
-
-#### remove
-
-Removes a file from the GridFS database.
-
-[source,java]
-------------------------------------------------------------------------
-// from("direct:remove").to("gridfs?database=tickets&operation=remove");
-Map<String, Object> headers = new HashMap<String, Object>();
-headers.put(Exchange.FILE_NAME, "filename.txt");
-template.requestBodyAndHeaders("direct:remove", "", headers);
-------------------------------------------------------------------------
-
-### GridFS Consumer
-
-See also
-
-* http://www.mongodb.org/[MongoDB website]
-* http://en.wikipedia.org/wiki/NoSQL[NoSQL Wikipedia article]
-* http://api.mongodb.org/java/current/[MongoDB Java driver API docs -
-current version]
-*
-http://svn.apache.org/viewvc/camel/trunk/components/camel-mongodb/src/test/[Unit
-tests] for more examples of usage
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc b/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
new file mode 100644
index 0000000..0dcd751
--- /dev/null
+++ b/components/camel-mongodb-gridfs/src/main/docs/mongodb-gridfs-component.adoc
@@ -0,0 +1,197 @@
+## MongoDB GridFS Component
+
+*Available as of Camel version 2.18.0*
+
+Maven users will need to add the following dependency to their `pom.xml`
+for this component:
+
+[source,xml]
+------------------------------------------------------------
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-mongodb-gridfs</artifactId>
+ <version>x.y.z</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+------------------------------------------------------------
+
+### URI format
+
+[source,java]
+------------------------------------------------------------------------------
+mongodb-gridfs:connectionBean?database=databaseName&bucket=bucketName[&moreOptions...]
+------------------------------------------------------------------------------
+
+### MongoDB GridFS options
+
+
+// component options: START
+The MongoDB GridFS component has no options.
+// component options: END
+
+
+
+
+// endpoint options: START
+The MongoDB GridFS component supports 18 endpoint options which are listed below:
+
+{% raw %}
+[width="100%",cols="2,1,1m,1m,5",options="header"]
+|=======================================================================
+| Name | Group | Default | Java Type | Description
+| connectionBean | common | | String | *Required* Name of com.mongodb.Mongo to use.
+| bucket | common | fs | String | Sets the name of the GridFS bucket within the database. Default is fs.
+| database | common | | String | *Required* Sets the name of the MongoDB database to target
+| readPreference | common | | ReadPreference | Sets a MongoDB ReadPreference on the Mongo connection. Read preferences set directly on the connection will be overridden by this setting. The link com.mongodb.ReadPreferencevalueOf(String) utility method is used to resolve the passed readPreference value. Some examples for the possible values are nearest primary or secondary etc.
+| writeConcern | common | | WriteConcern | Set the WriteConcern for write operations on MongoDB using the standard ones. Resolved from the fields of the WriteConcern class by calling the link WriteConcernvalueOf(String) method.
+| writeConcernRef | common | | WriteConcern | Set the WriteConcern for write operations on MongoDB passing in the bean ref to a custom WriteConcern which exists in the Registry. You can also use standard WriteConcerns by passing in their key. See the link setWriteConcern(String) setWriteConcern method.
+| bridgeErrorHandler | consumer | false | boolean | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored.
+| delay | consumer | 500 | long | Sets the delay between polls within the Consumer. Default is 500ms
+| fileAttributeName | consumer | camel-processed | String | If the QueryType uses a FileAttribute this sets the name of the attribute that is used. Default is camel-processed.
+| initialDelay | consumer | 1000 | long | Sets the initialDelay before the consumer will start polling. Default is 1000ms
+| persistentTSCollection | consumer | camel-timestamps | String | If the QueryType uses a persistent timestamp this sets the name of the collection within the DB to store the timestamp.
+| persistentTSObject | consumer | camel-timestamp | String | If the QueryType uses a persistent timestamp this is the ID of the object in the collection to store the timestamp.
+| query | consumer | | String | Additional query parameters (in JSON) that are used to configure the query used for finding files in the GridFsConsumer
+| queryStrategy | consumer | TimeStamp | QueryStrategy | Sets the QueryStrategy that is used for polling for new files. Default is Timestamp
+| exceptionHandler | consumer (advanced) | | ExceptionHandler | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored.
+| exchangePattern | consumer (advanced) | | ExchangePattern | Sets the exchange pattern when the consumer creates an exchange.
+| operation | producer | | String | Sets the operation this endpoint will execute against GridRS.
+| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
+|=======================================================================
+{% endraw %}
+// endpoint options: END
+
+
+
+### Configuration of database in Spring XML
+
+The following Spring XML creates a bean defining the connection to a
+MongoDB instance.
+
+[source,xml]
+----------------------------------------------------------------------------------------------------------------------------------
+<beans xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+ <bean id="mongoBean" class="com.mongodb.Mongo">
+ <constructor-arg name="host" value="${mongodb.host}" />
+ <constructor-arg name="port" value="${mongodb.port}" />
+ </bean>
+</beans>
+----------------------------------------------------------------------------------------------------------------------------------
+
+### Sample route
+
+The following route defined in Spring XML executes the operation
+link:mongodb-gridfs.html[*findOne*] on a collection.
+
+*Get a file from GridFS*
+
+[source,xml]
+----------------------------------------------------------------------------------
+<route>
+ <from uri="direct:start" />
+ <!-- using bean 'mongoBean' defined above -->
+ <to uri="mongodb-gridfs:mongoBean?database=${mongodb.database}&operation=findOne" />
+ <to uri="direct:result" />
+</route>
+----------------------------------------------------------------------------------
+
+�
+
+### GridFS operations - producer endpoint
+
+#### count
+
+Returns the total number of file in the collection, returning an Integer
+as the OUT message body. +
+ +
+
+[source,java]
+---------------------------------------------------------------------------------
+// from("direct:count").to("mongodb-gridfs?database=tickets&operation=count");
+Integer result = template.requestBodyAndHeader("direct:count", "irrelevantBody");
+assertTrue("Result is not of type Long", result instanceof Integer);
+---------------------------------------------------------------------------------
+
+You can provide a filename header to provide a count of files matching
+that filename.
+
+[source,java]
+-------------------------------------------------------------------------------
+Map<String, Object> headers = new HashMap<String, Object>();
+headers.put(Exchange.FILE_NAME, "filename.txt");
+Integer count = template.requestBodyAndHeaders("direct:count", query, headers);
+-------------------------------------------------------------------------------
+
+#### listAll
+
+Returns an Reader that lists all the filenames and their IDs in a tab
+separated stream.
+
+[source,java]
+----------------------------------------------------------------------------------
+// from("direct:listAll").to("mongodb-gridfs?database=tickets&operation=listAll");
+Reader result = template.requestBodyAndHeader("direct:listAll", "irrelevantBody");
+
+filename1.txt 1252314321
+filename2.txt 2897651254
+----------------------------------------------------------------------------------
+
+�
+
+#### *findOne*
+
+Finds a file in the GridFS system and sets the body to an InputStream of
+the content. � Also provides the metadata has headers. �It uses
+Exchange.FILE_NAME from the incoming headers to determine the file to
+find.
+
+[source,java]
+-------------------------------------------------------------------------------------------------
+// from("direct:findOne").to("mongodb-gridfs?database=tickets&operation=findOne");
+Map<String, Object> headers = new HashMap<String, Object>();
+headers.put(Exchange.FILE_NAME, "filename.txt");
+InputStream result = template.requestBodyAndHeaders("direct:findOne", "irrelevantBody", headers);
+-------------------------------------------------------------------------------------------------
+
+�
+
+#### create
+
+Creates a new file in the GridFs database. It uses the
+Exchange.FILE_NAME from the incoming headers for the name and the body
+contents (as an InputStream) as the content.
+
+[source,java]
+------------------------------------------------------------------------
+// from("direct:create").to("mongodb-gridfs?database=tickets&operation=create");
+Map<String, Object> headers = new HashMap<String, Object>();
+headers.put(Exchange.FILE_NAME, "filename.txt");
+InputStream stream = ... the data for the file ...
+template.requestBodyAndHeaders("direct:create", stream, headers);
+------------------------------------------------------------------------
+
+#### remove
+
+Removes a file from the GridFS database.
+
+[source,java]
+------------------------------------------------------------------------
+// from("direct:remove").to("mongodb-gridfs?database=tickets&operation=remove");
+Map<String, Object> headers = new HashMap<String, Object>();
+headers.put(Exchange.FILE_NAME, "filename.txt");
+template.requestBodyAndHeaders("direct:remove", "", headers);
+------------------------------------------------------------------------
+
+### GridFS Consumer
+
+See also
+
+* http://www.mongodb.org/[MongoDB website]
+* http://en.wikipedia.org/wiki/NoSQL[NoSQL Wikipedia article]
+* http://api.mongodb.org/java/current/[MongoDB Java driver API docs -
+current version]
+*
+http://svn.apache.org/viewvc/camel/trunk/components/camel-mongodb/src/test/[Unit
+tests] for more examples of usage
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsComponent.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsComponent.java
deleted file mode 100644
index 5b91313..0000000
--- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsComponent.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * 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.gridfs;
-
-import java.util.Map;
-
-import org.apache.camel.Endpoint;
-import org.apache.camel.impl.UriEndpointComponent;
-
-
-public class GridFsComponent extends UriEndpointComponent {
-
- public GridFsComponent() {
- super(GridFsEndpoint.class);
- }
-
- protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
-
- GridFsEndpoint endpoint = new GridFsEndpoint(uri, this);
- endpoint.setConnectionBean(remaining);
- setProperties(endpoint, parameters);
-
- return endpoint;
- }
-
- @Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsConsumer.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsConsumer.java
deleted file mode 100644
index 6df4391..0000000
--- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsConsumer.java
+++ /dev/null
@@ -1,177 +0,0 @@
-/**
- * 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.gridfs;
-
-import java.io.InputStream;
-import java.util.concurrent.ExecutorService;
-
-import com.mongodb.BasicDBObject;
-import com.mongodb.BasicDBObjectBuilder;
-import com.mongodb.DBCollection;
-import com.mongodb.DBCursor;
-import com.mongodb.DBObject;
-import com.mongodb.MongoException;
-import com.mongodb.gridfs.GridFSDBFile;
-import com.mongodb.util.JSON;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.Processor;
-import org.apache.camel.impl.DefaultConsumer;
-
-/**
- *
- */
-public class GridFsConsumer extends DefaultConsumer implements Runnable {
- final GridFsEndpoint endpoint;
- private ExecutorService executor;
-
- /**
- * @param endpoint
- * @param processor
- */
- public GridFsConsumer(GridFsEndpoint endpoint, Processor processor) {
- super(endpoint, processor);
- this.endpoint = endpoint;
- }
-
- @Override
- protected void doStop() throws Exception {
- super.doStop();
- if (executor != null) {
- executor.shutdown();
- executor = null;
- }
- }
-
- @Override
- protected void doStart() throws Exception {
- super.doStart();
- executor = endpoint.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, endpoint.getEndpointUri(), 1);
- executor.execute(this);
- }
-
- @Override
- public void run() {
- DBCursor c = null;
- java.util.Date fromDate = null;
-
- QueryStrategy s = endpoint.getQueryStrategy();
- boolean usesTimestamp = s != QueryStrategy.FileAttribute;
- boolean persistsTimestamp = s == QueryStrategy.PersistentTimestamp || s == QueryStrategy.PersistentTimestampAndFileAttribute;
- boolean usesAttribute = s == QueryStrategy.FileAttribute
- || s == QueryStrategy.TimeStampAndFileAttribute
- || s == QueryStrategy.PersistentTimestampAndFileAttribute;
-
- DBCollection ptsCollection = null;
- DBObject persistentTimestamp = null;
- if (persistsTimestamp) {
- ptsCollection = endpoint.getDB().getCollection(endpoint.getPersistentTSCollection());
- // ensure standard indexes as long as collections are small
- try {
- if (ptsCollection.count() < 1000) {
- ptsCollection.createIndex(new BasicDBObject("id", 1));
- }
- } catch (MongoException e) {
- //TODO: Logging
- }
- persistentTimestamp = ptsCollection.findOne(new BasicDBObject("id", endpoint.getPersistentTSObject()));
- if (persistentTimestamp == null) {
- persistentTimestamp = new BasicDBObject("id", endpoint.getPersistentTSObject());
- fromDate = new java.util.Date();
- persistentTimestamp.put("timestamp", fromDate);
- ptsCollection.save(persistentTimestamp);
- }
- fromDate = (java.util.Date)persistentTimestamp.get("timestamp");
- } else if (usesTimestamp) {
- fromDate = new java.util.Date();
- }
- try {
- Thread.sleep(endpoint.getInitialDelay());
- while (isStarted()) {
- if (c == null || c.getCursorId() == 0) {
- if (c != null) {
- c.close();
- }
- String queryString = endpoint.getQuery();
- DBObject query;
- if (queryString == null) {
- query = new BasicDBObject();
- } else {
- query = (DBObject) JSON.parse(queryString);
- }
- if (usesTimestamp) {
- query.put("uploadDate", new BasicDBObject("$gt", fromDate));
- }
- if (usesAttribute) {
- query.put(endpoint.getFileAttributeName(), null);
- }
- c = endpoint.getFilesCollection().find(query);
- }
- boolean dateModified = false;
- while (c.hasNext() && isStarted()) {
- GridFSDBFile file = (GridFSDBFile)c.next();
- GridFSDBFile forig = file;
- if (usesAttribute) {
- file.put(endpoint.getFileAttributeName(), "processing");
- DBObject q = BasicDBObjectBuilder.start("_id", file.getId()).append("camel-processed", null).get();
- forig = (GridFSDBFile)endpoint.getFilesCollection().findAndModify(q, null, null, false, file, true, false);
- }
- if (forig != null) {
- file = endpoint.getGridFs().findOne(new BasicDBObject("_id", file.getId()));
-
- Exchange exchange = endpoint.createExchange();
- exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_METADATA, JSON.serialize(file.getMetaData()));
- exchange.getIn().setHeader(Exchange.FILE_CONTENT_TYPE, file.getContentType());
- exchange.getIn().setHeader(Exchange.FILE_LENGTH, file.getLength());
- exchange.getIn().setHeader(Exchange.FILE_LAST_MODIFIED, file.getUploadDate());
- exchange.getIn().setBody(file.getInputStream(), InputStream.class);
- try {
- getProcessor().process(exchange);
- //System.out.println("Processing " + file.getFilename());
- if (usesAttribute) {
- forig.put(endpoint.getFileAttributeName(), "done");
- endpoint.getFilesCollection().save(forig);
- }
- if (usesTimestamp) {
- if (file.getUploadDate().compareTo(fromDate) > 0) {
- fromDate = file.getUploadDate();
- dateModified = true;
- }
- }
- } catch (Exception e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
- }
- if (persistsTimestamp && dateModified) {
- persistentTimestamp.put("timestamp", fromDate);
- ptsCollection.save(persistentTimestamp);
- }
- Thread.sleep(endpoint.getDelay());
- }
- } catch (Throwable e1) {
- // TODO Auto-generated catch block
- e1.printStackTrace();
- }
- if (c != null) {
- c.close();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsEndpoint.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsEndpoint.java
deleted file mode 100644
index 6f4057d..0000000
--- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsEndpoint.java
+++ /dev/null
@@ -1,369 +0,0 @@
-/**
- * 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.gridfs;
-
-import com.mongodb.DB;
-import com.mongodb.DBCollection;
-import com.mongodb.Mongo;
-import com.mongodb.MongoClient;
-import com.mongodb.ReadPreference;
-import com.mongodb.WriteConcern;
-import com.mongodb.gridfs.GridFS;
-import org.apache.camel.Consumer;
-import org.apache.camel.Processor;
-import org.apache.camel.Producer;
-import org.apache.camel.impl.DefaultEndpoint;
-import org.apache.camel.spi.Metadata;
-import org.apache.camel.spi.UriEndpoint;
-import org.apache.camel.spi.UriParam;
-import org.apache.camel.spi.UriPath;
-import org.apache.camel.util.CamelContextHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Component for working with MongoDB GridFS.
- */
-@UriEndpoint(firstVersion = "2.18.0", scheme = "gridfs", title = "MongoDB GridFS", syntax = "gridfs:connectionBean", label = "database,nosql")
-public class GridFsEndpoint extends DefaultEndpoint {
-
- public static final String GRIDFS_OPERATION = "gridfs.operation";
- public static final String GRIDFS_METADATA = "gridfs.metadata";
- public static final String GRIDFS_CHUNKSIZE = "gridfs.chunksize";
- public static final String GRIDFS_FILE_ID_PRODUCED = "gridfs.fileid";
-
- private static final Logger LOG = LoggerFactory.getLogger(GridFsEndpoint.class);
-
- @UriPath @Metadata(required = "true")
- private String connectionBean;
- @UriParam @Metadata(required = "true")
- private String database;
- @UriParam(defaultValue = GridFS.DEFAULT_BUCKET)
- private String bucket;
- @UriParam(enums = "ACKNOWLEDGED,W1,W2,W3,UNACKNOWLEDGED,JOURNALED,MAJORITY,SAFE")
- private WriteConcern writeConcern;
- @UriParam
- private WriteConcern writeConcernRef;
- @UriParam
- private ReadPreference readPreference;
-
- @UriParam(label = "producer")
- private String operation;
-
- @UriParam(label = "consumer")
- private String query;
- @UriParam(label = "consumer", defaultValue = "1000")
- private long initialDelay = 1000;
- @UriParam(label = "consumer", defaultValue = "500")
- private long delay = 500;
-
- @UriParam(label = "consumer", defaultValue = "TimeStamp")
- private QueryStrategy queryStrategy = QueryStrategy.TimeStamp;
- @UriParam(label = "consumer", defaultValue = "camel-timestamps")
- private String persistentTSCollection = "camel-timestamps";
- @UriParam(label = "consumer", defaultValue = "camel-timestamp")
- private String persistentTSObject = "camel-timestamp";
- @UriParam(label = "consumer", defaultValue = "camel-processed")
- private String fileAttributeName = "camel-processed";
-
-
- private Mongo mongoConnection;
- private DB db;
- private GridFS gridFs;
- private DBCollection filesCollection;
-
- public GridFsEndpoint(String uri, GridFsComponent component) {
- super(uri, component);
- }
-
- @Override
- public Producer createProducer() throws Exception {
- initializeConnection();
- return new GridFsProducer(this);
- }
-
- @Override
- public Consumer createConsumer(Processor processor) throws Exception {
- initializeConnection();
- return new GridFsConsumer(this, processor);
- }
-
- public boolean isSingleton() {
- return true;
- }
-
- @SuppressWarnings("deprecation")
- public void initializeConnection() throws Exception {
- LOG.info("Initialize GridFS endpoint: {}", this.toString());
- if (database == null) {
- throw new IllegalStateException("Missing required endpoint configuration: database");
- }
- db = mongoConnection.getDB(database);
- if (db == null) {
- throw new IllegalStateException("Could not initialize GridFsComponent. Database " + database + " does not exist.");
- }
- gridFs = new GridFS(db, bucket == null ? GridFS.DEFAULT_BUCKET : bucket) {
- {
- filesCollection = getFilesCollection();
- }
- };
- }
-
-
- @Override
- protected void doStart() throws Exception {
- if (writeConcern != null && writeConcernRef != null) {
- String msg = "Cannot set both writeConcern and writeConcernRef at the same time. Respective values: " + writeConcern
- + ", " + writeConcernRef + ". Aborting initialization.";
- throw new IllegalArgumentException(msg);
- }
- mongoConnection = CamelContextHelper.mandatoryLookup(getCamelContext(), connectionBean, MongoClient.class);
- LOG.debug("Resolved the connection with the name {} as {}", connectionBean, mongoConnection);
- setWriteReadOptionsOnConnection();
- super.doStart();
- }
-
- @Override
- protected void doStop() throws Exception {
- super.doStop();
- if (mongoConnection != null) {
- LOG.debug("Closing connection");
- mongoConnection.close();
- }
- }
-
- private void setWriteReadOptionsOnConnection() {
- // Set the WriteConcern
- if (writeConcern != null) {
- mongoConnection.setWriteConcern(writeConcern);
- } else if (writeConcernRef != null) {
- mongoConnection.setWriteConcern(writeConcernRef);
- }
-
- // Set the ReadPreference
- if (readPreference != null) {
- mongoConnection.setReadPreference(readPreference);
- }
- }
-
-
-
-
- // ======= Getters and setters ===============================================
- public String getConnectionBean() {
- return connectionBean;
- }
- /**
- * Name of {@link com.mongodb.Mongo} to use.
- */
- public void setConnectionBean(String connectionBean) {
- this.connectionBean = connectionBean;
- }
-
- public Mongo getMongoConnection() {
- return mongoConnection;
- }
- /**
- * Sets the Mongo instance that represents the backing connection
- *
- * @param mongoConnection the connection to the database
- */
- public void setMongoConnection(Mongo mongoConnection) {
- this.mongoConnection = mongoConnection;
- }
-
- public DB getDB() {
- return db;
- }
-
- public String getDatabase() {
- return database;
- }
- /**
- * Sets the name of the MongoDB database to target
- *
- * @param database name of the MongoDB database
- */
- public void setDatabase(String database) {
- this.database = database;
- }
- /**
- * Sets the name of the GridFS bucket within the database. Default is "fs".
- *
- * @param database name of the MongoDB database
- */
- public String getBucket() {
- return bucket;
- }
- public void setBucket(String bucket) {
- this.bucket = bucket;
- }
-
- public String getQuery() {
- return query;
- }
- /**
- * Additional query parameters (in JSON) that are used to configure the query used for finding
- * files in the GridFsConsumer
- * @param query
- */
- public void setQuery(String query) {
- this.query = query;
- }
- public long getDelay() {
- return delay;
- }
- /**
- * Sets the delay between polls within the Consumer. Default is 500ms
- * @param delay
- */
- public void setDelay(long delay) {
- this.delay = delay;
- }
- public long getInitialDelay() {
- return initialDelay;
- }
- /**
- * Sets the initialDelay before the consumer will start polling. Default is 1000ms
- * @param initialDelay
- */
- public void setInitialDelay(long initialDelay) {
- this.initialDelay = delay;
- }
-
- /**
- * Sets the QueryStrategy that is used for polling for new files. Default is Timestamp
- * @see QueryStrategy
- * @param s
- */
- public void setQueryStrategy(String s) {
- queryStrategy = QueryStrategy.valueOf(s);
- }
- public QueryStrategy getQueryStrategy() {
- return queryStrategy;
- }
- /**
- * If the QueryType uses a persistent timestamp, this sets the name of the collection within
- * the DB to store the timestamp.
- * @param s
- */
- public void setPersistentTSCollection(String s) {
- persistentTSCollection = s;
- }
- public String getPersistentTSCollection() {
- return persistentTSCollection;
- }
- /**
- * If the QueryType uses a persistent timestamp, this is the ID of the object in the collection
- * to store the timestamp.
- * @param s
- */
- public void setPersistentTSObject(String id) {
- persistentTSObject = id;
- }
- public String getPersistentTSObject() {
- return persistentTSObject;
- }
-
- /**
- * If the QueryType uses a FileAttribute, this sets the name of the attribute that is used. Default is "camel-processed".
- * @param f
- */
- public void setFileAttributeName(String f) {
- fileAttributeName = f;
- }
- public String getFileAttributeName() {
- return fileAttributeName;
- }
-
- /**
- * Set the {@link WriteConcern} for write operations on MongoDB using the standard ones.
- * Resolved from the fields of the WriteConcern class by calling the {@link WriteConcern#valueOf(String)} method.
- *
- * @param writeConcern the standard name of the WriteConcern
- * @see <a href="http://api.mongodb.org/java/current/com/mongodb/WriteConcern.html#valueOf(java.lang.String)">possible options</a>
- */
- public void setWriteConcern(String writeConcern) {
- this.writeConcern = WriteConcern.valueOf(writeConcern);
- }
-
- public WriteConcern getWriteConcern() {
- return writeConcern;
- }
-
- /**
- * Set the {@link WriteConcern} for write operations on MongoDB, passing in the bean ref to a custom WriteConcern which exists in the Registry.
- * You can also use standard WriteConcerns by passing in their key. See the {@link #setWriteConcern(String) setWriteConcern} method.
- *
- * @param writeConcernRef the name of the bean in the registry that represents the WriteConcern to use
- */
- public void setWriteConcernRef(String writeConcernRef) {
- WriteConcern wc = this.getCamelContext().getRegistry().lookupByNameAndType(writeConcernRef, WriteConcern.class);
- if (wc == null) {
- String msg = "Camel MongoDB component could not find the WriteConcern in the Registry. Verify that the "
- + "provided bean name (" + writeConcernRef + ") is correct. Aborting initialization.";
- throw new IllegalArgumentException(msg);
- }
-
- this.writeConcernRef = wc;
- }
-
- public WriteConcern getWriteConcernRef() {
- return writeConcernRef;
- }
-
- /**
- * Sets a MongoDB {@link ReadPreference} on the Mongo connection. Read preferences set directly on the connection will be
- * overridden by this setting.
- * <p/>
- * The {@link com.mongodb.ReadPreference#valueOf(String)} utility method is used to resolve the passed {@code readPreference}
- * value. Some examples for the possible values are {@code nearest}, {@code primary} or {@code secondary} etc.
- *
- * @param readPreference the name of the read preference to set
- */
- public void setReadPreference(String readPreference) {
- this.readPreference = ReadPreference.valueOf(readPreference);
- }
-
- public ReadPreference getReadPreference() {
- return readPreference;
- }
-
-
- /**
- * Sets the operation this endpoint will execute against GridRS.
- */
- public void setOperation(String operation) {
- this.operation = operation;
- }
-
- public String getOperation() {
- return operation;
- }
-
- public GridFS getGridFs() {
- return gridFs;
- }
-
- public void setGridFs(GridFS gridFs) {
- this.gridFs = gridFs;
- }
- public DBCollection getFilesCollection() {
- return filesCollection;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsProducer.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsProducer.java
deleted file mode 100644
index 397cc03..0000000
--- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/GridFsProducer.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/**
- * 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.gridfs;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.Reader;
-
-import com.mongodb.BasicDBObject;
-import com.mongodb.DBCursor;
-import com.mongodb.DBObject;
-import com.mongodb.gridfs.GridFSDBFile;
-import com.mongodb.gridfs.GridFSInputFile;
-import com.mongodb.util.JSON;
-import org.apache.camel.Exchange;
-import org.apache.camel.impl.DefaultProducer;
-
-
-public class GridFsProducer extends DefaultProducer {
- private GridFsEndpoint endpoint;
-
- public GridFsProducer(GridFsEndpoint endpoint) {
- super(endpoint);
- this.endpoint = endpoint;
- }
-
- public void process(Exchange exchange) throws Exception {
- String operation = endpoint.getOperation();
- if (operation == null) {
- operation = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_OPERATION, String.class);
- }
- if (operation == null || "create".equals(operation)) {
- final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
- Long chunkSize = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_CHUNKSIZE, Long.class);
-
- InputStream ins = exchange.getIn().getMandatoryBody(InputStream.class);
- GridFSInputFile gfsFile = endpoint.getGridFs().createFile(ins, filename, true);
- if (chunkSize != null && chunkSize > 0) {
- gfsFile.setChunkSize(chunkSize);
- }
- final String ct = exchange.getIn().getHeader(Exchange.CONTENT_TYPE, String.class);
- if (ct != null) {
- gfsFile.setContentType(ct);
- }
- String metaData = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_METADATA, String.class);
- DBObject dbObject = (DBObject) JSON.parse(metaData);
- gfsFile.setMetaData(dbObject);
- gfsFile.save();
- //add headers with the id and file name produced by the driver.
- exchange.getIn().setHeader(Exchange.FILE_NAME_PRODUCED, gfsFile.getFilename());
- exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_FILE_ID_PRODUCED, gfsFile.getId());
- } else if ("remove".equals(operation)) {
- final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
- endpoint.getGridFs().remove(filename);
- } else if ("findOne".equals(operation)) {
- final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
- GridFSDBFile file = endpoint.getGridFs().findOne(filename);
- if (file != null) {
- exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_METADATA, JSON.serialize(file.getMetaData()));
- exchange.getIn().setHeader(Exchange.FILE_CONTENT_TYPE, file.getContentType());
- exchange.getIn().setHeader(Exchange.FILE_LENGTH, file.getLength());
- exchange.getIn().setHeader(Exchange.FILE_LAST_MODIFIED, file.getUploadDate());
- exchange.getIn().setBody(file.getInputStream(), InputStream.class);
- } else {
- throw new FileNotFoundException("No GridFS file for " + filename);
- }
- } else if ("listAll".equals(operation)) {
- final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
- DBCursor cursor;
- if (filename == null) {
- cursor = endpoint.getGridFs().getFileList();
- } else {
- cursor = endpoint.getGridFs().getFileList(new BasicDBObject("filename", filename));
- }
- exchange.getIn().setBody(new DBCursorFilenameReader(cursor), Reader.class);
- } else if ("count".equals(operation)) {
- final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
- DBCursor cursor;
- if (filename == null) {
- cursor = endpoint.getGridFs().getFileList();
- } else {
- cursor = endpoint.getGridFs().getFileList(new BasicDBObject("filename", filename));
- }
- exchange.getIn().setBody(cursor.count(), Integer.class);
- }
-
- }
-
-
- private class DBCursorFilenameReader extends Reader {
- DBCursor cursor;
- StringBuilder current;
- int pos;
-
- DBCursorFilenameReader(DBCursor c) {
- cursor = c;
- current = new StringBuilder(4096);
- pos = 0;
- fill();
- }
- void fill() {
- if (pos > 0) {
- current.delete(0, pos);
- pos = 0;
- }
- while (cursor.hasNext() && current.length() < 4000) {
- DBObject o = cursor.next();
- current.append(o.get("filename")).append("\t").append(o.get("_id")).append("\n");
- }
- }
- @Override
- public int read(char[] cbuf, int off, int len) throws IOException {
- if (pos == current.length()) {
- fill();
- }
- if (pos == current.length()) {
- return -1;
- }
- if (len > (current.length() - pos)) {
- len = current.length() - pos;
- }
- current.getChars(pos, pos + len, cbuf, off);
- pos += len;
- return len;
- }
-
- @Override
- public void close() throws IOException {
- cursor.close();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/QueryStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/QueryStrategy.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/QueryStrategy.java
deleted file mode 100644
index 9b29f2a..0000000
--- a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/gridfs/QueryStrategy.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/**
- * 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.gridfs;
-
-public enum QueryStrategy {
- TimeStamp,
- PersistentTimestamp,
- FileAttribute,
- TimeStampAndFileAttribute,
- PersistentTimestampAndFileAttribute
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsComponent.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsComponent.java
new file mode 100644
index 0000000..99c677f
--- /dev/null
+++ b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsComponent.java
@@ -0,0 +1,45 @@
+/**
+ * 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.mongodb.gridfs;
+
+import java.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+
+
+public class GridFsComponent extends UriEndpointComponent {
+
+ public GridFsComponent() {
+ super(GridFsEndpoint.class);
+ }
+
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+
+ GridFsEndpoint endpoint = new GridFsEndpoint(uri, this);
+ endpoint.setConnectionBean(remaining);
+ setProperties(endpoint, parameters);
+
+ return endpoint;
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ super.doShutdown();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
new file mode 100644
index 0000000..2153997
--- /dev/null
+++ b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsConsumer.java
@@ -0,0 +1,177 @@
+/**
+ * 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.mongodb.gridfs;
+
+import java.io.InputStream;
+import java.util.concurrent.ExecutorService;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.BasicDBObjectBuilder;
+import com.mongodb.DBCollection;
+import com.mongodb.DBCursor;
+import com.mongodb.DBObject;
+import com.mongodb.MongoException;
+import com.mongodb.gridfs.GridFSDBFile;
+import com.mongodb.util.JSON;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultConsumer;
+
+/**
+ *
+ */
+public class GridFsConsumer extends DefaultConsumer implements Runnable {
+ final GridFsEndpoint endpoint;
+ private ExecutorService executor;
+
+ /**
+ * @param endpoint
+ * @param processor
+ */
+ public GridFsConsumer(GridFsEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (executor != null) {
+ executor.shutdown();
+ executor = null;
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ executor = endpoint.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, endpoint.getEndpointUri(), 1);
+ executor.execute(this);
+ }
+
+ @Override
+ public void run() {
+ DBCursor c = null;
+ java.util.Date fromDate = null;
+
+ QueryStrategy s = endpoint.getQueryStrategy();
+ boolean usesTimestamp = s != QueryStrategy.FileAttribute;
+ boolean persistsTimestamp = s == QueryStrategy.PersistentTimestamp || s == QueryStrategy.PersistentTimestampAndFileAttribute;
+ boolean usesAttribute = s == QueryStrategy.FileAttribute
+ || s == QueryStrategy.TimeStampAndFileAttribute
+ || s == QueryStrategy.PersistentTimestampAndFileAttribute;
+
+ DBCollection ptsCollection = null;
+ DBObject persistentTimestamp = null;
+ if (persistsTimestamp) {
+ ptsCollection = endpoint.getDB().getCollection(endpoint.getPersistentTSCollection());
+ // ensure standard indexes as long as collections are small
+ try {
+ if (ptsCollection.count() < 1000) {
+ ptsCollection.createIndex(new BasicDBObject("id", 1));
+ }
+ } catch (MongoException e) {
+ //TODO: Logging
+ }
+ persistentTimestamp = ptsCollection.findOne(new BasicDBObject("id", endpoint.getPersistentTSObject()));
+ if (persistentTimestamp == null) {
+ persistentTimestamp = new BasicDBObject("id", endpoint.getPersistentTSObject());
+ fromDate = new java.util.Date();
+ persistentTimestamp.put("timestamp", fromDate);
+ ptsCollection.save(persistentTimestamp);
+ }
+ fromDate = (java.util.Date)persistentTimestamp.get("timestamp");
+ } else if (usesTimestamp) {
+ fromDate = new java.util.Date();
+ }
+ try {
+ Thread.sleep(endpoint.getInitialDelay());
+ while (isStarted()) {
+ if (c == null || c.getCursorId() == 0) {
+ if (c != null) {
+ c.close();
+ }
+ String queryString = endpoint.getQuery();
+ DBObject query;
+ if (queryString == null) {
+ query = new BasicDBObject();
+ } else {
+ query = (DBObject) JSON.parse(queryString);
+ }
+ if (usesTimestamp) {
+ query.put("uploadDate", new BasicDBObject("$gt", fromDate));
+ }
+ if (usesAttribute) {
+ query.put(endpoint.getFileAttributeName(), null);
+ }
+ c = endpoint.getFilesCollection().find(query);
+ }
+ boolean dateModified = false;
+ while (c.hasNext() && isStarted()) {
+ GridFSDBFile file = (GridFSDBFile)c.next();
+ GridFSDBFile forig = file;
+ if (usesAttribute) {
+ file.put(endpoint.getFileAttributeName(), "processing");
+ DBObject q = BasicDBObjectBuilder.start("_id", file.getId()).append("camel-processed", null).get();
+ forig = (GridFSDBFile)endpoint.getFilesCollection().findAndModify(q, null, null, false, file, true, false);
+ }
+ if (forig != null) {
+ file = endpoint.getGridFs().findOne(new BasicDBObject("_id", file.getId()));
+
+ Exchange exchange = endpoint.createExchange();
+ exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_METADATA, JSON.serialize(file.getMetaData()));
+ exchange.getIn().setHeader(Exchange.FILE_CONTENT_TYPE, file.getContentType());
+ exchange.getIn().setHeader(Exchange.FILE_LENGTH, file.getLength());
+ exchange.getIn().setHeader(Exchange.FILE_LAST_MODIFIED, file.getUploadDate());
+ exchange.getIn().setBody(file.getInputStream(), InputStream.class);
+ try {
+ getProcessor().process(exchange);
+ //System.out.println("Processing " + file.getFilename());
+ if (usesAttribute) {
+ forig.put(endpoint.getFileAttributeName(), "done");
+ endpoint.getFilesCollection().save(forig);
+ }
+ if (usesTimestamp) {
+ if (file.getUploadDate().compareTo(fromDate) > 0) {
+ fromDate = file.getUploadDate();
+ dateModified = true;
+ }
+ }
+ } catch (Exception e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+ }
+ if (persistsTimestamp && dateModified) {
+ persistentTimestamp.put("timestamp", fromDate);
+ ptsCollection.save(persistentTimestamp);
+ }
+ Thread.sleep(endpoint.getDelay());
+ }
+ } catch (Throwable e1) {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ if (c != null) {
+ c.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
new file mode 100644
index 0000000..7896d40
--- /dev/null
+++ b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsEndpoint.java
@@ -0,0 +1,369 @@
+/**
+ * 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.mongodb.gridfs;
+
+import com.mongodb.DB;
+import com.mongodb.DBCollection;
+import com.mongodb.Mongo;
+import com.mongodb.MongoClient;
+import com.mongodb.ReadPreference;
+import com.mongodb.WriteConcern;
+import com.mongodb.gridfs.GridFS;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+import org.apache.camel.util.CamelContextHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Component for working with MongoDB GridFS.
+ */
+@UriEndpoint(firstVersion = "2.18.0", scheme = "mongodb-gridfs", title = "MongoDB GridFS", syntax = "mongodb-gridfs:connectionBean", label = "database,nosql")
+public class GridFsEndpoint extends DefaultEndpoint {
+
+ public static final String GRIDFS_OPERATION = "gridfs.operation";
+ public static final String GRIDFS_METADATA = "gridfs.metadata";
+ public static final String GRIDFS_CHUNKSIZE = "gridfs.chunksize";
+ public static final String GRIDFS_FILE_ID_PRODUCED = "gridfs.fileid";
+
+ private static final Logger LOG = LoggerFactory.getLogger(GridFsEndpoint.class);
+
+ @UriPath @Metadata(required = "true")
+ private String connectionBean;
+ @UriParam @Metadata(required = "true")
+ private String database;
+ @UriParam(defaultValue = GridFS.DEFAULT_BUCKET)
+ private String bucket;
+ @UriParam(enums = "ACKNOWLEDGED,W1,W2,W3,UNACKNOWLEDGED,JOURNALED,MAJORITY,SAFE")
+ private WriteConcern writeConcern;
+ @UriParam
+ private WriteConcern writeConcernRef;
+ @UriParam
+ private ReadPreference readPreference;
+
+ @UriParam(label = "producer")
+ private String operation;
+
+ @UriParam(label = "consumer")
+ private String query;
+ @UriParam(label = "consumer", defaultValue = "1000")
+ private long initialDelay = 1000;
+ @UriParam(label = "consumer", defaultValue = "500")
+ private long delay = 500;
+
+ @UriParam(label = "consumer", defaultValue = "TimeStamp")
+ private QueryStrategy queryStrategy = QueryStrategy.TimeStamp;
+ @UriParam(label = "consumer", defaultValue = "camel-timestamps")
+ private String persistentTSCollection = "camel-timestamps";
+ @UriParam(label = "consumer", defaultValue = "camel-timestamp")
+ private String persistentTSObject = "camel-timestamp";
+ @UriParam(label = "consumer", defaultValue = "camel-processed")
+ private String fileAttributeName = "camel-processed";
+
+
+ private Mongo mongoConnection;
+ private DB db;
+ private GridFS gridFs;
+ private DBCollection filesCollection;
+
+ public GridFsEndpoint(String uri, GridFsComponent component) {
+ super(uri, component);
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ initializeConnection();
+ return new GridFsProducer(this);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ initializeConnection();
+ return new GridFsConsumer(this, processor);
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ @SuppressWarnings("deprecation")
+ public void initializeConnection() throws Exception {
+ LOG.info("Initialize GridFS endpoint: {}", this.toString());
+ if (database == null) {
+ throw new IllegalStateException("Missing required endpoint configuration: database");
+ }
+ db = mongoConnection.getDB(database);
+ if (db == null) {
+ throw new IllegalStateException("Could not initialize GridFsComponent. Database " + database + " does not exist.");
+ }
+ gridFs = new GridFS(db, bucket == null ? GridFS.DEFAULT_BUCKET : bucket) {
+ {
+ filesCollection = getFilesCollection();
+ }
+ };
+ }
+
+
+ @Override
+ protected void doStart() throws Exception {
+ if (writeConcern != null && writeConcernRef != null) {
+ String msg = "Cannot set both writeConcern and writeConcernRef at the same time. Respective values: " + writeConcern
+ + ", " + writeConcernRef + ". Aborting initialization.";
+ throw new IllegalArgumentException(msg);
+ }
+ mongoConnection = CamelContextHelper.mandatoryLookup(getCamelContext(), connectionBean, MongoClient.class);
+ LOG.debug("Resolved the connection with the name {} as {}", connectionBean, mongoConnection);
+ setWriteReadOptionsOnConnection();
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (mongoConnection != null) {
+ LOG.debug("Closing connection");
+ mongoConnection.close();
+ }
+ }
+
+ private void setWriteReadOptionsOnConnection() {
+ // Set the WriteConcern
+ if (writeConcern != null) {
+ mongoConnection.setWriteConcern(writeConcern);
+ } else if (writeConcernRef != null) {
+ mongoConnection.setWriteConcern(writeConcernRef);
+ }
+
+ // Set the ReadPreference
+ if (readPreference != null) {
+ mongoConnection.setReadPreference(readPreference);
+ }
+ }
+
+
+
+
+ // ======= Getters and setters ===============================================
+ public String getConnectionBean() {
+ return connectionBean;
+ }
+ /**
+ * Name of {@link com.mongodb.Mongo} to use.
+ */
+ public void setConnectionBean(String connectionBean) {
+ this.connectionBean = connectionBean;
+ }
+
+ public Mongo getMongoConnection() {
+ return mongoConnection;
+ }
+ /**
+ * Sets the Mongo instance that represents the backing connection
+ *
+ * @param mongoConnection the connection to the database
+ */
+ public void setMongoConnection(Mongo mongoConnection) {
+ this.mongoConnection = mongoConnection;
+ }
+
+ public DB getDB() {
+ return db;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+ /**
+ * Sets the name of the MongoDB database to target
+ *
+ * @param database name of the MongoDB database
+ */
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+ /**
+ * Sets the name of the GridFS bucket within the database. Default is "fs".
+ *
+ * @param database name of the MongoDB database
+ */
+ public String getBucket() {
+ return bucket;
+ }
+ public void setBucket(String bucket) {
+ this.bucket = bucket;
+ }
+
+ public String getQuery() {
+ return query;
+ }
+ /**
+ * Additional query parameters (in JSON) that are used to configure the query used for finding
+ * files in the GridFsConsumer
+ * @param query
+ */
+ public void setQuery(String query) {
+ this.query = query;
+ }
+ public long getDelay() {
+ return delay;
+ }
+ /**
+ * Sets the delay between polls within the Consumer. Default is 500ms
+ * @param delay
+ */
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+ public long getInitialDelay() {
+ return initialDelay;
+ }
+ /**
+ * Sets the initialDelay before the consumer will start polling. Default is 1000ms
+ * @param initialDelay
+ */
+ public void setInitialDelay(long initialDelay) {
+ this.initialDelay = delay;
+ }
+
+ /**
+ * Sets the QueryStrategy that is used for polling for new files. Default is Timestamp
+ * @see QueryStrategy
+ * @param s
+ */
+ public void setQueryStrategy(String s) {
+ queryStrategy = QueryStrategy.valueOf(s);
+ }
+ public QueryStrategy getQueryStrategy() {
+ return queryStrategy;
+ }
+ /**
+ * If the QueryType uses a persistent timestamp, this sets the name of the collection within
+ * the DB to store the timestamp.
+ * @param s
+ */
+ public void setPersistentTSCollection(String s) {
+ persistentTSCollection = s;
+ }
+ public String getPersistentTSCollection() {
+ return persistentTSCollection;
+ }
+ /**
+ * If the QueryType uses a persistent timestamp, this is the ID of the object in the collection
+ * to store the timestamp.
+ * @param s
+ */
+ public void setPersistentTSObject(String id) {
+ persistentTSObject = id;
+ }
+ public String getPersistentTSObject() {
+ return persistentTSObject;
+ }
+
+ /**
+ * If the QueryType uses a FileAttribute, this sets the name of the attribute that is used. Default is "camel-processed".
+ * @param f
+ */
+ public void setFileAttributeName(String f) {
+ fileAttributeName = f;
+ }
+ public String getFileAttributeName() {
+ return fileAttributeName;
+ }
+
+ /**
+ * Set the {@link WriteConcern} for write operations on MongoDB using the standard ones.
+ * Resolved from the fields of the WriteConcern class by calling the {@link WriteConcern#valueOf(String)} method.
+ *
+ * @param writeConcern the standard name of the WriteConcern
+ * @see <a href="http://api.mongodb.org/java/current/com/mongodb/WriteConcern.html#valueOf(java.lang.String)">possible options</a>
+ */
+ public void setWriteConcern(String writeConcern) {
+ this.writeConcern = WriteConcern.valueOf(writeConcern);
+ }
+
+ public WriteConcern getWriteConcern() {
+ return writeConcern;
+ }
+
+ /**
+ * Set the {@link WriteConcern} for write operations on MongoDB, passing in the bean ref to a custom WriteConcern which exists in the Registry.
+ * You can also use standard WriteConcerns by passing in their key. See the {@link #setWriteConcern(String) setWriteConcern} method.
+ *
+ * @param writeConcernRef the name of the bean in the registry that represents the WriteConcern to use
+ */
+ public void setWriteConcernRef(String writeConcernRef) {
+ WriteConcern wc = this.getCamelContext().getRegistry().lookupByNameAndType(writeConcernRef, WriteConcern.class);
+ if (wc == null) {
+ String msg = "Camel MongoDB component could not find the WriteConcern in the Registry. Verify that the "
+ + "provided bean name (" + writeConcernRef + ") is correct. Aborting initialization.";
+ throw new IllegalArgumentException(msg);
+ }
+
+ this.writeConcernRef = wc;
+ }
+
+ public WriteConcern getWriteConcernRef() {
+ return writeConcernRef;
+ }
+
+ /**
+ * Sets a MongoDB {@link ReadPreference} on the Mongo connection. Read preferences set directly on the connection will be
+ * overridden by this setting.
+ * <p/>
+ * The {@link com.mongodb.ReadPreference#valueOf(String)} utility method is used to resolve the passed {@code readPreference}
+ * value. Some examples for the possible values are {@code nearest}, {@code primary} or {@code secondary} etc.
+ *
+ * @param readPreference the name of the read preference to set
+ */
+ public void setReadPreference(String readPreference) {
+ this.readPreference = ReadPreference.valueOf(readPreference);
+ }
+
+ public ReadPreference getReadPreference() {
+ return readPreference;
+ }
+
+
+ /**
+ * Sets the operation this endpoint will execute against GridRS.
+ */
+ public void setOperation(String operation) {
+ this.operation = operation;
+ }
+
+ public String getOperation() {
+ return operation;
+ }
+
+ public GridFS getGridFs() {
+ return gridFs;
+ }
+
+ public void setGridFs(GridFS gridFs) {
+ this.gridFs = gridFs;
+ }
+ public DBCollection getFilesCollection() {
+ return filesCollection;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e067ae91/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
new file mode 100644
index 0000000..8459356
--- /dev/null
+++ b/components/camel-mongodb-gridfs/src/main/java/org/apache/camel/component/mongodb/gridfs/GridFsProducer.java
@@ -0,0 +1,147 @@
+/**
+ * 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.mongodb.gridfs;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBCursor;
+import com.mongodb.DBObject;
+import com.mongodb.gridfs.GridFSDBFile;
+import com.mongodb.gridfs.GridFSInputFile;
+import com.mongodb.util.JSON;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultProducer;
+
+
+public class GridFsProducer extends DefaultProducer {
+ private GridFsEndpoint endpoint;
+
+ public GridFsProducer(GridFsEndpoint endpoint) {
+ super(endpoint);
+ this.endpoint = endpoint;
+ }
+
+ public void process(Exchange exchange) throws Exception {
+ String operation = endpoint.getOperation();
+ if (operation == null) {
+ operation = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_OPERATION, String.class);
+ }
+ if (operation == null || "create".equals(operation)) {
+ final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
+ Long chunkSize = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_CHUNKSIZE, Long.class);
+
+ InputStream ins = exchange.getIn().getMandatoryBody(InputStream.class);
+ GridFSInputFile gfsFile = endpoint.getGridFs().createFile(ins, filename, true);
+ if (chunkSize != null && chunkSize > 0) {
+ gfsFile.setChunkSize(chunkSize);
+ }
+ final String ct = exchange.getIn().getHeader(Exchange.CONTENT_TYPE, String.class);
+ if (ct != null) {
+ gfsFile.setContentType(ct);
+ }
+ String metaData = exchange.getIn().getHeader(GridFsEndpoint.GRIDFS_METADATA, String.class);
+ DBObject dbObject = (DBObject) JSON.parse(metaData);
+ gfsFile.setMetaData(dbObject);
+ gfsFile.save();
+ //add headers with the id and file name produced by the driver.
+ exchange.getIn().setHeader(Exchange.FILE_NAME_PRODUCED, gfsFile.getFilename());
+ exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_FILE_ID_PRODUCED, gfsFile.getId());
+ } else if ("remove".equals(operation)) {
+ final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
+ endpoint.getGridFs().remove(filename);
+ } else if ("findOne".equals(operation)) {
+ final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
+ GridFSDBFile file = endpoint.getGridFs().findOne(filename);
+ if (file != null) {
+ exchange.getIn().setHeader(GridFsEndpoint.GRIDFS_METADATA, JSON.serialize(file.getMetaData()));
+ exchange.getIn().setHeader(Exchange.FILE_CONTENT_TYPE, file.getContentType());
+ exchange.getIn().setHeader(Exchange.FILE_LENGTH, file.getLength());
+ exchange.getIn().setHeader(Exchange.FILE_LAST_MODIFIED, file.getUploadDate());
+ exchange.getIn().setBody(file.getInputStream(), InputStream.class);
+ } else {
+ throw new FileNotFoundException("No GridFS file for " + filename);
+ }
+ } else if ("listAll".equals(operation)) {
+ final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
+ DBCursor cursor;
+ if (filename == null) {
+ cursor = endpoint.getGridFs().getFileList();
+ } else {
+ cursor = endpoint.getGridFs().getFileList(new BasicDBObject("filename", filename));
+ }
+ exchange.getIn().setBody(new DBCursorFilenameReader(cursor), Reader.class);
+ } else if ("count".equals(operation)) {
+ final String filename = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
+ DBCursor cursor;
+ if (filename == null) {
+ cursor = endpoint.getGridFs().getFileList();
+ } else {
+ cursor = endpoint.getGridFs().getFileList(new BasicDBObject("filename", filename));
+ }
+ exchange.getIn().setBody(cursor.count(), Integer.class);
+ }
+
+ }
+
+
+ private class DBCursorFilenameReader extends Reader {
+ DBCursor cursor;
+ StringBuilder current;
+ int pos;
+
+ DBCursorFilenameReader(DBCursor c) {
+ cursor = c;
+ current = new StringBuilder(4096);
+ pos = 0;
+ fill();
+ }
+ void fill() {
+ if (pos > 0) {
+ current.delete(0, pos);
+ pos = 0;
+ }
+ while (cursor.hasNext() && current.length() < 4000) {
+ DBObject o = cursor.next();
+ current.append(o.get("filename")).append("\t").append(o.get("_id")).append("\n");
+ }
+ }
+ @Override
+ public int read(char[] cbuf, int off, int len) throws IOException {
+ if (pos == current.length()) {
+ fill();
+ }
+ if (pos == current.length()) {
+ return -1;
+ }
+ if (len > (current.length() - pos)) {
+ len = current.length() - pos;
+ }
+ current.getChars(pos, pos + len, cbuf, off);
+ pos += len;
+ return len;
+ }
+
+ @Override
+ public void close() throws IOException {
+ cursor.close();
+ }
+ }
+}