Newer
Older
import java.util.Collection;
import java.util.LinkedList;
import java.util.NavigableMap;
import java.util.TreeMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import teetime.stage.trace.traceReconstruction.EventBasedTrace;
import teetime.stage.trace.traceReconstruction.EventBasedTraceFactory;
import teetime.stage.trace.traceReduction.EventBasedTraceComperator;
import teetime.stage.trace.traceReduction.TraceAggregationBuffer;
import teetime.util.concurrent.hashmap.ConcurrentHashMapWithDefault;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import experiment.fse15.CounterActor;
import experiment.fse15.StartMessage;
import experiment.fse15.tcpreader.TcpReaderActor;
import experiment.fse15.tcpreconstruction.TcpTraceReconstructionActor;
public class TraceReductionAnalysisConfiguration {
private static final Logger LOGGER = LoggerFactory.getLogger(TraceReductionAnalysisConfiguration.class);
private final ConcurrentHashMapWithDefault<Long, EventBasedTrace> traceId2trace;
private final NavigableMap<EventBasedTrace, TraceAggregationBuffer> trace2Buffer;
private ActorRef tcpReaderActor;
private final Collection<Object> validTraces = new LinkedList<>();
public TraceReductionAnalysisConfiguration() {
traceId2trace = new ConcurrentHashMapWithDefault<Long, EventBasedTrace>(EventBasedTraceFactory.INSTANCE);
trace2Buffer = new TreeMap<EventBasedTrace, TraceAggregationBuffer>(new EventBasedTraceComperator());
init();
}
private void init() {
ActorSystem system = ActorSystem.create(TraceReductionAnalysisConfiguration.class.getSimpleName());
// create watch actor for termination
Props watchProps = Props.create(WatchActor.class);
final ActorRef watchActor = system.actorOf(watchProps);
Props counterProps = Props.create(CounterActor.class, watchActor);
Props tcpTraceReductionProps = Props.create(TcpTraceReductionActor.class, trace2Buffer, counterProps, watchActor);
Props tcpTraceReconstructionProps = Props.create(TcpTraceReconstructionActor.class, traceId2trace, tcpTraceReductionProps, Props.empty(), watchActor);
Props tcpReaderProps = Props.create(TcpReaderActor.class, tcpTraceReconstructionProps, watchActor);
// create actors
final ActorRef tcpReaderActor = system.actorOf(tcpReaderProps, "TcpReaderActor");
// final ActorRef tcpTraceReconstructonActor = system.actorOf(tcpTraceReconstructionProps, "TcpTraceReconstructionActor");
// final ActorRef tcpTraceReductionActor = system.actorOf(tcpTraceReductionProps, "TcpTraceReductionActor");
// final ActorRef validTraceReceiver = system.actorOf(collectorValidProps, "CollectorValid");
// final ActorRef invalidTraceReceiver = system.actorOf(collectorInvalidProps, "CollectorInvalid");
// final ActorRef tcpReaderActor = system.actorOf(Props.create(TcpReaderActor.class,
// new Creator<TcpReaderActor>() {
// @Override
// public TcpReaderActor create() throws Exception {
// return new TcpReaderActor(tcpTraceReconstructonActor);
// }
// }));
// tcpReader = TypedActor.get(system).typedActorOf(new TypedProps<TcpReaderActor>(ITcpReader.class, new
// Creator<TcpReaderActor>() {
// @Override
// public TcpReaderActor create() throws Exception {
// return new TcpReaderActor(tcpTraceReconstructonActor);
// }
// }));
this.tcpReaderActor = tcpReaderActor;
LOGGER.info("Configuration initialized.");
}
public void start() throws Exception {
LOGGER.info("Starting analysis...");
// tcpReader.onStarting();
// tcpReader.execute();
// tcpReader.onTerminating();
tcpReaderActor.tell(new StartMessage(), ActorRef.noSender());
LOGGER.info("Analysis started.");
}
public Collection<Object> getValidTraces() {
return validTraces;
}
}