Merge branch 'dev-group3-supervisor-fix' into 'dev-group3'

Added tell supervisor action to make system init work

Fixes #16 by introducing a new tell supervisor action, that is sent additionally to the ActionJoinAnswer message.
If the answering node does not know the supervisor yet, it will postpone the "tell supervisor" action until it does know one.
That way, we are able to spawn large amounts of wallets

See merge request !4

src/main/java/fucoin/Main.java

@@ -5,7 +5,7 @@ import akka.actor.ActorSystem;
 import akka.actor.Props;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
-import fucoin.actions.join.ActionJoinAnswer;
+import fucoin.actions.join.ActionTellSupervisor;
 import fucoin.setup.NetworkInterfaceReader;
 import fucoin.supervisor.SuperVisorImpl;
 import fucoin.wallet.WalletImpl;
@@ -14,13 +14,11 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
-import static akka.dispatch.Futures.future;
-
 
 public class Main {
 
     private static int numberOfWallets = 2;
-    private static boolean createGUI = true;
+    private static boolean createGUI = false;
 
     private static ActorSystem cSystem;
 
@@ -69,7 +67,8 @@ public class Main {
             // first wallet does not have a neighbour, so it can't send a ActionJoin to anybody
             // instead we send directly an ActionJoinAnswer with the supervisor reference
             if (i == 0) {
-                actorRef.tell(new ActionJoinAnswer(future(() -> cSuperVisorActor, cSystem.dispatcher())), cSuperVisorActor);
+                //actorRef.tell(new ActionJoinAnswer(cSuperVisorActor), cSuperVisorActor);
+                actorRef.tell(new ActionTellSupervisor(cSuperVisorActor), cSuperVisorActor);
             }
 
             cActiveActors.add(actorRef);

src/main/java/fucoin/actions/join/ActionJoin.java

@@ -14,8 +14,14 @@ public class ActionJoin extends ClientAction {
     protected void onAction(ActorRef sender, ActorRef self, UntypedActorContext context, AbstractWallet node) {
 
         // send the joined node all known neighbours from node and a reference to the supervisor
-        ActionJoinAnswer aja = new ActionJoinAnswer(node.resolveSuperVisorActor());
+        ActionJoinAnswer aja = new ActionJoinAnswer();
         aja.someNeighbors.putAll(node.getKnownNeighbors());
         sender.tell(aja, self);
+
+        if (node.getRemoteSuperVisorActor() == null) {
+            node.deferSendOfSuperVisorActor(sender);
+        } else {
+            sender.tell(new ActionTellSupervisor(node.getRemoteSuperVisorActor()), self);
+        }
     }
 }

src/main/java/fucoin/actions/join/ActionJoinAnswer.java

@@ -24,11 +24,6 @@ import java.util.Map.Entry;
  */
 public class ActionJoinAnswer extends ClientAction {
     public final HashMap<String, ActorRef> someNeighbors = new HashMap<>();
-    public final Future<ActorRef> supervisor;
-
-    public ActionJoinAnswer(Future<ActorRef> supervisor) {
-        this.supervisor = supervisor;
-    }
 
     protected void onAction(ActorRef sender, ActorRef self,
                             UntypedActorContext context, AbstractWallet wallet) {
@@ -42,18 +37,6 @@ public class ActionJoinAnswer extends ClientAction {
         for (Entry<String, ActorRef> neighbor : someNeighbors.entrySet()) {
             neighbor.getValue().tell(new ActionSearchMyWallet(wallet.getName()), self);
         }
-
-        // register at the supervisor if the wallet just learned about it
-        if (wallet.getRemoteSuperVisorActor() == null) {
-            supervisor.onSuccess(new OnSuccess<ActorRef>() {
-
-                @Override
-                public void onSuccess(ActorRef result) throws Throwable {
-                    wallet.setRemoteSuperVisorActor(result);
-                }
-            }, context.system().dispatcher());
-            //wallet.setRemoteSuperVisorActor(supervisor);
-        }
     }
 
 }

src/main/java/fucoin/actions/join/ActionTellSupervisor.java

+package fucoin.actions.join;
+
+import akka.actor.ActorRef;
+import akka.actor.UntypedActorContext;
+import fucoin.actions.ClientAction;
+import fucoin.wallet.AbstractWallet;
+
+/**
+ * Tell the joining node the supervisor
+ */
+public class ActionTellSupervisor extends ClientAction {
+
+    public final ActorRef supervisor;
+
+    public ActionTellSupervisor(ActorRef supervisor) {
+        this.supervisor = supervisor;
+    }
+
+    @Override
+    protected void onAction(ActorRef sender, ActorRef self, UntypedActorContext context, AbstractWallet abstractNode) {
+        abstractNode.setRemoteSuperVisorActor(supervisor);
+    }
+}

src/main/java/fucoin/actions/join/ServerActionJoin.java

@@ -21,7 +21,7 @@ public class ServerActionJoin extends SuperVisorAction {
     @Override
     protected void onAction(ActorRef sender, ActorRef self,
                             UntypedActorContext context, SuperVisorImpl node) {
-        ActionJoinAnswer aja = new ActionJoinAnswer(node.resolveSelf());
+        ActionJoinAnswer aja = new ActionJoinAnswer(); // TODO: Might need added TellSupervisor
         aja.someNeighbors.putAll(node.getKnownNeighbors());
         sender.tell(aja, self);
         node.addKnownNeighbor(name, sender);

src/main/java/fucoin/gui/SuperVisorGuiControlImpl.java

@@ -3,6 +3,8 @@ package fucoin.gui;
 import fucoin.supervisor.SuperVisorImpl;
 
 import javax.swing.*;
+import javax.swing.event.TableModelEvent;
+import javax.swing.event.TableModelListener;
 import java.awt.*;
 import java.awt.event.ItemEvent;
 import java.awt.event.WindowAdapter;
@@ -31,6 +33,7 @@ public class SuperVisorGuiControlImpl implements SuperVisorGuiControl {
         //Init Amount Table and SuperVisorImpl
 
         JTable amountListView = new JTable(superVisor.getAmountTableModel());
+        superVisor.getAmountTableModel().addTableModelListener(e -> SwingUtilities.invokeLater(() -> frame.setTitle("Server (" + superVisor.getAmountTableModel().getRowCount() + " Wallets)")));
         contentPanel.add(new JScrollPane(amountListView));
 
         JPanel logPanel = new JPanel(new BorderLayout());

src/main/java/fucoin/supervisor/SuperVisorImpl.java

 package fucoin.supervisor;
 
-import akka.actor.ActorRef;
 import akka.actor.Props;
 import fucoin.actions.Action;
 import fucoin.actions.persist.ActionInvokeUpdate;
@@ -9,7 +8,6 @@ import fucoin.actions.transaction.SuperVisorAction;
 import fucoin.gui.SuperVisorGuiControl;
 import fucoin.AbstractNode;
 import fucoin.gui.TransactionLogger;
-import scala.concurrent.Future;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -84,10 +82,6 @@ public class SuperVisorImpl extends AbstractNode implements TransactionLogger{
         self().tell(new ActionUpdateQueue(), self());
     }
 
-    public Future<ActorRef> resolveSelf() {
-        return future(() -> self(), context().system().dispatcher());
-    }
-
     /**
      * filters the request for outdated and removes them
      *

src/main/java/fucoin/wallet/AbstractWallet.java

@@ -96,7 +96,7 @@ public abstract class AbstractWallet extends AbstractNode implements Serializabl
      */
     public abstract void setRemoteSuperVisorActor(ActorRef remoteSuperVisorActor);
 
-    public abstract Future<ActorRef> resolveSuperVisorActor();
+    public abstract void deferSendOfSuperVisorActor(ActorRef destinationWallet);
 
     /**
      * Sends amount FUCs to the wallet with the address adress

src/main/java/fucoin/wallet/WalletImpl.java

@@ -5,6 +5,7 @@ import akka.actor.Props;
 import fucoin.actions.ClientAction;
 import fucoin.actions.join.ActionJoin;
 import fucoin.actions.join.ActionJoinAnswer;
+import fucoin.actions.join.ActionTellSupervisor;
 import fucoin.actions.join.ServerActionJoin;
 import fucoin.actions.persist.ActionInvokeLeave;
 import fucoin.actions.persist.ActionInvokeRevive;
@@ -15,6 +16,8 @@ import scala.concurrent.Future;
 
 import static akka.dispatch.Futures.future;
 
+import java.util.concurrent.ConcurrentLinkedQueue;
+
 public class WalletImpl extends AbstractWallet {
 
     private ActorRef preKnownNeighbour;
@@ -22,6 +25,7 @@ public class WalletImpl extends AbstractWallet {
     private transient WalletGuiControl gui;
     private String preKnownNeighbourName;
     private boolean isActive;
+    private ConcurrentLinkedQueue<ActorRef> deferedSupervisorReceivers = new ConcurrentLinkedQueue<>();
 
     public WalletImpl(String name) {
         super(name);
@@ -85,11 +89,15 @@ public class WalletImpl extends AbstractWallet {
         if (preKnownNeighbour != null) {
             addKnownNeighbor(preKnownNeighbourName, preKnownNeighbour);
             preKnownNeighbour.tell(new ActionJoin(), getSelf());
-            ActionJoinAnswer aja = new ActionJoinAnswer(this.resolveSuperVisorActor());
+            ActionJoinAnswer aja = new ActionJoinAnswer();
             aja.someNeighbors.putAll(getKnownNeighbors());
             aja.someNeighbors.put(name, getSelf());
             preKnownNeighbour.tell(aja, getSelf());
 
+            if (this.getRemoteSuperVisorActor() != null) {
+                preKnownNeighbour.tell(new ActionTellSupervisor(this.getRemoteSuperVisorActor()), self());
+            }
+
         }
     }
 
@@ -158,6 +166,21 @@ public class WalletImpl extends AbstractWallet {
         if (this.remoteSuperVisorActor == null) {
             this.remoteSuperVisorActor = remoteSuperVisorActor;
             this.remoteSuperVisorActor.tell(new ServerActionJoin(getName()), getSelf());
+            this.tellDeferedSuperVisorReceivers(remoteSuperVisorActor);
+        }
+    }
+
+    @Override
+    public void deferSendOfSuperVisorActor(ActorRef destinationWallet) {
+        deferedSupervisorReceivers.add(destinationWallet);
+    }
+
+    protected void tellDeferedSuperVisorReceivers(ActorRef supervisor) {
+        while (deferedSupervisorReceivers.size() > 0) {
+            ActorRef receiver = deferedSupervisorReceivers.poll();
+            if (receiver != null) {
+                receiver.tell(new ActionTellSupervisor(supervisor), self());
+            }
         }
     }