davsclaus commented on a change in pull request #3944:
URL: https://github.com/apache/camel/pull/3944#discussion_r445366781
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
##########
@@ -70,6 +71,12 @@
@UriParam
private String bucket;
+ //
+ @UriParam(defaultValue = "_default")
+ private String collection = DEFAULT_COLLECTION;
Review comment:
See my prev comments about default
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
##########
@@ -16,65 +16,76 @@
*/
package org.apache.camel.component.couchbase;
-import com.couchbase.client.CouchbaseClient;
-import com.couchbase.client.protocol.views.Query;
-import com.couchbase.client.protocol.views.View;
-import com.couchbase.client.protocol.views.ViewResponse;
-import com.couchbase.client.protocol.views.ViewRow;
+import java.util.List;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+import com.couchbase.client.java.view.ViewOptions;
+import com.couchbase.client.java.view.ViewOrdering;
+import com.couchbase.client.java.view.ViewResult;
+import com.couchbase.client.java.view.ViewRow;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.support.DefaultScheduledPollConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_DESIGN_DOCUMENT_NAME;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_ID;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_KEY;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_VIEWNAME;
+import static org.apache.camel.component.couchbase.CouchbaseConstants.*;
public class CouchbaseConsumer extends DefaultScheduledPollConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(CouchbaseConsumer.class);
private final CouchbaseEndpoint endpoint;
- private final CouchbaseClient client;
- private final View view;
- private final Query query;
-
- public CouchbaseConsumer(CouchbaseEndpoint endpoint, CouchbaseClient
client, Processor processor) {
+ private final Bucket bucket;
+ private ViewOptions viewOptions;
+ private Collection collection;
+ public CouchbaseConsumer(CouchbaseEndpoint endpoint, Bucket client,
Processor processor) {
super(endpoint, processor);
- this.client = client;
+ this.bucket = client;
this.endpoint = endpoint;
- this.view = client.getView(endpoint.getDesignDocumentName(),
endpoint.getViewName());
- this.query = new Query();
+ Scope scope;
+ if (!endpoint.getScope().equals(DEFAULT_SCOPE)) {
Review comment:
Instead of a default value that is _default. Then why not leave it as
null/empty and then use that to know if a custom scope was set or not
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseProducer.java
##########
@@ -149,21 +151,20 @@ private Boolean setDocument(String id, int expiry, Object
obj, PersistTo persist
private Boolean setDocument(String id, int expiry, Object obj, int
retryAttempts, PersistTo persistTo, ReplicateTo replicateTo) throws Exception {
- OperationFuture<Boolean> result = client.set(id, expiry, obj,
persistTo, replicateTo);
+ UpsertOptions options = UpsertOptions.upsertOptions()
+ .expiry(Duration.ofSeconds(expiry))
+ .durability(persistTo, replicateTo)
+ .timeout(Duration.ofMillis(retryAttempts * producerRetryPause))
+
.retryStrategy(BestEffortRetryStrategy.withExponentialBackoff(Duration.ofMillis(producerRetryPause),
Duration.ofMillis(producerRetryPause), 1));
+
try {
- if (!result.get()) {
- throw new Exception("Unable to save Document. " + id);
- }
- return true;
+ MutationResult result = collection.upsert(id, obj, options);
+ LOG.debug(result.toString());
Review comment:
if isDebugEnabled
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
##########
@@ -70,6 +71,12 @@
@UriParam
private String bucket;
+ //
Review comment:
Remove this
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
##########
@@ -16,65 +16,76 @@
*/
package org.apache.camel.component.couchbase;
-import com.couchbase.client.CouchbaseClient;
-import com.couchbase.client.protocol.views.Query;
-import com.couchbase.client.protocol.views.View;
-import com.couchbase.client.protocol.views.ViewResponse;
-import com.couchbase.client.protocol.views.ViewRow;
+import java.util.List;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+import com.couchbase.client.java.view.ViewOptions;
+import com.couchbase.client.java.view.ViewOrdering;
+import com.couchbase.client.java.view.ViewResult;
+import com.couchbase.client.java.view.ViewRow;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.support.DefaultScheduledPollConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_DESIGN_DOCUMENT_NAME;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_ID;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_KEY;
-import static
org.apache.camel.component.couchbase.CouchbaseConstants.HEADER_VIEWNAME;
+import static org.apache.camel.component.couchbase.CouchbaseConstants.*;
public class CouchbaseConsumer extends DefaultScheduledPollConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(CouchbaseConsumer.class);
private final CouchbaseEndpoint endpoint;
- private final CouchbaseClient client;
- private final View view;
- private final Query query;
-
- public CouchbaseConsumer(CouchbaseEndpoint endpoint, CouchbaseClient
client, Processor processor) {
+ private final Bucket bucket;
+ private ViewOptions viewOptions;
+ private Collection collection;
+ public CouchbaseConsumer(CouchbaseEndpoint endpoint, Bucket client,
Processor processor) {
super(endpoint, processor);
- this.client = client;
+ this.bucket = client;
this.endpoint = endpoint;
- this.view = client.getView(endpoint.getDesignDocumentName(),
endpoint.getViewName());
- this.query = new Query();
+ Scope scope;
+ if (!endpoint.getScope().equals(DEFAULT_SCOPE)) {
+ scope = client.scope(endpoint.getScope());
+ } else {
+ scope = client.defaultScope();
+ }
+
+ if (!endpoint.getCollection().equals(DEFAULT_COLLECTION)) {
Review comment:
The same as for default scope
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
##########
@@ -225,6 +219,28 @@ public void setPort(int port) {
this.port = port;
}
+ /**
+ * The collection to use
+ */
+ public String getCollection() {
+ return this.collection;
+ }
+
+ public void setCollection(String collection) {
+ this.collection = collection;
+ }
+
+ public String getScope() {
+ return this.scope;
+ }
+
+ /**
+ * The scope to use
Review comment:
Do we know from couchbase what scopes is possible? If its like an enum
then we can specify that via @UriParam so we have that information
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
##########
@@ -434,98 +450,21 @@ public void setConsumerProcessedStrategy(String
consumerProcessedStrategy) {
this.consumerProcessedStrategy = consumerProcessedStrategy;
}
- public long getOpTimeOut() {
- return opTimeOut;
+ public long getQueryTimeout() {
+ return queryTimeout;
}
/**
* Define the operation timeout
Review comment:
in milli seconds
##########
File path:
components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseProducer.java
##########
@@ -149,21 +151,20 @@ private Boolean setDocument(String id, int expiry, Object
obj, PersistTo persist
private Boolean setDocument(String id, int expiry, Object obj, int
retryAttempts, PersistTo persistTo, ReplicateTo replicateTo) throws Exception {
- OperationFuture<Boolean> result = client.set(id, expiry, obj,
persistTo, replicateTo);
+ UpsertOptions options = UpsertOptions.upsertOptions()
+ .expiry(Duration.ofSeconds(expiry))
+ .durability(persistTo, replicateTo)
+ .timeout(Duration.ofMillis(retryAttempts * producerRetryPause))
+
.retryStrategy(BestEffortRetryStrategy.withExponentialBackoff(Duration.ofMillis(producerRetryPause),
Duration.ofMillis(producerRetryPause), 1));
+
try {
- if (!result.get()) {
- throw new Exception("Unable to save Document. " + id);
- }
- return true;
+ MutationResult result = collection.upsert(id, obj, options);
+ LOG.debug(result.toString());
} catch (Exception e) {
- if (retryAttempts <= 0) {
- throw e;
- } else {
- LOG.info("Unable to save Document, retrying in " +
producerRetryPause + "ms (" + retryAttempts + ")");
- Thread.sleep(producerRetryPause);
- return setDocument(id, expiry, obj, retryAttempts - 1,
persistTo, replicateTo);
- }
+ LOG.error("Unable to save Document");
Review comment:
Dont log and throw, just throw
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
