diff --git a/src/main/java/teetime/stage/basic/distributor/CloneStrategy.java b/src/main/java/teetime/stage/basic/distributor/CloneStrategy.java
index 9512aad0c39eee4528916429423b23bad95ab199..29df8f56e23bfb653a30e69974ed750fa3d4cbfb 100644
--- a/src/main/java/teetime/stage/basic/distributor/CloneStrategy.java
+++ b/src/main/java/teetime/stage/basic/distributor/CloneStrategy.java
@@ -19,13 +19,13 @@ import teetime.framework.OutputPort;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public final class CloneStrategy<T> implements IDistributorStrategy<T> {
+public final class CloneStrategy implements IDistributorStrategy {
@Override
- public boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
+ public <T> boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
throw new UnsupportedOperationException();
}
diff --git a/src/main/java/teetime/stage/basic/distributor/CopyByReferenceStrategy.java b/src/main/java/teetime/stage/basic/distributor/CopyByReferenceStrategy.java
index 3a31976eac4d3831fce0230c543e27631625ed39..ac6ad95c17fd872fff6e4c379bdd5f8086b557ff 100644
--- a/src/main/java/teetime/stage/basic/distributor/CopyByReferenceStrategy.java
+++ b/src/main/java/teetime/stage/basic/distributor/CopyByReferenceStrategy.java
@@ -19,13 +19,13 @@ import teetime.framework.OutputPort;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public final class CopyByReferenceStrategy<T> implements IDistributorStrategy<T> {
+public final class CopyByReferenceStrategy implements IDistributorStrategy {
@Override
- public boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
+ public <T> boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
for (final OutputPort<T> outputPort : outputPorts) {
outputPort.send(element);
}
diff --git a/src/main/java/teetime/stage/basic/distributor/Distributor.java b/src/main/java/teetime/stage/basic/distributor/Distributor.java
index 5fecfe44430510cf97bfdd17097bd6cee527bc84..aab742cfa5d0040a607695623003a62b72a2836d 100644
--- a/src/main/java/teetime/stage/basic/distributor/Distributor.java
+++ b/src/main/java/teetime/stage/basic/distributor/Distributor.java
@@ -29,7 +29,15 @@ import teetime.framework.OutputPort;
*/
public class Distributor<T> extends AbstractConsumerStage<T> {
- private IDistributorStrategy<T> strategy = new RoundRobinStrategy<T>();
+ private IDistributorStrategy strategy;
+
+ public Distributor() {
+ this(new RoundRobinStrategy());
+ }
+
+ public Distributor(final IDistributorStrategy strategy) {
+ this.strategy = strategy;
+ }
@SuppressWarnings("unchecked")
@Override
@@ -41,11 +49,11 @@ public class Distributor<T> extends AbstractConsumerStage<T> {
return this.createOutputPort();
}
- public IDistributorStrategy<T> getStrategy() {
+ public IDistributorStrategy getStrategy() {
return this.strategy;
}
- public void setStrategy(final IDistributorStrategy<T> strategy) {
+ public void setStrategy(final IDistributorStrategy strategy) {
this.strategy = strategy;
}
diff --git a/src/main/java/teetime/stage/basic/distributor/IDistributorStrategy.java b/src/main/java/teetime/stage/basic/distributor/IDistributorStrategy.java
index 69e537877b6b04181f527e6f59096d5557dd259b..9f67ccb66b9f1fc680c7c6550880e63aad68c5ad 100644
--- a/src/main/java/teetime/stage/basic/distributor/IDistributorStrategy.java
+++ b/src/main/java/teetime/stage/basic/distributor/IDistributorStrategy.java
@@ -19,11 +19,11 @@ import teetime.framework.OutputPort;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public interface IDistributorStrategy<T> {
+public interface IDistributorStrategy {
- public boolean distribute(final OutputPort<T>[] allOutputPorts, final T element);
+ public <T> boolean distribute(final OutputPort<T>[] allOutputPorts, final T element);
}
diff --git a/src/main/java/teetime/stage/basic/distributor/RoundRobinStrategy.java b/src/main/java/teetime/stage/basic/distributor/RoundRobinStrategy.java
index 64bf90f0fc8da79b3b2c0b999c35d6d7ace319be..e43ce7847fcf90b9acf207be56463dfd365ac1c6 100644
--- a/src/main/java/teetime/stage/basic/distributor/RoundRobinStrategy.java
+++ b/src/main/java/teetime/stage/basic/distributor/RoundRobinStrategy.java
@@ -19,15 +19,15 @@ import teetime.framework.OutputPort;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public final class RoundRobinStrategy<T> implements IDistributorStrategy<T> {
+public final class RoundRobinStrategy implements IDistributorStrategy {
private int index = 0;
@Override
- public boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
+ public <T> boolean distribute(final OutputPort<T>[] outputPorts, final T element) {
final OutputPort<T> outputPort = this.getNextPortInRoundRobinOrder(outputPorts);
outputPort.send(element);
@@ -35,7 +35,7 @@ public final class RoundRobinStrategy<T> implements IDistributorStrategy<T> {
return true;
}
- private OutputPort<T> getNextPortInRoundRobinOrder(final OutputPort<T>[] outputPorts) {
+ private <T> OutputPort<T> getNextPortInRoundRobinOrder(final OutputPort<T>[] outputPorts) {
final OutputPort<T> outputPort = outputPorts[this.index];
this.index = (this.index + 1) % outputPorts.length;
diff --git a/src/main/java/teetime/stage/basic/merger/IMergerStrategy.java b/src/main/java/teetime/stage/basic/merger/IMergerStrategy.java
index 0114cf9b54bd1f614fb9de9bd99c7e136bd3859c..d001f86ecc62facae8f95546063cff150db17054 100644
--- a/src/main/java/teetime/stage/basic/merger/IMergerStrategy.java
+++ b/src/main/java/teetime/stage/basic/merger/IMergerStrategy.java
@@ -17,11 +17,11 @@ package teetime.stage.basic.merger;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public interface IMergerStrategy<T> {
+public interface IMergerStrategy {
- public T getNextInput(Merger<T> merger);
+ public <T> T getNextInput(Merger<T> merger);
}
diff --git a/src/main/java/teetime/stage/basic/merger/Merger.java b/src/main/java/teetime/stage/basic/merger/Merger.java
index 69b778a4751b8777cf9fab2656972b7af4c4daff..aa84f0e34b7479ef6db5ffaad55c16fef2ef25fc 100644
--- a/src/main/java/teetime/stage/basic/merger/Merger.java
+++ b/src/main/java/teetime/stage/basic/merger/Merger.java
@@ -42,10 +42,18 @@ public final class Merger<T> extends AbstractStage {
private final OutputPort<T> outputPort = this.createOutputPort();
- private IMergerStrategy<T> strategy = new RoundRobinStrategy<T>();
+ private IMergerStrategy strategy;
private final Map<Class<?>, Set<InputPort<?>>> signalMap = new HashMap<Class<?>, Set<InputPort<?>>>();
+ public Merger() {
+ this(new RoundRobinStrategy());
+ }
+
+ public Merger(final IMergerStrategy strategy) {
+ this.strategy = strategy;
+ }
+
@Override
public void executeWithPorts() {
final T token = this.strategy.getNextInput(this);
@@ -90,11 +98,11 @@ public final class Merger<T> extends AbstractStage {
}
- public IMergerStrategy<T> getMergerStrategy() {
+ public IMergerStrategy getMergerStrategy() {
return this.strategy;
}
- public void setStrategy(final IMergerStrategy<T> strategy) {
+ public void setStrategy(final IMergerStrategy strategy) {
this.strategy = strategy;
}
diff --git a/src/main/java/teetime/stage/basic/merger/RoundRobinStrategy.java b/src/main/java/teetime/stage/basic/merger/RoundRobinStrategy.java
index 3edb04787f01f3d2d26587def8638e13bb7aa265..0ed2af2f9567c80e0fc967eb884f8af324ccb1ff 100644
--- a/src/main/java/teetime/stage/basic/merger/RoundRobinStrategy.java
+++ b/src/main/java/teetime/stage/basic/merger/RoundRobinStrategy.java
@@ -19,15 +19,15 @@ import teetime.framework.InputPort;
/**
* @author Nils Christian Ehmke
- *
- * @since 1.10
+ *
+ * @since 1.0
*/
-public final class RoundRobinStrategy<T> implements IMergerStrategy<T> {
+public final class RoundRobinStrategy implements IMergerStrategy {
private int index = 0;
@Override
- public T getNextInput(final Merger<T> merger) {
+ public <T> T getNextInput(final Merger<T> merger) {
@SuppressWarnings("unchecked")
InputPort<T>[] inputPorts = (InputPort<T>[]) merger.getInputPorts();
int size = inputPorts.length;
@@ -42,7 +42,7 @@ public final class RoundRobinStrategy<T> implements IMergerStrategy<T> {
return null;
}
- private InputPort<T> getNextPortInRoundRobinOrder(final InputPort<T>[] inputPorts) {
+ private <T> InputPort<T> getNextPortInRoundRobinOrder(final InputPort<T>[] inputPorts) {
InputPort<T> inputPort = inputPorts[this.index];
this.index = (this.index + 1) % inputPorts.length;
diff --git a/src/main/java/teetime/stage/io/EveryXthPrinter.java b/src/main/java/teetime/stage/io/EveryXthPrinter.java
index 1af922a1521d94f5db344d878bf751771c461aa1..aacc5936acd18870ab2215ffe859ac6398e5139d 100644
--- a/src/main/java/teetime/stage/io/EveryXthPrinter.java
+++ b/src/main/java/teetime/stage/io/EveryXthPrinter.java
@@ -29,7 +29,7 @@ public final class EveryXthPrinter<T> extends Stage {
pipeFactory.create(distributor.getNewOutputPort(), everyXthStage.getInputPort());
pipeFactory.create(everyXthStage.getOutputPort(), printer.getInputPort());
- distributor.setStrategy(new CopyByReferenceStrategy<T>());
+ distributor.setStrategy(new CopyByReferenceStrategy());
}
@Override
diff --git a/src/site/resources/browserconfig.xml b/src/site/resources/browserconfig.xml
index bcbd6c9ff886964340341d7d629edaf071f77894..587f54d577a87e56ca0d70b4ffa46eccb694a28b 100644
--- a/src/site/resources/browserconfig.xml
+++ b/src/site/resources/browserconfig.xml
@@ -1,12 +1,12 @@
-<?xml version="1.0" encoding="utf-8"?>
-<browserconfig>
- <msapplication>
- <tile>
- <square70x70logo src="/mstile-70x70.png"/>
- <square150x150logo src="/mstile-150x150.png"/>
- <square310x310logo src="/mstile-310x310.png"/>
- <wide310x150logo src="/mstile-310x150.png"/>
- <TileColor>#2b5797</TileColor>
- </tile>
- </msapplication>
-</browserconfig>
+<?xml version="1.0" encoding="utf-8"?>
+<browserconfig>
+ <msapplication>
+ <tile>
+ <square70x70logo src="/mstile-70x70.png"/>
+ <square150x150logo src="/mstile-150x150.png"/>
+ <square310x310logo src="/mstile-310x310.png"/>
+ <wide310x150logo src="/mstile-310x150.png"/>
+ <TileColor>#2b5797</TileColor>
+ </tile>
+ </msapplication>
+</browserconfig>
diff --git a/src/site/resources/images/teetime.pptx b/src/site/resources/images/teetime.pptx
index 187ff4868a46f3693ebeaa838d434a6a811ec42e..4abb18e1a4d3f7885dbd4b1e2e432ab0310ce63f 100644
Binary files a/src/site/resources/images/teetime.pptx and b/src/site/resources/images/teetime.pptx differ
diff --git a/src/test/java/teetime/stage/basic/distributor/DistributorTest.java b/src/test/java/teetime/stage/basic/distributor/DistributorTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..58116c478cc15846765ff036d19711aec890862d
--- /dev/null
+++ b/src/test/java/teetime/stage/basic/distributor/DistributorTest.java
@@ -0,0 +1,100 @@
+package teetime.stage.basic.distributor;
+
+import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import teetime.framework.pipe.IPipeFactory;
+import teetime.framework.pipe.SingleElementPipeFactory;
+import teetime.stage.CollectorSink;
+
+/**
+ * @author Nils Christian Ehmke
+ *
+ * @since 1.0
+ */
+public class DistributorTest {
+
+ @Rule
+ public ExpectedException expectedException = ExpectedException.none();
+
+ private Distributor<Integer> distributorUnderTest;
+ private CollectorSink<Integer> fstCollector;
+ private CollectorSink<Integer> sndCollector;
+
+ @Before
+ public void initializeDistributor() throws Exception {
+ this.distributorUnderTest = new Distributor<Integer>();
+ this.fstCollector = new CollectorSink<Integer>();
+ this.sndCollector = new CollectorSink<Integer>();
+
+ final IPipeFactory pipeFactory = new SingleElementPipeFactory();
+ pipeFactory.create(this.distributorUnderTest.getNewOutputPort(), this.fstCollector.getInputPort());
+ pipeFactory.create(this.distributorUnderTest.getNewOutputPort(), this.sndCollector.getInputPort());
+
+ distributorUnderTest.onStarting();
+ }
+
+ @Test
+ public void roundRobinShouldWork() {
+ distributorUnderTest.setStrategy(new RoundRobinStrategy());
+
+ this.distributorUnderTest.execute(1);
+ this.distributorUnderTest.execute(2);
+ this.distributorUnderTest.execute(3);
+ this.distributorUnderTest.execute(4);
+ this.distributorUnderTest.execute(5);
+
+ assertThat(this.fstCollector.getElements(), contains(1, 3, 5));
+ assertThat(this.sndCollector.getElements(), contains(2, 4));
+ }
+
+ @Test
+ public void singleElementRoundRobinShouldWork() {
+ distributorUnderTest.setStrategy(new RoundRobinStrategy());
+
+ this.distributorUnderTest.execute(1);
+
+ assertThat(this.fstCollector.getElements(), contains(1));
+ assertThat(this.sndCollector.getElements(), is(empty()));
+ }
+
+ @Test
+ public void copyByReferenceShouldWork() {
+ distributorUnderTest.setStrategy(new CopyByReferenceStrategy());
+
+ this.distributorUnderTest.execute(1);
+ this.distributorUnderTest.execute(2);
+ this.distributorUnderTest.execute(3);
+ this.distributorUnderTest.execute(4);
+ this.distributorUnderTest.execute(5);
+
+ assertThat(this.fstCollector.getElements(), contains(1, 2, 3, 4, 5));
+ assertThat(this.sndCollector.getElements(), contains(1, 2, 3, 4, 5));
+ }
+
+ @Test
+ public void singleElementCopyByReferenceShouldWork() {
+ distributorUnderTest.setStrategy(new CopyByReferenceStrategy());
+
+ this.distributorUnderTest.execute(1);
+
+ assertThat(this.fstCollector.getElements(), contains(1));
+ assertThat(this.sndCollector.getElements(), contains(1));
+ }
+
+ @Test
+ public void cloneShouldNotWork() {
+ distributorUnderTest.setStrategy(new CloneStrategy());
+
+ expectedException.expect(UnsupportedOperationException.class);
+ this.distributorUnderTest.execute(1);
+ }
+
+}
diff --git a/src/test/java/teetime/stage/basic/merger/MergerSignalTest.java b/src/test/java/teetime/stage/basic/merger/MergerSignalTest.java
new file mode 100644
index 0000000000000000000000000000000000000000..cbda2f77c50b7f8e1b465b61dfff9a8c08cf5119
--- /dev/null
+++ b/src/test/java/teetime/stage/basic/merger/MergerSignalTest.java
@@ -0,0 +1,86 @@
+package teetime.stage.basic.merger;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import teetime.framework.InputPort;
+import teetime.framework.signal.StartingSignal;
+import teetime.framework.signal.TerminatingSignal;
+
+public class MergerSignalTest {
+
+ private Merger<Integer> merger;
+ private InputPort<Integer> firstPort;
+ private InputPort<Integer> secondPort;
+ private MergerTestingPipe testPipe;
+
+ public void beforeSignalTesting() {
+ merger = new Merger<Integer>();
+
+ firstPort = merger.getNewInputPort();
+ secondPort = merger.getNewInputPort();
+
+ testPipe = new MergerTestingPipe();
+ merger.getOutputPort().setPipe(testPipe);
+ }
+
+ @Test
+ public void testSameSignal() {
+ this.beforeSignalTesting();
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+
+ merger.onSignal(new StartingSignal(), secondPort);
+ Assert.assertTrue(testPipe.startSent());
+ }
+
+ @Test
+ public void testDifferentSignals() {
+ this.beforeSignalTesting();
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+
+ merger.onSignal(new TerminatingSignal(), secondPort);
+ Assert.assertFalse(testPipe.startSent());
+ }
+
+ @Test
+ public void testInterleavedSignals() {
+ this.beforeSignalTesting();
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+ Assert.assertFalse(testPipe.terminateSent());
+
+ merger.onSignal(new TerminatingSignal(), secondPort);
+ Assert.assertFalse(testPipe.startSent());
+ Assert.assertFalse(testPipe.terminateSent());
+
+ merger.onSignal(new TerminatingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+ Assert.assertTrue(testPipe.terminateSent());
+
+ merger.onSignal(new TerminatingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+ Assert.assertTrue(testPipe.terminateSent());
+
+ merger.onSignal(new StartingSignal(), secondPort);
+ Assert.assertTrue(testPipe.startSent());
+ Assert.assertTrue(testPipe.terminateSent());
+ }
+
+ @Test
+ public void testMultipleSignals() {
+ this.beforeSignalTesting();
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+
+ merger.onSignal(new StartingSignal(), firstPort);
+ Assert.assertFalse(testPipe.startSent());
+
+ merger.onSignal(new StartingSignal(), secondPort);
+ Assert.assertTrue(testPipe.startSent());
+ }
+}
diff --git a/src/test/java/teetime/stage/basic/merger/MergerTest.java b/src/test/java/teetime/stage/basic/merger/MergerTest.java
index 7afbb35d67f3af495dde66bd87f0a91ee219fa0f..0ea68fc12b2b07456a53cdbada525f74c65f31a5 100644
--- a/src/test/java/teetime/stage/basic/merger/MergerTest.java
+++ b/src/test/java/teetime/stage/basic/merger/MergerTest.java
@@ -1,86 +1,60 @@
package teetime.stage.basic.merger;
-import org.junit.Assert;
+import static org.hamcrest.Matchers.contains;
+import static org.junit.Assert.assertThat;
+
+import org.junit.Before;
import org.junit.Test;
-import teetime.framework.InputPort;
-import teetime.framework.signal.StartingSignal;
-import teetime.framework.signal.TerminatingSignal;
+import teetime.framework.pipe.IPipeFactory;
+import teetime.framework.pipe.SingleElementPipeFactory;
+import teetime.stage.CollectorSink;
+import teetime.stage.InitialElementProducer;
+/**
+ * @author Nils Christian Ehmke
+ *
+ * @since 1.0
+ */
public class MergerTest {
- private Merger<Integer> merger;
- private InputPort<Integer> firstPort;
- private InputPort<Integer> secondPort;
- private MergerTestingPipe testPipe;
+ private Merger<Integer> mergerUnderTest;
+ private CollectorSink<Integer> collector;
+ private InitialElementProducer<Integer> fstProducer;
+ private InitialElementProducer<Integer> sndProducer;
- public void beforeSignalTesting() {
- merger = new Merger<Integer>();
+ @Before
+ public void initializeMerger() throws Exception {
+ this.mergerUnderTest = new Merger<Integer>();
+ this.collector = new CollectorSink<Integer>();
+ this.fstProducer = new InitialElementProducer<Integer>(1, 2, 3);
+ this.sndProducer = new InitialElementProducer<Integer>(4, 5, 6);
- firstPort = merger.getNewInputPort();
- secondPort = merger.getNewInputPort();
+ final IPipeFactory pipeFactory = new SingleElementPipeFactory();
+ pipeFactory.create(this.fstProducer.getOutputPort(), this.mergerUnderTest.getNewInputPort());
+ pipeFactory.create(this.sndProducer.getOutputPort(), this.mergerUnderTest.getNewInputPort());
+ pipeFactory.create(this.mergerUnderTest.getOutputPort(), this.collector.getInputPort());
- testPipe = new MergerTestingPipe();
- merger.getOutputPort().setPipe(testPipe);
+ mergerUnderTest.onStarting();
}
@Test
- public void testSameSignal() {
- this.beforeSignalTesting();
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
-
- merger.onSignal(new StartingSignal(), secondPort);
- Assert.assertTrue(testPipe.startSent());
- }
+ public void roundRobinShouldWork() {
+ mergerUnderTest.setStrategy(new RoundRobinStrategy());
- @Test
- public void testDifferentSignals() {
- this.beforeSignalTesting();
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
+ this.fstProducer.executeWithPorts();
+ this.sndProducer.executeWithPorts();
- merger.onSignal(new TerminatingSignal(), secondPort);
- Assert.assertFalse(testPipe.startSent());
+ assertThat(this.collector.getElements(), contains(1, 2, 3, 4, 5, 6));
}
@Test
- public void testInterleavedSignals() {
- this.beforeSignalTesting();
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
- Assert.assertFalse(testPipe.terminateSent());
-
- merger.onSignal(new TerminatingSignal(), secondPort);
- Assert.assertFalse(testPipe.startSent());
- Assert.assertFalse(testPipe.terminateSent());
-
- merger.onSignal(new TerminatingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
- Assert.assertTrue(testPipe.terminateSent());
+ public void roundRobinWithSingleProducerShouldWork() {
+ mergerUnderTest.setStrategy(new RoundRobinStrategy());
- merger.onSignal(new TerminatingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
- Assert.assertTrue(testPipe.terminateSent());
+ this.fstProducer.executeWithPorts();
- merger.onSignal(new StartingSignal(), secondPort);
- Assert.assertTrue(testPipe.startSent());
- Assert.assertTrue(testPipe.terminateSent());
+ assertThat(this.collector.getElements(), contains(1, 2, 3));
}
- @Test
- public void testMultipleSignals() {
- this.beforeSignalTesting();
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
-
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
-
- merger.onSignal(new StartingSignal(), firstPort);
- Assert.assertFalse(testPipe.startSent());
-
- merger.onSignal(new StartingSignal(), secondPort);
- Assert.assertTrue(testPipe.startSent());
- }
}