package org.vanilladb.comm.protocols.p2pcounting;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.vanilladb.comm.process.ProcessList;
import org.vanilladb.comm.protocols.events.ProcessListInit;

import net.sf.appia.core.AppiaEventException;
import net.sf.appia.core.Direction;
import net.sf.appia.core.Event;
import net.sf.appia.core.Layer;
import net.sf.appia.core.Session;
import net.sf.appia.core.events.SendableEvent;

public class P2pCountingSession extends Session {
	
	private static final long REPORT_PERIOD = 5000;

	private ProcessList processList;
	private Map<SocketAddress, AtomicInteger> froms = new ConcurrentHashMap<SocketAddress, AtomicInteger>();
	private Map<SocketAddress, AtomicInteger> tos = new ConcurrentHashMap<SocketAddress, AtomicInteger>();
	
	P2pCountingSession(Layer layer) {
		super(layer);
		
		new Thread(new Runnable() {

			@Override
			public void run() {
				long startTime = System.currentTimeMillis();
				long lastTime = startTime;
				
				while (true) {
					long currentTime = System.currentTimeMillis();
					
					if (currentTime - lastTime > REPORT_PERIOD) {
						lastTime = currentTime;
						
						double t = (currentTime - startTime) / 1000;
						StringBuilder sb = new StringBuilder();
						
						sb.append("===================================\n");
						sb.append(String.format("At %.2f:\n", t));
						
						for (int i = 0; i < processList.getSize(); i++) {
							SocketAddress addr = processList.getProcess(i).getAddress();
							sb.append(String.format("From server %d: %d\n", i, froms.get(addr).getAndSet(0)));
							sb.append(String.format("To server %d: %d\n", i, tos.get(addr).getAndSet(0)));
						}
						
						sb.append("===================================\n");
						System.out.println(sb.toString());
					}
					
					try {
						Thread.sleep(REPORT_PERIOD / 10);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
			
		}).start();
	}
	
	@Override
	public void handle(Event event) {
		if (event instanceof ProcessListInit)
			handleProcessListInit((ProcessListInit) event);
		else if (event instanceof SendableEvent)
			handleSendableEvent((SendableEvent) event);
	}
	
	private void handleProcessListInit(ProcessListInit event) {
		// Save the list
		this.processList = event.copyProcessList();
		
		// Initializes the counters
		for (int i = 0; i < processList.getSize(); i++) {
			froms.put(processList.getProcess(i).getAddress(), new AtomicInteger());
			tos.put(processList.getProcess(i).getAddress(), new AtomicInteger());
		}
		
		// Let the event continue
		try {
			event.go();
		} catch (AppiaEventException e) {
			e.printStackTrace();
		}
	}
	
	private void handleSendableEvent(SendableEvent event) {
		if (event.getDir() == Direction.UP) {
			SocketAddress from = (SocketAddress) event.source;
			AtomicInteger count = froms.getOrDefault(from, new AtomicInteger());
			count.incrementAndGet();
		} else {
			SocketAddress to = (SocketAddress) event.dest;
			AtomicInteger count = tos.getOrDefault(to, new AtomicInteger());
			count.incrementAndGet();
		}
		
		// Let the event continue
		try {
			event.go();
		} catch (AppiaEventException e) {
			e.printStackTrace();
		}
	}
}

@@ -877,16 +877,108 @@
 ger%3E();%0A
+%09private Map%3CString, AtomicInteger%3E types = new ConcurrentHashMap%3CString, AtomicInteger%3E();%0A
 %09%0A%09P2pCo
@@ -1864,24 +1864,166 @@
 ));%0A%09%09%09%09%09%09%7D%0A
+%09%09%09%09%09%09for (String name : types.keySet()) %7B%0A%09%09%09%09%09%09%09sb.append(String.format(%22Count for %25s: %25d%5Cn%22, name, types.get(name).getAndSet(0)));%0A%09%09%09%09%09%09%7D%0A
 %09%09%09%09%09%09%0A%09%09%09%09%09
@@ -3413,32 +3413,261 @@
 ndGet();%0A%09%09%7D%0A%09%09%0A
+%09%09// Record the type%0A%09%09String name = event.getClass().getSimpleName();%0A%09%09AtomicInteger count = types.get(name);%0A%09%09if (count == null) %7B%0A%09%09%09count = new AtomicInteger(0);%0A%09%09%09types.put(name, count);%0A%09%09%7D%0A%09%09count.incrementAndGet();%0A%09%09%0A
 %09%09// Let the eve

Add message type counting