package node.company;

import java.util.*;

import node.data.RepositoryException;
import transaction.*;
import transaction.Lock.Token;
import communication.CommunicationException;
import communication.messages.DataTransferMessage;
import communication.messages.DataTransferReplyMessage;
import communication.protocols.*;
import communication.unicast.UnicastSocketClient;
import data.request.DataTransfer;
import data.trainnetwork.BookableSection;
import data.trainnetwork.Seat;

public class DistributedRepositoryMaster extends DistributedRepository {

	// Time between backups (in seconds)
	private final int TIME_BETWEEN_BACKUPS = 100;
	
	public DistributedRepositoryMaster() throws RepositoryException {
		super();
		scheduledBackup();
	}

	@Override
	protected Set<Protocol> getProtocols() {
		Set<Protocol> protocols = new HashSet<Protocol>();
		
		// Accept and handle `Hello` requests from other nodes
		protocols.add(new HelloProtocol(nodes));
		
		// Accept and handle distributed transactions
		protocols.add(new TransactionExecutionProtocol<String, Vault<BookableSection>, Set<Seat>>(transactions, communicationLock));
		protocols.add(new TransactionExecutionReplyProtocol<String, Vault<BookableSection>, Set<Seat>>(transactionCoordinators));
		protocols.add(new TransactionCommitProtocol<String, Vault<BookableSection>, Set<Seat>>(transactions, communicationLock));
		protocols.add(new TransactionCommitReplyProtocol<String, Vault<BookableSection>, Set<Seat>>(transactionCoordinators));
		
		return protocols;
	}
	
	private void scheduledBackup() {
		while (true) {
			try {
				sleep(1000 * TIME_BETWEEN_BACKUPS);
			} catch (InterruptedException e) {
				System.err.println("Failed to wait for backup: " + e.getMessage());
			}
			
			backup();
		}
	}
	
	// This will discard data if cannot send it, it is assumed for now that the node will not fail
	private void backup() {
		Collection<Vault<BookableSection>> toGet = null;
		
		Token sectionsToken = sections.readLock();
		try {
			 toGet = sections.getReadable(sectionsToken).values();
		} catch (LockException e) {
			// Failed to acquire read lock, oh well... will try next time...
			System.err.println("DistributedRepositoryMaster|backup: " + e.getMessage());
			return;
		} finally {
			sections.writeUnlock(sectionsToken);
		}
		
		Set<BookableSection> toSend = new HashSet<BookableSection>();
		while (toGet != null && toGet.size() > 0) {
			Collection<Vault<BookableSection>> failed = new HashSet<Vault<BookableSection>>();
			
			Iterator<Vault<BookableSection>> iterator = toGet.iterator();
			while (iterator.hasNext()) {
				Vault<BookableSection> vault = iterator.next();
				Token token = vault.readLock();
				try {
					toSend.add(vault.getReadable(token));
				} catch (LockException e) {
					failed.add(vault);
				} finally {
					vault.readUnlock(token);
				}
			}
			
			toGet = failed;
		}
		
		DataTransfer<BookableSection> transfer = new DataTransfer<BookableSection>(toSend);
		DataTransferMessage<BookableSection> message = new DataTransferMessage<BookableSection>(transfer, "BookableSection");
		
		Token t = communicationLock.writeLock();
		try {
			DataTransferReplyMessage reply = (DataTransferReplyMessage) 
					UnicastSocketClient.sendOneMessage(DATA_STORE_LOCATION, DATA_STORE_PORT, message, true);
			
			if (!message.getTransferId().equals((String) reply.getContents())) {
				// Something is very wrong on the other side if this happens
				new DataLoadException("Mismatching data transfer IDs");
			}
		} catch (CommunicationException e) {
			// Problems while talking to DataStore
			System.err.println("DistributedRepositoryMaster|backup: " + e.getMessage());
		} finally {
			communicationLock.writeUnlock(t);
		}
	}

	public void test() {
		try {
			sleep(5000);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		
		TransactionContent<String, Vault<BookableSection>, Set<Seat>> c = TransactionContentGenerator.getTestContent();
		
		TransactionCoordinator<String, Vault<BookableSection>, Set<Seat>> tc
			= new TransactionCoordinator<String, Vault<BookableSection>, Set<Seat>>(c, sections, nodes, communicationLock);
		
		transactionCoordinators.put(tc.getTransactionId(), tc);
		
		tc.start();
		
		while (true) {
			System.out.println(nodes);
			try {
				sleep(2000);
			} catch (InterruptedException e) {
				// Just loop around, no biggie
			}
		}
	}
	
	
	public static void main(String[] args) {
		DistributedRepositoryMaster r;
		try {
			if (args.length < 1 || !(args[0] instanceof String)) {
				throw new RepositoryException("Arg1 required to be the master node's location of the static data cluster");
			}
			
			DistributedRepositoryMaster.DATA_STORE_LOCATION = args[0];
			r = new DistributedRepositoryMaster();
			r.start();
			r.test();
		} catch (RepositoryException e) {
			e.printStackTrace();
		}
	}
}

@@ -715,29 +715,8 @@
 ();%0A
-%09%09scheduledBackup();%0A
 %09%7D%0A%0A
@@ -3705,670 +3705,8 @@
 %09%7D%0A%0A
-%09public void test() %7B%0A%09%09try %7B%0A%09%09%09sleep(5000);%0A%09%09%7D catch (InterruptedException e1) %7B%0A%09%09%09e1.printStackTrace();%0A%09%09%7D%0A%09%09%0A%09%09TransactionContent%3CString, Vault%3CBookableSection%3E, Set%3CSeat%3E%3E c = TransactionContentGenerator.getTestContent();%0A%09%09%0A%09%09TransactionCoordinator%3CString, Vault%3CBookableSection%3E, Set%3CSeat%3E%3E tc%0A%09%09%09= new TransactionCoordinator%3CString, Vault%3CBookableSection%3E, Set%3CSeat%3E%3E(c, sections, nodes, communicationLock);%0A%09%09%0A%09%09transactionCoordinators.put(tc.getTransactionId(), tc);%0A%09%09%0A%09%09tc.start();%0A%09%09%0A%09%09while (true) %7B%0A%09%09%09System.out.println(nodes);%0A%09%09%09try %7B%0A%09%09%09%09sleep(2000);%0A%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09// Just loop around, no biggie%0A%09%09%09%7D%0A%09%09%7D%0A%09%7D%0A%09%0A%09%0A
 %09pub
@@ -3743,16 +3743,16 @@
 args) %7B%0A
+
 %09%09Distri
@@ -4090,12 +4090,23 @@
 %09%09r.
-test
+scheduledBackup
 ();%0A

Unblock DistributedRepositoryMaster