camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [2/2] camel git commit: CAMEL-10271: Fixed camel-jt400 consumer to be scheduled so it reads from the queue as the jt400 library is not event based.
Date Mon, 05 Sep 2016 16:24:02 GMT
CAMEL-10271: Fixed camel-jt400 consumer to be scheduled so it reads from the queue as the jt400
library is not event based.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/77e417c2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/77e417c2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/77e417c2

Branch: refs/heads/camel-2.17.x
Commit: 77e417c2775b15f24cc6f47df7fbdfe7c4a436af
Parents: 1f7abfc
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Sep 5 18:22:28 2016 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Sep 5 18:23:18 2016 +0200

----------------------------------------------------------------------
 .../component/jt400/Jt400Configuration.java     | 14 +++++++
 .../component/jt400/Jt400DataQueueConsumer.java | 42 ++++++++++++++------
 .../camel/component/jt400/Jt400Endpoint.java    | 22 +++++-----
 3 files changed, 54 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/77e417c2/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Configuration.java
----------------------------------------------------------------------
diff --git a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Configuration.java
b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Configuration.java
index b33a25d..9a2383f 100644
--- a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Configuration.java
+++ b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Configuration.java
@@ -113,6 +113,9 @@ public class Jt400Configuration {
     @UriParam
     private Integer[] outputFieldsLengthArray;
 
+    @UriParam(label = "consumer", defaultValue = "30000")
+    private int readTimeout = 30000;
+
     public Jt400Configuration(String endpointUri, AS400ConnectionPool connectionPool) throws
URISyntaxException {
         ObjectHelper.notNull(endpointUri, "endpointUri", this);
         ObjectHelper.notNull(connectionPool, "connectionPool", this);
@@ -301,6 +304,17 @@ public class Jt400Configuration {
         this.outputFieldsLengthArray = outputFieldsLengthArray;
     }
 
+    public int getReadTimeout() {
+        return readTimeout;
+    }
+
+    /**
+     * Timeout in millis the consumer will wait while trying to read a new message of the
data queue.
+     */
+    public void setReadTimeout(int readTimeout) {
+        this.readTimeout = readTimeout;
+    }
+
     public void setOutputFieldsIdx(String outputFieldsIdx) {
         if (outputFieldsIdx != null) {
             String[] outputArray = outputFieldsIdx.split(",");

http://git-wip-us.apache.org/repos/asf/camel/blob/77e417c2/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
index c3ff46e..6bf9788 100644
--- a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
+++ b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
@@ -22,16 +22,15 @@ import com.ibm.as400.access.DataQueueEntry;
 import com.ibm.as400.access.KeyedDataQueue;
 import com.ibm.as400.access.KeyedDataQueueEntry;
 import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
 import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.impl.PollingConsumerSupport;
+import org.apache.camel.impl.ScheduledPollConsumer;
 
 /**
- * {@link org.apache.camel.PollingConsumer} that polls a data queue for data
+ * A scheduled {@link org.apache.camel.Consumer} that polls a data queue for data
  */
-public class Jt400DataQueueConsumer extends PollingConsumerSupport {
+public class Jt400DataQueueConsumer extends ScheduledPollConsumer {
 
-    private final Jt400Endpoint endpoint;
-    
     /**
      * Performs the lifecycle logic of this consumer.
      */
@@ -40,13 +39,28 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
     /**
      * Creates a new consumer instance
      */
-    protected Jt400DataQueueConsumer(Jt400Endpoint endpoint) {
-        super(endpoint);
-        this.endpoint = endpoint;
+    public Jt400DataQueueConsumer(Jt400Endpoint endpoint, Processor processor) {
+        super(endpoint, processor);
         this.queueService = new Jt400DataQueueService(endpoint);
     }
 
     @Override
+    public Jt400Endpoint getEndpoint() {
+        return (Jt400Endpoint) super.getEndpoint();
+    }
+
+    @Override
+    protected int poll() throws Exception {
+        Exchange exchange = receive(getEndpoint().getReadTimeout());
+        if (exchange != null) {
+            getProcessor().process(exchange);
+            return 1;
+        } else {
+            return 0;
+        }
+    }
+
+    @Override
     protected void doStart() throws Exception {
         queueService.start();
     }
@@ -56,11 +70,13 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
         queueService.stop();
     }
 
+    @Deprecated
     public Exchange receive() {
         // -1 to indicate a blocking read from data queue
         return receive(-1);
     }
 
+    @Deprecated
     public Exchange receiveNoWait() {
         return receive(0);
     }
@@ -85,7 +101,7 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
     public Exchange receive(long timeout) {
         BaseDataQueue queue = queueService.getDataQueue();
         try {
-            if (endpoint.isKeyed()) {
+            if (getEndpoint().isKeyed()) {
                 return receive((KeyedDataQueue) queue, timeout);
             } else {
                 return receive((DataQueue) queue, timeout);
@@ -109,7 +125,7 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
         Exchange exchange = getEndpoint().createExchange();
         if (entry != null) {
             exchange.getIn().setHeader(Jt400Endpoint.SENDER_INFORMATION, entry.getSenderInformation());
-            if (endpoint.getFormat() == Jt400Configuration.Format.binary) {
+            if (getEndpoint().getFormat() == Jt400Configuration.Format.binary) {
                 exchange.getIn().setBody(entry.getData());
             } else {
                 exchange.getIn().setBody(entry.getString());
@@ -120,8 +136,8 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
     }
 
     private Exchange receive(KeyedDataQueue queue, long timeout) throws Exception {
-        String key = endpoint.getSearchKey();
-        String searchType = endpoint.getSearchType().name();
+        String key = getEndpoint().getSearchKey();
+        String searchType = getEndpoint().getSearchType().name();
         KeyedDataQueueEntry entry;
         if (timeout >= 0) {
             int seconds = (int) timeout / 1000;
@@ -135,7 +151,7 @@ public class Jt400DataQueueConsumer extends PollingConsumerSupport {
         Exchange exchange = getEndpoint().createExchange();
         if (entry != null) {
             exchange.getIn().setHeader(Jt400Endpoint.SENDER_INFORMATION, entry.getSenderInformation());
-            if (endpoint.getFormat() == Jt400Configuration.Format.binary) {
+            if (getEndpoint().getFormat() == Jt400Configuration.Format.binary) {
                 exchange.getIn().setBody(entry.getData());
                 exchange.getIn().setHeader(Jt400Endpoint.KEY, entry.getKey());
             } else {

http://git-wip-us.apache.org/repos/asf/camel/blob/77e417c2/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Endpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Endpoint.java
b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Endpoint.java
index 849e683..e3ad785 100644
--- a/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Endpoint.java
+++ b/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400Endpoint.java
@@ -24,10 +24,9 @@ import com.ibm.as400.access.AS400;
 import com.ibm.as400.access.AS400ConnectionPool;
 import org.apache.camel.CamelException;
 import org.apache.camel.Consumer;
-import org.apache.camel.PollingConsumer;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
-import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.impl.ScheduledPollEndpoint;
 import org.apache.camel.spi.UriEndpoint;
 import org.apache.camel.spi.UriParam;
 import org.apache.camel.util.ObjectHelper;
@@ -37,7 +36,7 @@ import org.apache.camel.util.URISupport;
  * The jt400 component allows you to exchanges messages with an AS/400 system using data
queues or program call.
  */
 @UriEndpoint(scheme = "jt400", title = "JT400", syntax = "jt400:userID:password/systemName/objectPath.type",
consumerClass = Jt400DataQueueConsumer.class, label = "messaging")
-public class Jt400Endpoint extends DefaultEndpoint {
+public class Jt400Endpoint extends ScheduledPollEndpoint {
 
     public static final String KEY = "KEY";
     public static final String SENDER_INFORMATION = "SENDER_INFORMATION";
@@ -70,13 +69,6 @@ public class Jt400Endpoint extends DefaultEndpoint {
     }
 
     @Override
-    public PollingConsumer createPollingConsumer() throws Exception {
-        Jt400DataQueueConsumer answer = new Jt400DataQueueConsumer(this);
-        configurePollingConsumer(answer);
-        return answer;
-    }
-
-    @Override
     public Producer createProducer() throws Exception {
         if (Jt400Type.DTAQ == configuration.getType()) {
             return new Jt400DataQueueProducer(this);
@@ -88,7 +80,7 @@ public class Jt400Endpoint extends DefaultEndpoint {
     @Override
     public Consumer createConsumer(Processor processor) throws Exception {
         if (Jt400Type.DTAQ == configuration.getType()) {
-            Consumer consumer = new Jt400DataQueueConsumer(this);
+            Consumer consumer = new Jt400DataQueueConsumer(this, processor);
             configureConsumer(consumer);
             return consumer;
         } else {
@@ -260,4 +252,12 @@ public class Jt400Endpoint extends DefaultEndpoint {
         return configuration.isSecured();
     }
 
+    public int getReadTimeout() {
+        return configuration.getReadTimeout();
+    }
+
+    public void setReadTimeout(int readTimeout) {
+        configuration.setReadTimeout(readTimeout);
+    }
+
 }


Mime
View raw message