TraceReconstructionAnalysisConfiguration.java

package experiment.fse15.tcpreconstruction;

import java.util.Collection;
import java.util.LinkedList;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import teetime.stage.trace.traceReconstruction.EventBasedTrace;
import teetime.stage.trace.traceReconstruction.EventBasedTraceFactory;
import teetime.util.concurrent.hashmap.ConcurrentHashMapWithDefault;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import experiment.fse15.CollectorActor;
import experiment.fse15.CounterActor;
import experiment.fse15.StartMessage;
import experiment.fse15.WatchActor;
import experiment.fse15.tcpreader.TcpReaderActor;

public class TraceReconstructionAnalysisConfiguration {

	private static final Logger LOGGER = LoggerFactory.getLogger(TraceReconstructionAnalysisConfiguration.class);

	private final ConcurrentHashMapWithDefault<Long, EventBasedTrace> traceId2trace;

	private ActorRef tcpReaderActor;
	private final Collection<Object> validTraces = new LinkedList<>();

	public TraceReconstructionAnalysisConfiguration() {
		traceId2trace = new ConcurrentHashMapWithDefault<>(EventBasedTraceFactory.INSTANCE);
		init();
	}

	private void init() {
		ActorSystem system = ActorSystem.create(TraceReconstructionAnalysisConfiguration.class.getSimpleName());

		// create watch actor for termination
		Props watchProps = Props.create(WatchActor.class);
		final ActorRef watchActor = system.actorOf(watchProps);

		// specify actors
		// Props collectorValidProps = Props.create(Collector.class, validTraces); // do not use a collector since it
		// pollutes the heap
		Props collectorValidProps = Props.create(CounterActor.class, watchActor);
		Props collectorInvalidProps = Props.create(CollectorActor.class, watchActor);
		Props tcpTraceReconstructionProps = Props.create(TcpTraceReconstructionActor.class, traceId2trace,
				collectorValidProps, collectorInvalidProps, 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 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;
	}

}