camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cmuel...@apache.org
Subject svn commit: r1324972 - in /camel/branches/camel-2.9.x/components/camel-jaxb/src: main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java test/java/org/apache/camel/example/DataFormatConcurrentTest.java
Date Wed, 11 Apr 2012 20:14:11 GMT
Author: cmueller
Date: Wed Apr 11 20:14:10 2012
New Revision: 1324972

URL: http://svn.apache.org/viewvc?rev=1324972&view=rev
Log:
CAMEL-3776: Add pooling support for JAXB data format

Modified:
    camel/branches/camel-2.9.x/components/camel-jaxb/src/main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java
    camel/branches/camel-2.9.x/components/camel-jaxb/src/test/java/org/apache/camel/example/DataFormatConcurrentTest.java

Modified: camel/branches/camel-2.9.x/components/camel-jaxb/src/main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-jaxb/src/main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java?rev=1324972&r1=1324971&r2=1324972&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-jaxb/src/main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-jaxb/src/main/java/org/apache/camel/converter/jaxb/JaxbDataFormat.java
Wed Apr 11 20:14:10 2012
@@ -21,6 +21,7 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.util.concurrent.locks.ReentrantLock;
 import javax.xml.bind.JAXBContext;
 import javax.xml.bind.JAXBElement;
 import javax.xml.bind.JAXBException;
@@ -66,6 +67,9 @@ public class JaxbDataFormat extends Serv
     private String partClass;
     private Class partialClass;
 
+    private Unmarshaller unmarshaller;
+    private ReentrantLock lock = new ReentrantLock();
+
     public JaxbDataFormat() {
     }
 
@@ -131,26 +135,29 @@ public class JaxbDataFormat extends Serv
     @SuppressWarnings("unchecked")
     public Object unmarshal(Exchange exchange, InputStream stream) throws IOException {
         try {
-            // must create a new instance of unmarshaller as its not thread safe
             Object answer;
-            Unmarshaller unmarshaller = getContext().createUnmarshaller();
 
-            if (partialClass != null) {
-                // partial unmarshalling
-                Source source;
-                if (needFiltering(exchange)) {
-                    source = new StreamSource(createNonXmlFilterReader(exchange, stream));
+            lock.lock();
+            try {
+                if (partialClass != null) {
+                    // partial unmarshalling
+                    Source source;
+                    if (needFiltering(exchange)) {
+                        source = new StreamSource(createNonXmlFilterReader(exchange, stream));
+                    } else {
+                        source = new StreamSource(stream);
+                    }
+                    answer = unmarshaller.unmarshal(source, partialClass);
                 } else {
-                    source = new StreamSource(stream);
-                }
-                answer = unmarshaller.unmarshal(source, partialClass);
-            } else {
-                if (needFiltering(exchange)) {
-                    NonXmlFilterReader reader = createNonXmlFilterReader(exchange, stream);
-                    answer = unmarshaller.unmarshal(reader);
-                } else  {
-                    answer = unmarshaller.unmarshal(stream);
+                    if (needFiltering(exchange)) {
+                        NonXmlFilterReader reader = createNonXmlFilterReader(exchange, stream);
+                        answer = unmarshaller.unmarshal(reader);
+                    } else  {
+                        answer = unmarshaller.unmarshal(stream);
+                    }
                 }
+            }  finally {
+                lock.unlock();
             }
 
             if (answer instanceof JAXBElement && isIgnoreJAXBElement()) {
@@ -264,6 +271,7 @@ public class JaxbDataFormat extends Serv
         if (partClass != null) {
             partialClass = camelContext.getClassResolver().resolveMandatoryClass(partClass);
         }
+        unmarshaller = getContext().createUnmarshaller();
     }
 
     @Override

Modified: camel/branches/camel-2.9.x/components/camel-jaxb/src/test/java/org/apache/camel/example/DataFormatConcurrentTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.9.x/components/camel-jaxb/src/test/java/org/apache/camel/example/DataFormatConcurrentTest.java?rev=1324972&r1=1324971&r2=1324972&view=diff
==============================================================================
--- camel/branches/camel-2.9.x/components/camel-jaxb/src/test/java/org/apache/camel/example/DataFormatConcurrentTest.java
(original)
+++ camel/branches/camel-2.9.x/components/camel-jaxb/src/test/java/org/apache/camel/example/DataFormatConcurrentTest.java
Wed Apr 11 20:14:10 2012
@@ -16,8 +16,10 @@
  */
 package org.apache.camel.example;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.component.mock.MockEndpoint;
@@ -27,13 +29,37 @@ import org.apache.camel.test.junit4.Came
 import org.junit.Test;
 
 /**
- * @version 
+ * @version
  */
 public class DataFormatConcurrentTest extends CamelTestSupport {
 
     private int size = 2000;
 
     @Test
+    public void testUnmarshallConcurrent() throws Exception {
+        int counter = 10000;
+        final String payload = "<purchaseOrder name='Wine' amount='123.45' price='2.22'/>";
+        final CountDownLatch latch = new CountDownLatch(counter);
+        template.setDefaultEndpointUri("direct:unmarshal");
+
+        ExecutorService pool = Executors.newFixedThreadPool(20);
+        //long start = System.currentTimeMillis();
+        for (int i = 0; i < counter; i++) {
+            pool.execute(new Runnable() {
+                public void run() {
+                    template.sendBody(payload);
+                    latch.countDown();
+                }
+            });
+        }
+
+        // should finish on fast machines in less than 3 seconds
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        //long end = System.currentTimeMillis();
+        //System.out.println("took " + (end - start) + "ms");
+    }
+
+    @Test
     public void testSendConcurrent() throws Exception {
         MockEndpoint mock = getMockEndpoint("mock:result");
         mock.expectedMessageCount(size);
@@ -73,6 +99,10 @@ public class DataFormatConcurrentTest ex
 
                 // use seda that supports concurrent consumers for concurrency
                 from("seda:start?size=" + size + "&concurrentConsumers=5").marshal(jaxb).convertBodyTo(String.class).to("mock:result");
+
+                from("direct:unmarshal")
+                        .unmarshal(jaxb)
+                        .to("mock:result");
             }
         };
     }



Mime
View raw message