Commit 95edd195 authored by Sören Henning's avatar Sören Henning

redesigned RecordDistributor

parent 0a5b65a1
package anomalydetection.kieker;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import teetime.framework.AbstractConsumerStage;
import teetime.framework.OutputPort;
public class RecordDistributor extends AbstractConsumerStage<MonitoringRecord> {
private final Map<RecordFilter, OutputPort<MonitoringRecord>> outputPorts = new HashMap<>();
@Override
protected void execute(final MonitoringRecord record) {
for (Entry<RecordFilter, OutputPort<MonitoringRecord>> entry : outputPorts.entrySet()) {
if (entry.getKey().test(record)) {
entry.getValue().send(record);
}
}
}
}
package anomalydetection.kieker;
import kieker.analysis.stage.tracediagnosis.Filter;
import teetime.framework.CompositeStage;
import teetime.framework.InputPort;
import teetime.framework.OutputPort;
import teetime.stage.basic.distributor.Distributor;
import teetime.stage.basic.distributor.strategy.CopyByReferenceStrategy;
public class RecordDistributorStage extends CompositeStage {
private final InputPort<MonitoringRecord> inputPort;
private final Distributor<MonitoringRecord> distributor;
public RecordDistributorStage() {
this.distributor = new Distributor<>(new CopyByReferenceStrategy());
this.inputPort = this.distributor.getInputPort();
}
public OutputPort<MonitoringRecord> getNewOutputPort(final RecordFilter filter) {
final Filter<MonitoringRecord> filterStage = new Filter<>(filter);
super.connectPorts(this.distributor.getNewOutputPort(), filterStage.getInputPort());
return filterStage.getOutputPort();
}
public InputPort<MonitoringRecord> getInputPort() {
return this.inputPort;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment