diff --git a/src/main/java/teetime/framework/AbstractInterThreadPipe.java b/src/main/java/teetime/framework/AbstractInterThreadPipe.java
index 9821a41bbe6a66f3284a65f95ad0cc88c3917581..58d83fe8c74dbcd20507fccd77e2767774b9961a 100644
--- a/src/main/java/teetime/framework/AbstractInterThreadPipe.java
+++ b/src/main/java/teetime/framework/AbstractInterThreadPipe.java
@@ -23,6 +23,9 @@ public abstract class AbstractInterThreadPipe extends AbstractPipe {
this.signalQueue.offer(signal);
Thread owningThread = cachedTargetStage.getOwningThread();
+ if (owningThread == null) {
+ System.err.println("cachedTargetStage: " + cachedTargetStage);
+ }
if (null != owningThread && isThreadWaiting(owningThread)) { // FIXME remove the null check for performance
owningThread.interrupt();
}
diff --git a/src/performancetest/java/teetime/examples/ChwHomeComparisonMethodcallWithPorts.java b/src/performancetest/java/teetime/examples/ChwHomeComparisonMethodcallWithPorts.java
index 77d7dbaa3727a06e8a97b4cac189af19fa3b216f..faf67e8dfc91d00cbdf0e7af485033c4e22e8ca2 100644
--- a/src/performancetest/java/teetime/examples/ChwHomeComparisonMethodcallWithPorts.java
+++ b/src/performancetest/java/teetime/examples/ChwHomeComparisonMethodcallWithPorts.java
@@ -63,9 +63,12 @@ public class ChwHomeComparisonMethodcallWithPorts extends AbstractProfiledPerfor
// assertEquals(78, value17, 4.1); // +3
// since 13.12.2014 (incl.)
- assertEquals(40, value15, 4.1); // -28
+ // assertEquals(40, value15, 4.1); // -28
// assertEquals(43, value17, 4.1); // -35
+ // since 28.12.2014 (incl.)
+ assertEquals(30, value15, 4.1); // -10
+
// check speedup
assertEquals(2, (double) test19a.overallDurationInNs / test19b.overallDurationInNs, 0.3);
assertEquals(2.5, (double) test19a.overallDurationInNs / test19c.overallDurationInNs, 0.3);
diff --git a/src/performancetest/java/teetime/examples/ComparisonMethodcallWithPorts.java b/src/performancetest/java/teetime/examples/ComparisonMethodcallWithPorts.java
index e9652516668a8b7ba9d5e05d2bfcc24749abf52f..44b9227c3f2a3cd7c900a745709e56cf8b81e4dd 100644
--- a/src/performancetest/java/teetime/examples/ComparisonMethodcallWithPorts.java
+++ b/src/performancetest/java/teetime/examples/ComparisonMethodcallWithPorts.java
@@ -9,7 +9,6 @@ import org.junit.runners.Suite.SuiteClasses;
import teetime.examples.experiment01.MethodCallThoughputTimestampAnalysis1Test;
import teetime.examples.experiment09pipeimpls.MethodCallThoughputTimestampAnalysis9Test;
import teetime.examples.experiment11.MethodCallThoughputTimestampAnalysis11Test;
-import teetime.examples.experiment14.MethodCallThoughputTimestampAnalysis14Test;
import teetime.examples.experiment15.MethodCallThoughputTimestampAnalysis15Test;
import teetime.examples.experiment16.MethodCallThoughputTimestampAnalysis16Test;
import teetime.examples.experiment19.MethodCallThoughputTimestampAnalysis19Test;
@@ -21,7 +20,6 @@ import util.test.PerformanceCheckProfileRepository;
MethodCallThoughputTimestampAnalysis1Test.class,
MethodCallThoughputTimestampAnalysis9Test.class,
MethodCallThoughputTimestampAnalysis11Test.class,
- MethodCallThoughputTimestampAnalysis14Test.class,
MethodCallThoughputTimestampAnalysis15Test.class,
MethodCallThoughputTimestampAnalysis16Test.class,
MethodCallThoughputTimestampAnalysis19Test.class,
diff --git a/src/performancetest/java/teetime/examples/experiment09pipeimpls/AbstractPerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment09pipeimpls/AbstractPerformanceCheck.java
index 783af21b8663262eeb670fe124c226bbfe367a77..f821b423ef9c2fdfc0faf722494d6393718621ab 100644
--- a/src/performancetest/java/teetime/examples/experiment09pipeimpls/AbstractPerformanceCheck.java
+++ b/src/performancetest/java/teetime/examples/experiment09pipeimpls/AbstractPerformanceCheck.java
@@ -11,6 +11,7 @@ abstract class AbstractPerformanceCheck extends AbstractProfiledPerformanceAsser
protected PerformanceResult test01;
protected PerformanceResult test09CommittablePipes;
protected PerformanceResult test09SingleElementPipes;
+ protected PerformanceResult test09OrderedGrowableArrayPipes;
@Override
public void check() {
@@ -22,6 +23,9 @@ abstract class AbstractPerformanceCheck extends AbstractProfiledPerformanceAsser
testMethodIdentifier = MeasurementRepository.buildTestMethodIdentifier(MethodCallThoughputTimestampAnalysis9Test.class, "testSingleElementPipes");
test09SingleElementPipes = PerformanceTest.measurementRepository.performanceResults.get(testMethodIdentifier);
+
+ testMethodIdentifier = MeasurementRepository.buildTestMethodIdentifier(MethodCallThoughputTimestampAnalysis9Test.class, "testOrderedGrowableArrayPipes");
+ test09OrderedGrowableArrayPipes = PerformanceTest.measurementRepository.performanceResults.get(testMethodIdentifier);
}
}
diff --git a/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwHomePerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwHomePerformanceCheck.java
index 5b6cd1cb7b5bc30d76893f763b793696bc22e921..369326fc8cdf85b8b1d147b0210d03ff72e06be6 100644
--- a/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwHomePerformanceCheck.java
+++ b/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwHomePerformanceCheck.java
@@ -16,6 +16,7 @@ class ChwHomePerformanceCheck extends AbstractPerformanceCheck {
checkCommittablePipes();
checkSingleElementPipes();
+ checkOrderedGrowableArrayPipes();
}
private void checkCommittablePipes() {
@@ -36,9 +37,7 @@ class ChwHomePerformanceCheck extends AbstractPerformanceCheck {
// since 04.11.2014 (incl.)
// assertEquals(71, medianSpeedup, 2.1); // +27
// since 05.12.2014 (incl.)
- // assertEquals(45, medianSpeedup, 2.1); // -26
- // since 28.12.2014 (incl.)
- assertEquals(56, medianSpeedup, 2.1); // +11
+ assertEquals(45, medianSpeedup, 2.1); // -26 (45-56)
}
private void checkSingleElementPipes() {
@@ -60,4 +59,29 @@ class ChwHomePerformanceCheck extends AbstractPerformanceCheck {
assertEquals(24, medianSpeedup, 3.2); // -16
}
+ private void checkOrderedGrowableArrayPipes() {
+ double medianSpeedup = (double) test09OrderedGrowableArrayPipes.quantiles.get(0.5) / test01.quantiles.get(0.5);
+
+ System.out.println("medianSpeedup (09 ordered growable array pipes): " + medianSpeedup);
+
+ // until 25.06.2014 (incl.)
+ // assertEquals(60, (double) test14.quantiles.get(0.5) / test1.quantiles.get(0.5), 5.1);
+ // since 26.06.2014 (incl.)
+ // assertEquals(76, medianSpeedup, 5.1); // +16
+ // since 04.07.2014 (incl.)
+ // assertEquals(86, medianSpeedup, 5.1); // +16
+ // since 11.08.2014 (incl.)
+ // assertEquals(103, medianSpeedup, 5.1); // +17
+ // since 31.08.2014 (incl.)
+ // assertEquals(62, medianSpeedup, 2.1); // -41
+ // since 04.11.2014 (incl.)
+ // assertEquals(84, medianSpeedup, 2.1); // +22
+ // since 05.12.2014 (incl.)
+ // assertEquals(75, medianSpeedup, 2.1); // -9
+ // since 13.12.2014 (incl.)
+ // assertEquals(44, medianSpeedup, 2.1); // -31
+ // since 28.12.2014 (incl.)
+ assertEquals(46, medianSpeedup, 2.1); // +2
+ }
+
}
diff --git a/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwWorkPerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwWorkPerformanceCheck.java
index f29535c8da8b77cda0e0a86716a0339ed0da60cd..a84f8fab6b438d71f81f72937ae00872215607f2 100644
--- a/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwWorkPerformanceCheck.java
+++ b/src/performancetest/java/teetime/examples/experiment09pipeimpls/ChwWorkPerformanceCheck.java
@@ -15,6 +15,7 @@ class ChwWorkPerformanceCheck extends AbstractPerformanceCheck {
checkCommittablePipes();
checkSingleElementPipes();
+ checkOrderedGrowableArrayPipes();
}
private void checkCommittablePipes() {
@@ -52,4 +53,24 @@ class ChwWorkPerformanceCheck extends AbstractPerformanceCheck {
// since 14.10.2014 (incl.)
assertEquals(25, medianSpeedup, 3.1); // -31
}
+
+ private void checkOrderedGrowableArrayPipes() {
+ double medianSpeedup = (double) test09OrderedGrowableArrayPipes.quantiles.get(0.5) / test01.quantiles.get(0.5);
+
+ System.out.println("medianSpeedup (09 ordered growable array pipes): " + medianSpeedup);
+
+ // until 25.06.2014 (incl.)
+ // assertEquals(60, (double) test14.quantiles.get(0.5) / test1.quantiles.get(0.5), 5.1);
+ // since 26.06.2014 (incl.)
+ // assertEquals(76, medianSpeedup, 5.1); // +16
+ // since 04.07.2014 (incl.)
+ // assertEquals(86, medianSpeedup, 5.1); // +16
+ // since 27.08.2014 (incl.)
+ // assertEquals(102, medianSpeedup, 5.1); // +16
+ // since 14.10.2014 (incl.)
+ // assertEquals(81, medianSpeedup, 5.1); // -21
+ // since 19.12.2014 (incl.)
+ assertEquals(56, medianSpeedup, 5.1); // -25
+ }
+
}
diff --git a/src/performancetest/java/teetime/examples/experiment09pipeimpls/MethodCallThoughputTimestampAnalysis9Test.java b/src/performancetest/java/teetime/examples/experiment09pipeimpls/MethodCallThoughputTimestampAnalysis9Test.java
index f51b30424a75df3c79bd268b9ed610af1de24831..b4f1bcbc237021798b7ac326038cfeed3ababc55 100644
--- a/src/performancetest/java/teetime/examples/experiment09pipeimpls/MethodCallThoughputTimestampAnalysis9Test.java
+++ b/src/performancetest/java/teetime/examples/experiment09pipeimpls/MethodCallThoughputTimestampAnalysis9Test.java
@@ -21,6 +21,7 @@ import org.junit.Test;
import teetime.framework.pipe.CommittablePipeFactory;
import teetime.framework.pipe.IPipeFactory;
+import teetime.framework.pipe.OrderedGrowableArrayPipeFactory;
import teetime.framework.pipe.SingleElementPipeFactory;
import teetime.util.ConstructorClosure;
import teetime.util.TimestampObject;
@@ -58,6 +59,12 @@ public class MethodCallThoughputTimestampAnalysis9Test extends PerformanceTest {
testWithManyObjects(pipeFactory);
}
+ @Test
+ public void testOrderedGrowableArrayPipes() throws Exception {
+ IPipeFactory pipeFactory = new OrderedGrowableArrayPipeFactory();
+ testWithManyObjects(pipeFactory);
+ }
+
private void testWithManyObjects(final IPipeFactory pipeFactory) {
System.out.println("Testing teetime (mc) with NUM_OBJECTS_TO_CREATE=" + NUM_OBJECTS_TO_CREATE + ", NUM_NOOP_FILTERS="
+ NUM_NOOP_FILTERS + "...");
diff --git a/src/performancetest/java/teetime/examples/experiment14/AbstractPerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment14/AbstractPerformanceCheck.java
deleted file mode 100644
index b3dfc404714d86910880d8556c2e1596dc670b90..0000000000000000000000000000000000000000
--- a/src/performancetest/java/teetime/examples/experiment14/AbstractPerformanceCheck.java
+++ /dev/null
@@ -1,22 +0,0 @@
-package teetime.examples.experiment14;
-
-import teetime.examples.experiment01.MethodCallThoughputTimestampAnalysis1Test;
-import util.test.MeasurementRepository;
-import util.test.PerformanceResult;
-import util.test.PerformanceTest;
-import util.test.AbstractProfiledPerformanceAssertion;
-
-abstract class AbstractPerformanceCheck extends AbstractProfiledPerformanceAssertion {
-
- protected PerformanceResult test01;
- protected PerformanceResult test14;
-
- @Override
- public void check() {
- String testMethodIdentifier = MeasurementRepository.buildTestMethodIdentifier(MethodCallThoughputTimestampAnalysis1Test.class, "testWithManyObjects");
- test01 = PerformanceTest.measurementRepository.performanceResults.get(testMethodIdentifier);
- testMethodIdentifier = MeasurementRepository.buildTestMethodIdentifier(MethodCallThoughputTimestampAnalysis14Test.class, "testWithManyObjects");
- test14 = PerformanceTest.measurementRepository.performanceResults.get(testMethodIdentifier);
- }
-
-}
diff --git a/src/performancetest/java/teetime/examples/experiment14/ChwHomePerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment14/ChwHomePerformanceCheck.java
deleted file mode 100644
index 4c89ea59b83368762ed1806ba5c80d6ea171f3b7..0000000000000000000000000000000000000000
--- a/src/performancetest/java/teetime/examples/experiment14/ChwHomePerformanceCheck.java
+++ /dev/null
@@ -1,40 +0,0 @@
-package teetime.examples.experiment14;
-
-import static org.junit.Assert.assertEquals;
-import teetime.examples.HostName;
-
-class ChwHomePerformanceCheck extends AbstractPerformanceCheck {
-
- @Override
- public String getCorrespondingPerformanceProfile() {
- return HostName.CHW_HOME.toString();
- }
-
- @Override
- public void check() {
- super.check();
-
- double medianSpeedup = (double) test14.quantiles.get(0.5) / test01.quantiles.get(0.5);
-
- System.out.println("medianSpeedup (14): " + medianSpeedup);
-
- // until 25.06.2014 (incl.)
- // assertEquals(60, (double) test14.quantiles.get(0.5) / test1.quantiles.get(0.5), 5.1);
- // since 26.06.2014 (incl.)
- // assertEquals(76, medianSpeedup, 5.1); // +16
- // since 04.07.2014 (incl.)
- // assertEquals(86, medianSpeedup, 5.1); // +16
- // since 11.08.2014 (incl.)
- // assertEquals(103, medianSpeedup, 5.1); // +17
- // since 31.08.2014 (incl.)
- // assertEquals(62, medianSpeedup, 2.1); // -41
- // since 04.11.2014 (incl.)
- // assertEquals(84, medianSpeedup, 2.1); // +22
- // since 05.12.2014 (incl.)
- // assertEquals(75, medianSpeedup, 2.1); // -9
- // since 13.12.2014 (incl.)
- // assertEquals(44, medianSpeedup, 2.1); // -31
- // since 28.12.2014 (incl.)
- assertEquals(46, medianSpeedup, 2.1); // +2
- }
-}
diff --git a/src/performancetest/java/teetime/examples/experiment14/ChwWorkPerformanceCheck.java b/src/performancetest/java/teetime/examples/experiment14/ChwWorkPerformanceCheck.java
deleted file mode 100644
index b2d4deed66518910d7475f1dea702e425c491dbd..0000000000000000000000000000000000000000
--- a/src/performancetest/java/teetime/examples/experiment14/ChwWorkPerformanceCheck.java
+++ /dev/null
@@ -1,33 +0,0 @@
-package teetime.examples.experiment14;
-
-import static org.junit.Assert.assertEquals;
-
-class ChwWorkPerformanceCheck extends AbstractPerformanceCheck {
-
- @Override
- public String getCorrespondingPerformanceProfile() {
- return "ChwWork";
- }
-
- @Override
- public void check() {
- super.check();
-
- double medianSpeedup = (double) test14.quantiles.get(0.5) / test01.quantiles.get(0.5);
-
- System.out.println("medianSpeedup (14): " + medianSpeedup);
-
- // until 25.06.2014 (incl.)
- // assertEquals(60, (double) test14.quantiles.get(0.5) / test1.quantiles.get(0.5), 5.1);
- // since 26.06.2014 (incl.)
- // assertEquals(76, medianSpeedup, 5.1); // +16
- // since 04.07.2014 (incl.)
- // assertEquals(86, medianSpeedup, 5.1); // +16
- // since 27.08.2014 (incl.)
- // assertEquals(102, medianSpeedup, 5.1); // +16
- // since 14.10.2014 (incl.)
- // assertEquals(81, medianSpeedup, 5.1); // -21
- // since 19.12.2014 (incl.)
- assertEquals(56, medianSpeedup, 5.1); // -25
- }
-}
diff --git a/src/performancetest/java/teetime/examples/experiment14/MethodCallThoughputTimestampAnalysis14Test.java b/src/performancetest/java/teetime/examples/experiment14/MethodCallThoughputTimestampAnalysis14Test.java
deleted file mode 100644
index 75e7df4e331481e06dee39bddc311e2532488fb8..0000000000000000000000000000000000000000
--- a/src/performancetest/java/teetime/examples/experiment14/MethodCallThoughputTimestampAnalysis14Test.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/***************************************************************************
- * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
- *
- * Licensed 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 teetime.examples.experiment14;
-
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
-import teetime.util.ConstructorClosure;
-import teetime.util.TimestampObject;
-import util.test.PerformanceTest;
-import util.test.AbstractProfiledPerformanceAssertion;
-
-/**
- * @author Christian Wulf
- *
- * @since 1.10
- */
-public class MethodCallThoughputTimestampAnalysis14Test extends PerformanceTest {
-
- @BeforeClass
- public static void beforeClass() {
- PERFORMANCE_CHECK_PROFILE_REPOSITORY.register(MethodCallThoughputTimestampAnalysis14Test.class, new ChwWorkPerformanceCheck());
- PERFORMANCE_CHECK_PROFILE_REPOSITORY.register(MethodCallThoughputTimestampAnalysis14Test.class, new ChwHomePerformanceCheck());
- };
-
- @AfterClass
- public static void afterClass() {
- AbstractProfiledPerformanceAssertion performanceCheckProfile = PERFORMANCE_CHECK_PROFILE_REPOSITORY.get(MethodCallThoughputTimestampAnalysis14Test.class);
- performanceCheckProfile.check();
- };
-
- @Test
- public void testWithManyObjects() {
- System.out.println("Testing teetime (mc) with NUM_OBJECTS_TO_CREATE=" + NUM_OBJECTS_TO_CREATE + ", NUM_NOOP_FILTERS="
- + NUM_NOOP_FILTERS + "...");
-
- final MethodCallThroughputAnalysis14 analysis = new MethodCallThroughputAnalysis14();
- analysis.setNumNoopFilters(NUM_NOOP_FILTERS);
- analysis.setTimestampObjects(this.timestampObjects);
- analysis.setInput(NUM_OBJECTS_TO_CREATE, new ConstructorClosure<TimestampObject>() {
- @Override
- public TimestampObject create() {
- return new TimestampObject();
- }
- });
- analysis.init();
-
- this.stopWatch.start();
- try {
- analysis.start();
- } finally {
- this.stopWatch.end();
- }
- }
-
-}
diff --git a/src/performancetest/java/teetime/examples/experiment14/MethodCallThroughputAnalysis14.java b/src/performancetest/java/teetime/examples/experiment14/MethodCallThroughputAnalysis14.java
deleted file mode 100644
index 2f871761f6fbe15881927fdf22359e3ef25fc0a6..0000000000000000000000000000000000000000
--- a/src/performancetest/java/teetime/examples/experiment14/MethodCallThroughputAnalysis14.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/***************************************************************************
- * Copyright 2014 Kieker Project (http://kieker-monitoring.net)
- *
- * Licensed 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 teetime.examples.experiment14;
-
-import java.util.List;
-
-import teetime.framework.Stage;
-import teetime.framework.OldHeadPipeline;
-import teetime.framework.RunnableProducerStage;
-import teetime.framework.pipe.IPipeFactory;
-import teetime.framework.pipe.PipeFactoryRegistry;
-import teetime.framework.pipe.PipeFactoryRegistry.PipeOrdering;
-import teetime.framework.pipe.PipeFactoryRegistry.ThreadCommunication;
-import teetime.stage.CollectorSink;
-import teetime.stage.NoopFilter;
-import teetime.stage.ObjectProducer;
-import teetime.stage.StartTimestampFilter;
-import teetime.stage.StopTimestampFilter;
-import teetime.util.ConstructorClosure;
-import teetime.util.TimestampObject;
-
-/**
- * @author Christian Wulf
- *
- * @since 1.10
- */
-public class MethodCallThroughputAnalysis14 {
-
- private long numInputObjects;
- private ConstructorClosure<TimestampObject> inputObjectCreator;
- private int numNoopFilters;
- private List<TimestampObject> timestampObjects;
- private Runnable runnable;
- private final PipeFactoryRegistry pipeFactory = PipeFactoryRegistry.INSTANCE;
-
- public void init() {
- Stage pipeline = this.buildPipeline();
- this.runnable = new RunnableProducerStage(pipeline);
- }
-
- /**
- * @param numNoopFilters
- * @return
- * @since 1.10
- */
- private OldHeadPipeline<ObjectProducer<TimestampObject>, CollectorSink<TimestampObject>> buildPipeline() {
- @SuppressWarnings("unchecked")
- final NoopFilter<TimestampObject>[] noopFilters = new NoopFilter[this.numNoopFilters];
- // create stages
- final ObjectProducer<TimestampObject> objectProducer = new ObjectProducer<TimestampObject>(this.numInputObjects, this.inputObjectCreator);
- final StartTimestampFilter startTimestampFilter = new StartTimestampFilter();
- for (int i = 0; i < noopFilters.length; i++) {
- noopFilters[i] = new NoopFilter<TimestampObject>();
- }
- final StopTimestampFilter stopTimestampFilter = new StopTimestampFilter();
- final CollectorSink<TimestampObject> collectorSink = new CollectorSink<TimestampObject>(this.timestampObjects);
-
- final OldHeadPipeline<ObjectProducer<TimestampObject>, CollectorSink<TimestampObject>> pipeline = new OldHeadPipeline<ObjectProducer<TimestampObject>, CollectorSink<TimestampObject>>();
- pipeline.setFirstStage(objectProducer);
- pipeline.setLastStage(collectorSink);
-
- IPipeFactory factory = this.pipeFactory.getPipeFactory(ThreadCommunication.INTRA, PipeOrdering.QUEUE_BASED, true);
-
- factory.create(objectProducer.getOutputPort(), startTimestampFilter.getInputPort());
- factory.create(startTimestampFilter.getOutputPort(), noopFilters[0].getInputPort());
- for (int i = 0; i < noopFilters.length - 1; i++) {
- factory.create(noopFilters[i].getOutputPort(), noopFilters[i + 1].getInputPort());
- }
- factory.create(noopFilters[noopFilters.length - 1].getOutputPort(), stopTimestampFilter.getInputPort());
- factory.create(stopTimestampFilter.getOutputPort(), collectorSink.getInputPort());
-
- return pipeline;
- }
-
- public void start() {
- this.runnable.run();
- }
-
- public void setInput(final int numInputObjects, final ConstructorClosure<TimestampObject> inputObjectCreator) {
- this.numInputObjects = numInputObjects;
- this.inputObjectCreator = inputObjectCreator;
- }
-
- public int getNumNoopFilters() {
- return this.numNoopFilters;
- }
-
- public void setNumNoopFilters(final int numNoopFilters) {
- this.numNoopFilters = numNoopFilters;
- }
-
- public List<TimestampObject> getTimestampObjects() {
- return this.timestampObjects;
- }
-
- public void setTimestampObjects(final List<TimestampObject> timestampObjects) {
- this.timestampObjects = timestampObjects;
- }
-}