src/jaceP2P/MsgQueue.java

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