moved fields into logic classes

src/main/java/akka/io/net/TcpTraceReconstructionActor.java

@@ -28,14 +28,14 @@ public class TcpTraceReconstructionActor extends UntypedActor implements ISendTr
 	private long maxTraceTimeout = Long.MAX_VALUE;
 	private long maxEncounteredLoggingTimestamp = -1;
 
-	private final ConcurrentHashMapWithDefault<Long, EventBasedTrace> traceId2trace;
+	private TraceReconstructor reconstructor;
 	private final ActorRef validTraceReceiver;
 	private final ActorRef invalidTraceReceiver;
 
 	public TcpTraceReconstructionActor(final ConcurrentHashMapWithDefault<Long, EventBasedTrace> traceId2trace,
 			final Props validTraceReceiverProps, final Props invalidTraceReceiverProps, ActorRef watchActor) {
 		super();
-		this.traceId2trace = traceId2trace;
+		this.reconstructor = new TraceReconstructor(traceId2trace);
 		this.validTraceReceiver = context().actorOf(validTraceReceiverProps);
 		this.invalidTraceReceiver = context().actorOf(invalidTraceReceiverProps);
 
@@ -45,7 +45,7 @@ public class TcpTraceReconstructionActor extends UntypedActor implements ISendTr
 	@Override
 	public void onReceive(final Object message) throws Exception {
 		if (message instanceof IFlowRecord) {
-			TraceReconstructor.execute((IFlowRecord) message, traceId2trace, this);
+			reconstructor.execute((IFlowRecord) message, this);
 		} else if (message instanceof StopMessage) {
 			this.onTerminating();
 			validTraceReceiver.tell(message, getSelf());
@@ -64,8 +64,8 @@ public class TcpTraceReconstructionActor extends UntypedActor implements ISendTr
 	}
 
 	private void onTerminating() {
-		LOGGER.trace("traces left: " + traceId2trace.keySet());
-		TraceReconstructor.terminate(traceId2trace, this);
+		LOGGER.trace("traces left: " + reconstructor.getTraceId2trace().keySet());
+		reconstructor.terminate(this);
 	}
 
 	public void sendTraceBuffer(final EventBasedTrace traceBuffer) {

src/main/java/akka/io/net/TcpTraceReductionActor.java

 package akka.io.net;
 
-import java.util.Iterator;
-import java.util.Map.Entry;
 import java.util.NavigableMap;
 
 import org.slf4j.Logger;
@@ -21,14 +19,14 @@ public class TcpTraceReductionActor extends UntypedActor implements ISendTraceAg
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(TcpTraceReductionActor.class);
 
-	private final NavigableMap<EventBasedTrace, TraceAggregationBuffer> trace2buffer;
+	private final TraceReductor reductor;
 	private final ActorRef traceAggregationReceiver;
 
 	private long maxCollectionDurationInNs;
 
 	public TcpTraceReductionActor(final NavigableMap<EventBasedTrace, TraceAggregationBuffer> trace2buffer,
 			final Props traceAggregationReceiverProps, ActorRef watchActor) {
-		this.trace2buffer = trace2buffer;
+		this.reductor = new TraceReductor(trace2buffer);
 		// output "ports"
 		this.traceAggregationReceiver = context().actorOf(traceAggregationReceiverProps);
 
@@ -39,13 +37,13 @@ public class TcpTraceReductionActor extends UntypedActor implements ISendTraceAg
 	public void onReceive(final Object message) throws Exception {
 		if (message instanceof EventBasedTrace) {
 			EventBasedTrace eventBasedTrace = (EventBasedTrace) message;
-			TraceReductor.countSameTraces(eventBasedTrace, trace2buffer);
+			reductor.countSameTraces(eventBasedTrace);
 			
 			
 		} else if (message instanceof Long) {
 			Long timestampInNs = (Long) message;
 			if (timestampInNs != null) {
-				TraceReductor.processTimeoutQueue(timestampInNs, maxCollectionDurationInNs, trace2buffer, this);
+				reductor.processTimeoutQueue(timestampInNs, maxCollectionDurationInNs, this);
 			}
 		} else if (message instanceof StopMessage) {
 			this.onTerminating();
@@ -71,7 +69,7 @@ public class TcpTraceReductionActor extends UntypedActor implements ISendTraceAg
 
 	private void onTerminating() {
 		LOGGER.debug("Terminating...");
-		TraceReductor.terminate(trace2buffer, this);
+		reductor.terminate(this);
 	}
 
 	public long getMaxCollectionDuration() {