src/jaceP2P/MsgQueue.java

   1 package jaceP2P;
   2
   3 import java.util.Vector;
   4
   5 public class MsgQueue {
   6
   7         public static MsgQueue Instance;
   8         // attributes
   9         private Vector<Message> liste;
  10
  11         // constructors
  12         public MsgQueue() {
  13                 liste = new Vector<Message>();
  14         }
  15
  16         public synchronized static MsgQueue Instance() {
  17                 if (Instance == null) {
  18                         Instance = new MsgQueue();
  19                 }
  20                 return Instance;
  21         }
  22
  23         // retourne l'index d'un Message de meme tag
  24         // retourne -1 si un tel Message existe pas
  25         private synchronized int exist(int tag, String appli, int time) {
  26                 int existe = -1;
  27                 int index = 0;
  28                 while ((existe == -1) && (index < liste.size())) {
  29                         if ((tag == ((Message) liste.get(index)).getTag())
  30                                         && (appli.equals(((Message) liste.get(index)).getSender()
  31                                                         .getAppliName()))
  32                                         && (time == ((Message) liste.get(index)).getTimeStep())
  33
  34                         ) {
  35                                 existe = index;
  36                         } else
  37                                 index++;
  38                 }
  39                 return existe;
  40         }
  41
  42         // retourne l'index d'un Message de meme tag ET meme sender ke "msg"
  43         // retourne -1 si un tel Message existe pas
  44         private synchronized int exist(int sender, int tag, String appli, int time) {
  45                 int existe = -1;
  46                 int index = 0;
  47                 while ((existe == -1) && (index < liste.size())) {
  48                         if ((tag == ((Message) liste.get(index)).getTag())
  49                                         && (sender == ((Message) liste.get(index)).getSender()
  50                                                         .getRank())
  51                                         && (appli.equals(((Message) liste.get(index)).getSender()
  52                                                         .getAppliName()))
  53                                         && (time == ((Message) liste.get(index)).getTimeStep())
  54
  55                         ) {
  56                                 existe = index;
  57                         } else
  58                                 index++;
  59                 }
  60                 return existe;
  61         }
  62
  63         // retourne l'index d'un Message de meme tag ET meme sender ke "msg"
  64         // retourne -1 si un tel Message existe pas
  65         private synchronized int exist(Message msg) {
  66                 int existe = -1;
  67                 int index = 0;
  68                 while ((existe == -1) && (index < liste.size())) {
  69                         if ((msg.getTag() == ((Message) liste.get(index)).getTag()) // meme
  70                                                                                                                                                 // tag
  71                                         && (msg.getSender().getRank() == ((Message) liste
  72                                                         .get(index)).getSender().getRank()) // meme tache
  73                                                                                                                                 // envoyeur
  74                                         && ((msg.getSender().getAppliName()).equals(JaceSession
  75                                                         .Instance().getTaskObject().getId().getAppliName())) // meme
  76                                                                                                                                                                         // appli
  77                                         && (msg.getTimeStep() == JaceSession.Instance()
  78                                                         .getTaskObject().getTimeStep())) { // meme timeStep
  79                                 existe = index;
  80                         } else
  81                                 index++;
  82                 }
  83                 return existe;
  84         }
  85
  86         // methods
  87         public void purge() {
  88                 Instance = null;
  89                 liste.clear();
  90         }
  91
  92         // ajoute un msg ds file
  93         public synchronized void add(Message msg) {
  94                 int is = -1;
  95                 is = exist(msg);
  96
  97                 // si existe deja 1 Message de meme tag ET meme envoyeur, on l'ecrase
  98                 if (is != -1) {
  99                         // liste.remove(is);
 100                         liste.setElementAt(msg, is);
 101                 }
 102
 103                 // si existe pas de Message de meme tag ET meme envoyeur, on l'ajoute
 104                 else {
 105                         liste.add(msg);
 106                 }
 107
 108                 // ensuite on reveille le thread de calcul en attente eventuelle sur le
 109                 // MsgQueue
 110                 /*
 111                  * try { //notifyAll(); } catch(Exception e){};
 112                  */
 113         }
 114
 115         // retourne Message de tag "tag"
 116         public synchronized Message get(int id, int tag) {
 117                 int is = -1;
 118                 if (liste.isEmpty()) {
 119                         return null;
 120                 } else {
 121                         // System.out.println("chercher "+tag+"---"+id);
 122                         is = exist(id, tag, JaceSession.Instance().getTaskObject().getId()
 123                                         .getAppliName(), JaceSession.Instance().getTaskObject()
 124                                         .getTimeStep());
 125                         if (is != -1) {
 126                                 // System.out.println("exitse ds la FA");
 127                                 Message tmp = (Message) liste.get(is);
 128                                 liste.remove(is);
 129                                 // System.out.println(tmp);
 130                                 // System.out.println("MSGQUEUE : j'ai recup " + tmp +
 131                                 // " de tag " + tmp.getTag() + " de  " +
 132                                 // tmp.getSender().getHostIP());
 133                                 return tmp;
 134                         } else {
 135                                 // System.out.println("existe pas :(((( "+tag+"---"+id);
 136                                 return null;
 137                         }
 138                 }
 139         }
 140
 141         public synchronized Message get(int tag) {
 142                 int is = -1;
 143                 // System.out.println("//////////kk viens de chercher un message");
 144                 if (liste.isEmpty()) {
 145                         return null;
 146                 } else {
 147                         is = exist(tag, JaceSession.Instance().getTaskObject().getId()
 148                                         .getAppliName(), JaceSession.Instance().getTaskObject()
 149                                         .getTimeStep());
 150                         if (is != -1) {
 151                                 Message tmp = (Message) liste.get(is);
 152                                 liste.remove(is);
 153                                 // System.out.println("MSGQUEUE : j'ai recup " + tmp +
 154                                 // " de tag " + tmp.getTag() + " de  " +
 155                                 // tmp.getSender().getHostIP());
 156                                 return tmp;
 157                         } else
 158                                 return null;
 159                 }
 160         }
 161
 162         // les reception bloquantes
 163         /*
 164          * public synchronized Message getBl(int id) { int is = -1; //
 165          * System.out.println("Taille de la file: "+liste.size()); if
 166          * (liste.isEmpty()) { try { wait(); } catch(Exception e){};} while ((is =
 167          * exist
 168          * (id,JaceSession.Instance().getTaskObject().getId().getAppliName(),JaceSession
 169          * .Instance().getTaskObject().getTimeStep())) == -1){ try{ wait(); }
 170          * catch(Exception e){}; try { //notifyAll(); } catch(Exception e){}; }
 171          * Message tmp; tmp = (Message)liste.get(is); liste.remove(is);
 172          * //System.out.println("MSGQUEUE : j'ai recup " + tmp + " de tag " +
 173          * tmp.getTag() + " de  " + tmp.getSender().getHostIP()); //
 174          * System.out.println("Taille de la file: "+liste.size()); return tmp; }
 175          *
 176          * public synchronized Message getBlTag(int id) { int is = -1; //
 177          * System.out.println("Taille de la file: "+liste.size()); if
 178          * (liste.isEmpty()) { try { wait(); } catch(Exception e){}; }
 179          *
 180          * while ((is =
 181          * exist(id,JaceSession.Instance().getTaskObject().getId().getAppliName
 182          * (),JaceSession.Instance().getTaskObject().getTimeStep())) == -1) { try{
 183          * wait(); } catch(Exception e){}; try { //notifyAll(); } catch(Exception
 184          * e){}; } Message tmp; tmp = (Message)liste.get(is); liste.remove(is);
 185          * //System.out.println("MSGQUEUE : j'ai recup " + tmp + " de tag " +
 186          * tmp.getTag() + " de  " + tmp.getSender().getHostIP()); //
 187          * System.out.println("Taille de la file: "+liste.size()); return tmp; }
 188          *
 189          * public synchronized Message getBl(int id,int tag) { int is = -1;
 190          * //System.out.println("je cherche   "+id+"  "+tag);
 191          * //System.out.println("Taille de la file: "+liste.size()); if
 192          * (liste.isEmpty()) { try { wait(); }catch(Exception e){}; } while ((is =
 193          * exist
 194          * (id,tag,JaceSession.Instance().getTaskObject().getId().getAppliName()
 195          * ,JaceSession.Instance().getTaskObject().getTimeStep())) == -1) { try {
 196          * wait(); } catch(Exception e){}; try { //notifyAll(); } catch(Exception
 197          * e){}; } Message tmp; tmp = (Message)liste.get(is); liste.remove(is);
 198          * //System.out.println("MSGQUEUE : j'ai recup " + tmp + " de tag " +
 199          * tmp.getTag() + " de " + tmp.getSender().getHostIP()); return tmp; }
 200          */
 201
 202         public synchronized int getSize() {
 203                 return liste.size();
 204         }
 205 }