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 2010/08/25 17:31:13 UTC
svn commit: r989172 - in /camel/trunk/components/camel-stream/src:
main/java/org/apache/camel/component/stream/
test/java/org/apache/camel/component/stream/
Author: davsclaus
Date: Wed Aug 25 15:31:13 2010
New Revision: 989172
URL: http://svn.apache.org/viewvc?rev=989172&view=rev
Log:
CAMEL-3080: Added option groupLines to stream consumer.
Added:
camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java (with props)
Modified:
camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
Modified: camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java?rev=989172&r1=989171&r2=989172&view=diff
==============================================================================
--- camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java (original)
+++ camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamConsumer.java Wed Aug 25 15:31:13 2010
@@ -25,8 +25,10 @@ import java.io.InputStreamReader;
import java.net.URL;
import java.net.URLConnection;
import java.nio.charset.Charset;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import org.apache.camel.Exchange;
@@ -52,6 +54,7 @@ public class StreamConsumer extends Defa
private StreamEndpoint endpoint;
private String uri;
private boolean initialPromptDone;
+ private final List<Object> lines = new CopyOnWriteArrayList<Object>();
public StreamConsumer(StreamEndpoint endpoint, Processor processor, String uri) throws Exception {
super(endpoint, processor);
@@ -68,6 +71,10 @@ public class StreamConsumer extends Defa
executor = endpoint.getCamelContext().getExecutorServiceStrategy().newSingleThreadExecutor(this, endpoint.getEndpointUri());
executor.execute(this);
+
+ if (endpoint.getGroupLines() < 0) {
+ throw new IllegalArgumentException("Option groupLines must be 0 or positive number, was " + endpoint.getGroupLines());
+ }
}
@Override
@@ -78,6 +85,7 @@ public class StreamConsumer extends Defa
executor.shutdownNow();
executor = null;
}
+ lines.clear();
super.doStop();
}
@@ -151,14 +159,37 @@ public class StreamConsumer extends Defa
/**
* Strategy method for processing the line
*/
- protected void processLine(Object line) throws Exception {
- Exchange exchange = endpoint.createExchange();
+ protected synchronized void processLine(Object line) throws Exception {
+ if (endpoint.getGroupLines() > 0) {
+ // remember line
+ lines.add(line);
+
+ // should we flush lines?
+ if (lines.size() >= endpoint.getGroupLines()) {
+ // spit out lines
+ Exchange exchange = endpoint.createExchange();
+
+ // create message with the lines
+ Message msg = new DefaultMessage();
+ List<Object> copy = new ArrayList<Object>(lines);
+ msg.setBody(copy);
+ exchange.setIn(msg);
- Message msg = new DefaultMessage();
- msg.setBody(line);
- exchange.setIn(msg);
+ // clear lines
+ lines.clear();
- getProcessor().process(exchange);
+ getProcessor().process(exchange);
+ }
+ } else {
+ // single line
+ Exchange exchange = endpoint.createExchange();
+
+ Message msg = new DefaultMessage();
+ msg.setBody(line);
+ exchange.setIn(msg);
+
+ getProcessor().process(exchange);
+ }
}
/**
Modified: camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java?rev=989172&r1=989171&r2=989172&view=diff
==============================================================================
--- camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java (original)
+++ camel/trunk/components/camel-stream/src/main/java/org/apache/camel/component/stream/StreamEndpoint.java Wed Aug 25 15:31:13 2010
@@ -38,6 +38,7 @@ public class StreamEndpoint extends Defa
private String promptMessage;
private long promptDelay;
private long initialPromptDelay = 2000;
+ private int groupLines;
public StreamEndpoint(String endpointUri, Component component) throws Exception {
super(endpointUri, component);
@@ -134,6 +135,14 @@ public class StreamEndpoint extends Defa
this.scanStreamDelay = scanStreamDelay;
}
+ public int getGroupLines() {
+ return groupLines;
+ }
+
+ public void setGroupLines(int groupLines) {
+ this.groupLines = groupLines;
+ }
+
// Implementations
//-------------------------------------------------------------------------
Added: camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java?rev=989172&view=auto
==============================================================================
--- camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java (added)
+++ camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java Wed Aug 25 15:31:13 2010
@@ -0,0 +1,88 @@
+/**
+ * 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.stream;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.util.List;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @version $Revision$
+ */
+public class StreamGroupLinesTest extends CamelTestSupport {
+
+ private FileOutputStream fos;
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ deleteDirectory("./target/stream");
+ createDirectory("./target/stream");
+
+ File file = new File("./target/stream/streamfile.txt");
+ file = file.getAbsoluteFile();
+ file.createNewFile();
+
+ fos = new FileOutputStream(file);
+ fos.write("A\n".getBytes());
+ fos.write("B\n".getBytes());
+ fos.write("C\n".getBytes());
+ fos.write("D\n".getBytes());
+ fos.write("E\n".getBytes());
+ fos.write("F\n".getBytes());
+
+ fos.close();
+
+ super.setUp();
+ }
+
+ @Test
+ public void testGroupLines() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(2);
+
+ assertMockEndpointsSatisfied();
+
+ List list = mock.getExchanges().get(0).getIn().getBody(List.class);
+ assertEquals(3, list.size());
+ assertEquals("A", list.get(0));
+ assertEquals("B", list.get(1));
+ assertEquals("C", list.get(2));
+
+ List list2 = mock.getExchanges().get(1).getIn().getBody(List.class);
+ assertEquals(3, list2.size());
+ assertEquals("D", list2.get(0));
+ assertEquals("E", list2.get(1));
+ assertEquals("F", list2.get(2));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("stream:file?fileName=./target/stream/streamfile.txt&groupLines=3").to("mock:result");
+ }
+ };
+ }
+}
Propchange: camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: camel/trunk/components/camel-stream/src/test/java/org/apache/camel/component/stream/StreamGroupLinesTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date