You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2012/02/22 08:06:58 UTC
svn commit: r1292147 - in
/camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400:
Jt400DataQueueConsumer.java Jt400DataQueueEndpoint.java
Jt400DataQueueProducer.java
Author: davsclaus
Date: Wed Feb 22 07:06:57 2012
New Revision: 1292147
URL: http://svn.apache.org/viewvc?rev=1292147&view=rev
Log:
CAMEL-5015: Added support for keyed data queues to camel-jt400. Thanks to Jeff Costello for the patch.
Modified:
camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueEndpoint.java
camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueProducer.java
Modified: camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java?rev=1292147&r1=1292146&r2=1292147&view=diff
==============================================================================
--- camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java (original)
+++ camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueConsumer.java Wed Feb 22 07:06:57 2012
@@ -16,22 +16,18 @@
*/
package org.apache.camel.component.jt400;
-import java.io.IOException;
import com.ibm.as400.access.AS400;
-import com.ibm.as400.access.AS400SecurityException;
+import com.ibm.as400.access.BaseDataQueue;
import com.ibm.as400.access.DataQueue;
import com.ibm.as400.access.DataQueueEntry;
-import com.ibm.as400.access.ErrorCompletingRequestException;
-import com.ibm.as400.access.IllegalObjectTypeException;
-import com.ibm.as400.access.ObjectDoesNotExistException;
-
+import com.ibm.as400.access.KeyedDataQueue;
+import com.ibm.as400.access.KeyedDataQueueEntry;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.jt400.Jt400DataQueueEndpoint.Format;
import org.apache.camel.impl.DefaultExchange;
import org.apache.camel.impl.PollingConsumerSupport;
-
/**
* {@link org.apache.camel.PollingConsumer} that polls a data queue for data
*/
@@ -79,46 +75,79 @@ public class Jt400DataQueueConsumer exte
* <code>byte[]</code>. If the endpoint's format is set to
* {@link Format#text}, the data queue entry's data will be received/sent as
* a <code>String</code>.
+ * <p/>
+ * The following message headers may be set by the receiver
+ * <ul>
+ * <li>SENDER_INFORMATION: The Sender Information from the Data Queue</li>
+ * <li>KEY: The message key if the endpoint is configured to connect to a <code>KeyedDataQueue</code></li>
+ * </ul>
*
* @param timeout time to wait when reading from data queue. A value of -1
- * indicates a blocking read.
+ * indicates a blocking read.
*/
public Exchange receive(long timeout) {
- DataQueue queue = endpoint.getDataQueue();
+ BaseDataQueue queue = endpoint.getDataQueue();
try {
- DataQueueEntry entry;
- if (timeout >= 0) {
- int seconds = (int)timeout / 1000;
- log.trace("Reading from data queue: {} with {} seconds timeout", queue.getName(), seconds);
- entry = queue.read(seconds);
+ if (endpoint.isKeyed()) {
+ return receive((KeyedDataQueue) queue, timeout);
} else {
- log.trace("Reading from data queue: {} with no timeout", queue.getName());
- entry = queue.read(-1);
+ return receive((DataQueue) queue, timeout);
}
+ } catch (Exception e) {
+ throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
+ }
+ }
- Exchange exchange = new DefaultExchange(endpoint.getCamelContext());
- if (entry != null) {
- if (endpoint.getFormat() == Format.binary) {
- exchange.getIn().setBody(entry.getData());
- } else {
- exchange.getIn().setBody(entry.getString());
- }
- return exchange;
+ private Exchange receive(DataQueue queue, long timeout) throws Exception {
+ DataQueueEntry entry;
+ if (timeout >= 0) {
+ int seconds = (int) timeout / 1000;
+ log.trace("Reading from data queue: {} with {} seconds timeout", queue.getName(), seconds);
+ entry = queue.read(seconds);
+ } else {
+ log.trace("Reading from data queue: {} with no timeout", queue.getName());
+ entry = queue.read(-1);
+ }
+
+ Exchange exchange = new DefaultExchange(endpoint.getCamelContext());
+ if (entry != null) {
+ exchange.getIn().setHeader(Jt400DataQueueEndpoint.SENDER_INFORMATION, entry.getSenderInformation());
+ if (endpoint.getFormat() == Format.binary) {
+ exchange.getIn().setBody(entry.getData());
+ } else {
+ exchange.getIn().setBody(entry.getString());
}
- } catch (AS400SecurityException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
- } catch (ErrorCompletingRequestException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
- } catch (IOException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
- } catch (IllegalObjectTypeException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
- } catch (InterruptedException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
- } catch (ObjectDoesNotExistException e) {
- throw new RuntimeCamelException("Unable to read from data queue: " + queue.getName(), e);
+ return exchange;
}
return null;
}
+ private Exchange receive(KeyedDataQueue queue, long timeout) throws Exception {
+ String key = endpoint.getSearchKey();
+ String searchType = endpoint.getSearchType().name();
+ KeyedDataQueueEntry entry;
+ if (timeout >= 0) {
+ int seconds = (int) timeout / 1000;
+ log.trace("Reading from data queue: {} with {} seconds timeout", queue.getName(), seconds);
+ entry = queue.read(key, seconds, searchType);
+ } else {
+ log.trace("Reading from data queue: {} with no timeout", queue.getName());
+ entry = queue.read(key, -1, searchType);
+ }
+
+ Exchange exchange = new DefaultExchange(endpoint.getCamelContext());
+ if (entry != null) {
+ exchange.getIn().setHeader(Jt400DataQueueEndpoint.SENDER_INFORMATION, entry.getSenderInformation());
+ if (endpoint.getFormat() == Format.binary) {
+ exchange.getIn().setBody(entry.getData());
+ exchange.getIn().setHeader(Jt400DataQueueEndpoint.KEY, entry.getKey());
+ } else {
+ exchange.getIn().setBody(entry.getString());
+ exchange.getIn().setHeader(Jt400DataQueueEndpoint.KEY, entry.getKeyString());
+ }
+
+ return exchange;
+ }
+ return null;
+ }
}
Modified: camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueEndpoint.java?rev=1292147&r1=1292146&r2=1292147&view=diff
==============================================================================
--- camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueEndpoint.java (original)
+++ camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueEndpoint.java Wed Feb 22 07:06:57 2012
@@ -21,8 +21,9 @@ import java.net.URI;
import java.net.URISyntaxException;
import com.ibm.as400.access.AS400;
+import com.ibm.as400.access.BaseDataQueue;
import com.ibm.as400.access.DataQueue;
-
+import com.ibm.as400.access.KeyedDataQueue;
import org.apache.camel.CamelException;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Producer;
@@ -35,6 +36,16 @@ import org.slf4j.LoggerFactory;
*/
public class Jt400DataQueueEndpoint extends DefaultPollingEndpoint {
+ public static final String KEY = "KEY";
+ public static final String SENDER_INFORMATION = "SENDER_INFORMATION";
+
+ /**
+ * SearchTypes for reading from Keyed Data Queues
+ */
+ public enum SearchType {
+ EQ, NE, LT, LE, GT, GE;
+ }
+
/**
* Enumeration of supported data formats
*/
@@ -49,13 +60,16 @@ public class Jt400DataQueueEndpoint exte
*/
binary;
}
-
+
private static final transient Logger LOG = LoggerFactory.getLogger(Jt400DataQueueEndpoint.class);
private final AS400 system;
private final String objectPath;
- private DataQueue dataqueue;
+ private BaseDataQueue dataQueue;
private Format format = Format.text;
+ private boolean keyed;
+ private String searchKey;
+ private SearchType searchType = SearchType.EQ;
/**
* Creates a new AS/400 data queue endpoint
@@ -74,7 +88,7 @@ public class Jt400DataQueueEndpoint exte
try {
system.setGuiAvailable(false);
} catch (PropertyVetoException e) {
- LOG.warn("Failed do disable AS/400 prompting in the environment running Camel.", e);
+ LOG.warn("Failed to disable AS/400 prompting in the environment running Camel. This exception will be ignored.", e);
}
}
@@ -90,6 +104,30 @@ public class Jt400DataQueueEndpoint exte
return format;
}
+ public void setKeyed(boolean keyed) {
+ this.keyed = keyed;
+ }
+
+ public boolean isKeyed() {
+ return keyed;
+ }
+
+ public void setSearchKey(String searchKey) {
+ this.searchKey = searchKey;
+ }
+
+ public String getSearchKey() {
+ return searchKey;
+ }
+
+ public void setSearchType(SearchType searchType) {
+ this.searchType = searchType;
+ }
+
+ public SearchType getSearchType() {
+ return searchType;
+ }
+
public void setGuiAvailable(boolean guiAvailable) throws PropertyVetoException {
this.system.setGuiAvailable(guiAvailable);
}
@@ -99,6 +137,7 @@ public class Jt400DataQueueEndpoint exte
return new Jt400DataQueueConsumer(this);
}
+ @Override
public Producer createProducer() throws Exception {
return new Jt400DataQueueProducer(this);
}
@@ -107,11 +146,11 @@ public class Jt400DataQueueEndpoint exte
return system;
}
- protected DataQueue getDataQueue() {
- if (dataqueue == null) {
- dataqueue = new DataQueue(system, objectPath);
+ protected BaseDataQueue getDataQueue() {
+ if (dataQueue == null) {
+ dataQueue = keyed ? new KeyedDataQueue(system, objectPath) : new DataQueue(system, objectPath);
}
- return dataqueue;
+ return dataQueue;
}
public boolean isSingleton() {
Modified: camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueProducer.java?rev=1292147&r1=1292146&r2=1292147&view=diff
==============================================================================
--- camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueProducer.java (original)
+++ camel/trunk/components/camel-jt400/src/main/java/org/apache/camel/component/jt400/Jt400DataQueueProducer.java Wed Feb 22 07:06:57 2012
@@ -17,7 +17,9 @@
package org.apache.camel.component.jt400;
import com.ibm.as400.access.AS400;
+import com.ibm.as400.access.BaseDataQueue;
import com.ibm.as400.access.DataQueue;
+import com.ibm.as400.access.KeyedDataQueue;
import org.apache.camel.Exchange;
import org.apache.camel.Producer;
import org.apache.camel.component.jt400.Jt400DataQueueEndpoint.Format;
@@ -41,9 +43,20 @@ public class Jt400DataQueueProducer exte
* data will be sent as a <code>byte[]</code>. If the endpoint's format is
* set to {@link Format#text}, the data queue entry's data will be sent as a
* <code>String</code>.
+ * <p/>
+ * If the endpoint is configured to publish to a {@link KeyedDataQueue},
+ * then the {@link org.apache.camel.Message} header <code>KEY</code> must be set.
*/
public void process(Exchange exchange) throws Exception {
- DataQueue queue = endpoint.getDataQueue();
+ BaseDataQueue queue = endpoint.getDataQueue();
+ if (endpoint.isKeyed()) {
+ process((KeyedDataQueue) queue, exchange);
+ } else {
+ process((DataQueue) queue, exchange);
+ }
+ }
+
+ private void process(DataQueue queue, Exchange exchange) throws Exception {
if (endpoint.getFormat() == Format.binary) {
queue.write(exchange.getIn().getBody(byte[].class));
} else {
@@ -51,6 +64,14 @@ public class Jt400DataQueueProducer exte
}
}
+ private void process(KeyedDataQueue queue, Exchange exchange) throws Exception {
+ if (endpoint.getFormat() == Format.binary) {
+ queue.write(exchange.getIn().getHeader(Jt400DataQueueEndpoint.KEY, byte[].class), exchange.getIn().getBody(byte[].class));
+ } else {
+ queue.write(exchange.getIn().getHeader(Jt400DataQueueEndpoint.KEY, String.class), exchange.getIn().getBody(String.class));
+ }
+ }
+
@Override
protected void doStart() throws Exception {
if (!endpoint.getSystem().isConnected()) {