[
https://issues.apache.org/jira/browse/APEXMALHAR-2106?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15319340#comment-15319340
]
ASF GitHub Bot commented on APEXMALHAR-2106:
--------------------------------------------
Github user bhupeshchawda commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/309#discussion_r66147299
--- Diff:
library/src/test/java/com/datatorrent/lib/stream/MultipleStreamMergerTest.java
---
@@ -0,0 +1,145 @@
+/**
+ * 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 com.datatorrent.lib.stream;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import javax.validation.ConstraintViolationException;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.LocalMode;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.lib.io.ConsoleOutputOperator;
+import com.datatorrent.lib.testbench.RandomWordGenerator;
+
+import static org.junit.Assert.assertEquals;
+
+public class MultipleStreamMergerTest {
+ private static Logger LOG =
LoggerFactory.getLogger(MultipleStreamMergerTest.class);
+
+ StreamMerger<byte[]> finalMerger;
+ ArrayList<MultipleStreamMerger<byte[]>.Stream> streamsToAddToDag;
+ ArrayList<MultipleStreamMerger<byte[]>.NamedMerger> operatorsToAdd;
+
+ @Before
+ public void setUp() throws Exception
+ {
+ finalMerger = new StreamMerger<>();
+ streamsToAddToDag = new ArrayList<>();
+ operatorsToAdd = new ArrayList<>();
+
+ }
+
+ @Test
+ public void mergeTwoStreams()
+ {
+ RandomWordGenerator randomWordGenerator = new RandomWordGenerator();
+ RandomWordGenerator randomWordGenerator2 = new RandomWordGenerator();
+
+ randomWordGenerator.setTuplesPerWindow(1);
+ randomWordGenerator2.setTuplesPerWindow(1);
+
+ MultipleStreamMerger<byte[]> merger = new MultipleStreamMerger<>();
+ merger.merge(randomWordGenerator.output)
+ .merge(randomWordGenerator2.output);
+
+ merger.constructMergeTree(streamsToAddToDag, operatorsToAdd);
+
+ assertEquals("Count of created streams", 2, streamsToAddToDag.size());
+ assertEquals("Count of created operators", 1, operatorsToAdd.size());
+
+ // Next check actual connections
+ assertEquals("Generator 1 stream", randomWordGenerator.output,
+ streamsToAddToDag.get(0).sourcePort);
+
+ assertEquals("Generator 2 stream", randomWordGenerator2.output,
+ streamsToAddToDag.get(1).sourcePort);
+
+ assertEquals("Final operator input_1",
operatorsToAdd.get(0).merger.data1, streamsToAddToDag.get(0).destPort);
+ assertEquals("Final operator input_2",
operatorsToAdd.get(0).merger.data2, streamsToAddToDag.get(1).destPort);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void mergeOneStream()
+ {
+ RandomWordGenerator randomWordGenerator = new RandomWordGenerator();
+ MultipleStreamMerger<byte[]> merger = new MultipleStreamMerger<>();
+ merger.merge(randomWordGenerator.output);
+ merger.constructMergeTree(streamsToAddToDag, operatorsToAdd);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void mergeZeroStream()
+ {
+ RandomWordGenerator randomWordGenerator = new RandomWordGenerator();
+ MultipleStreamMerger<byte[]> merger = new MultipleStreamMerger<>();
+ merger.constructMergeTree(streamsToAddToDag, operatorsToAdd);
+ }
+
+ static class Application implements StreamingApplication
+ {
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ LOG.debug("Application - PopulateDAG");
+ int count = 53;
+ RandomWordGenerator[] generators = new RandomWordGenerator[count];
+ MultipleStreamMerger<byte[]> merger = new MultipleStreamMerger<>();
+ for (int i = 0; i < count; i++) {
+ generators[i] = new RandomWordGenerator();
+ generators[i].setTuplesPerWindow(1);
+ dag.addOperator("Generator " + i, generators[i]);
+ merger.merge(generators[i].output);
+ }
+
+ merger.insertInto(dag, conf);
+
+ // This should connect all the relevant ports
+ dag.addModule("Merger", merger);
+
+ // And then we should see the output
+ ConsoleOutputOperator consoleOperator = dag.addOperator("console",
new ConsoleOutputOperator());
+ dag.addStream("merger-console", merger.streamOutput,
consoleOperator.input);
+ }
+ }
+
+ @Test
+ public void testApplication() throws IOException, Exception {
--- End diff --
Can you also add validation to this test? A very basic test could be done
by controlling the number of objects sent by each RandomWordGenerator and
counting them at the console operator.
> Support merging multiple streams with StreamMerger
> ---------------------------------------------------
>
> Key: APEXMALHAR-2106
> URL: https://issues.apache.org/jira/browse/APEXMALHAR-2106
> Project: Apache Apex Malhar
> Issue Type: New Feature
> Reporter: Ilya Ganelin
> Assignee: Ilya Ganelin
>
> To properly implement the Flatten transformation (and other Stream
> combination operations), Apex must support merging data from multiple
> sources. The StreamMerger operator can be improved to merge multiple streams,
> rather than just the two streams it can handle in the present implementation.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
