1 phase comit protocol works

1 files changed, 41 insertions, 6 deletions

diff --git a/sources/protocols/implementations/OnePhaseCommitProtocol.java b/sources/protocols/implementations/OnePhaseCommitProtocol.java
index c8dd7f2..3e4ae52 100644
--- a/sources/protocols/implementations/OnePhaseCommitProtocol.java
+++ b/sources/protocols/implementations/OnePhaseCommitProtocol.java
@@ -4,6 +4,7 @@
  */
 package protocols.implementations;
 
+import java.util.ArrayList;
 import java.util.Vector;
 
 import protocols.VSAbstractProtocol;
@@ -15,7 +16,8 @@ import core.VSMessage;
 public class OnePhaseCommitProtocol extends VSAbstractProtocol {
     private static final long serialVersionUID = 1L;
 
-    /* Client variables, coordinator */
+    /* Server variables, coordinator */
+    private ArrayList<Integer> pids;
 
     /* Client variables */
     private boolean ackSent;
@@ -30,9 +32,9 @@ public class OnePhaseCommitProtocol extends VSAbstractProtocol {
         Vector<Integer> vec = new Vector<Integer>();
         vec.add(2);
         vec.add(3);
-        vec.add(4);
 
         initVector("pids", vec, "PIDs beteilitger Prozesse");
+        initLong("timeout", 5000, "Zeit bis erneuerter Anfrage", "ms");
     }
 
     /* (non-Javadoc)
@@ -45,29 +47,54 @@ public class OnePhaseCommitProtocol extends VSAbstractProtocol {
      * @see protocols.VSAbstractProtocol#onClientReset()
      */
     protected void onClientReset() {
+        pids.clear();
+        pids.addAll(getVector("pids"));
     }
 
     /* (non-Javadoc)
      * @see protocols.VSAbstractProtocol#onClientStart()
      */
     protected void onClientStart() {
-        int numProcesses = getInteger("numProcesses");
-        VSMessage message = new VSMessage();
-        message.setBoolean("wantAck", true);
-        sendMessage(message);
+        if (pids == null) {
+            pids = new ArrayList<Integer>();
+            pids.addAll(getVector("pids"));
+
+        }
+
+        if (pids.size() != 0) {
+            long timeout = getLong("timeout") + process.getTime();
+            scheduleAt(timeout); /* Will run onClientSchedule() at the specified local time */
+
+            VSMessage message = new VSMessage();
+            message.setBoolean("wantAck", true);
+            sendMessage(message);
+        }
     }
 
     /* (non-Javadoc)
      * @see protocols.VSAbstractProtocol#onClientRecv(core.VSMessage)
      */
     protected void onClientRecv(VSMessage recvMessage) {
+        if (pids.size() == 0)
+            return;
 
+        if (recvMessage.getBoolean("isAck")) {
+            Integer pid = recvMessage.getIntegerObj("pid");
+            if (pids.contains(pid))
+                pids.remove(pid);
+
+            logg("ACK von Prozess " + pid + " erhalten!");
+
+            if (pids.size() == 0)
+                logg("ACKs von allen beteiligten Prozessen erhalten!");
+        }
     }
 
     /* (non-Javadoc)
      * @see protocols.VSAbstractProtocol#onClientSchedule()
      */
     protected void onClientSchedule() {
+        onClientStart();
     }
 
     /* (non-Javadoc)
@@ -81,6 +108,14 @@ public class OnePhaseCommitProtocol extends VSAbstractProtocol {
      * @see protocols.VSAbstractProtocol#onServerRecv(core.VSMessage)
      */
     protected void onServerRecv(VSMessage recvMessage) {
+        if (ackSent)
+            return;
+
+        VSMessage message = new VSMessage();
+        message.setBoolean("isAck", true);
+        message.setInteger("pid", process.getProcessID());
+        sendMessage(message);
+        ackSent = true;
     }
 
     /* (non-Javadoc)