package pt.up.fe.sdis.proj1.protocols.initiator;

import java.net.InetAddress;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import pt.up.fe.sdis.proj1.Chunk;
import pt.up.fe.sdis.proj1.messages.Message;
import pt.up.fe.sdis.proj1.protocols.AbstractProtocol;
import pt.up.fe.sdis.proj1.utils.Communicator;
import rx.Scheduler;
import rx.Scheduler.Inner;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

public class ChunkBackup extends AbstractProtocol {
    public ChunkBackup(final Communicator comm, final Chunk chunk) {
        super(comm.MC.Publisher);

        Schedulers.newThread().schedule(new Action1<Scheduler.Inner>() {

            @Override
            public void call(Inner arg0) {

                final Message msg = Message.makePutChunk(chunk);
                
                start(new Func1<Message, Boolean>() {
                    @Override
                    public Boolean call(Message arg0) {
                        return arg0.type == Message.Type.STORED
                                && arg0.getFileID().equals(chunk.fileID)
                                && arg0.getChunkNo().equals(chunk.chunkNo);
                    }
                });
                
                comm.MDB.Sender.Send(msg);
                
                Action1<Scheduler.Inner> act = new Action1<Scheduler.Inner>() {
                    int TimeInterval = 500;
                    int NumTimes = 0;
                    
                    @Override
                    public void call(Inner arg0) {
                        synchronized (_repliers) {
                            NumTimes++;
                            
                            int numRepliers = _repliers.size();
                            
                            if (numRepliers < chunk.replicationDeg) {
                                if (NumTimes == 5) {
                                	//TODO replace with some kind of warning
                                    finish();
                                } else {
                                    _repliers.clear();
                                    TimeInterval *= 2;
                                    comm.MDB.Sender.Send(msg);
                                    arg0.schedule(this, TimeInterval, TimeUnit.MILLISECONDS);
                                }
                            } else {
                                finish();
                            }
                        }
                    }
                };
                
                arg0.schedule(act, 500, TimeUnit.MILLISECONDS);
            }
        });
    }

    @Override
    protected void ProcessMessage(Message msg) {
        synchronized (_repliers) {
            _repliers.add(msg.Sender);
        }
    }
    
    private Set<InetAddress> _repliers = new HashSet<InetAddress>();
}

@@ -633,25 +633,24 @@
 isher);%0A
-%0A
         
 Schedule
@@ -645,148 +645,9 @@
     
-Schedulers.newThread().schedule(new Action1%3CScheduler.Inner%3E() %7B%0A%0A            @Override%0A            public void call(Inner arg0) %7B%0A%0A        
+%0A
     
@@ -703,36 +703,17 @@
 k);%0A
-                %0A   
+%0A
         
     
@@ -708,21 +708,16 @@
         
-     
 start(ne
@@ -754,32 +754,24 @@
             
-        
 @Override%0A  
@@ -776,32 +776,24 @@
             
-       
- 
 public Boole
@@ -812,32 +812,24 @@
 age arg0) %7B%0A
-        
             
@@ -892,32 +892,24 @@
             
- 
-       
 && arg0.getF
@@ -957,32 +957,24 @@
             
-   
-     
 && arg0.getC
@@ -1013,32 +1013,24 @@
             
-        
 %7D%0A          
@@ -1031,48 +1031,21 @@
     
- 
-       %7D);%0A                %0A   
+%7D);%0A%0A
         
     
@@ -1032,37 +1032,32 @@
    %7D);%0A%0A        
-     
 comm.MDB.Sender.
@@ -1071,33 +1071,9 @@
 g);%0A
-                %0A        
+%0A
     
@@ -1148,32 +1148,24 @@
             
-        
 int TimeInte
@@ -1184,32 +1184,24 @@
             
-   
-     
 int NumTimes
@@ -1210,37 +1210,9 @@
  0;%0A
-                    %0A        
+%0A
     
@@ -1237,32 +1237,24 @@
             
-      
-  
 public void 
@@ -1268,32 +1268,24 @@
 ner arg0) %7B%0A
-        
             
@@ -1335,24 +1335,16 @@
         
-        
 NumTimes
@@ -1351,45 +1351,9 @@
 ++;%0A
-                            %0A        
+%0A
     
@@ -1408,45 +1408,9 @@
 ();%0A
-                            %0A        
+%0A
     
@@ -1487,32 +1487,24 @@
             
-        
 if (NumTimes
@@ -1544,15 +1544,11 @@
     
-  
-  %09
 //
+ 
 TODO
@@ -1606,32 +1606,24 @@
             
-        
 finish();%0A  
@@ -1616,32 +1616,24 @@
   finish();%0A
-        
             
@@ -1677,32 +1677,24 @@
             
-        
 _repliers.cl
@@ -1696,32 +1696,24 @@
 rs.clear();%0A
-        
             
@@ -1775,24 +1775,16 @@
         
-    
-    
 comm.MDB
@@ -1802,24 +1802,16 @@
 d(msg);%0A
-        
         
@@ -1863,16 +1863,52 @@
 nterval,
+%0A                                   
  TimeUni
@@ -1944,37 +1944,26 @@
             
-        %7D%0A   
+%7D%0A
             
@@ -1962,37 +1962,32 @@
                 
-     
 %7D else %7B%0A       
@@ -1975,32 +1975,24 @@
    %7D else %7B%0A
-        
             
@@ -2033,34 +2033,18 @@
         
-       
- 
 %7D%0A
-        
         
@@ -2069,74 +2069,52 @@
     
-        %7D%0A                %7D;%0A                %0A                arg0
+%7D%0A        %7D;%0A%0A        Schedulers.newThread()
 .sch
@@ -2157,34 +2157,8 @@
 S);%0A
-            %7D%0A        %7D);%0A
     
@@ -2313,20 +2313,16 @@
 %7D%0A    %7D%0A
-    
 %0A    pri

Synchronize Chunk Backup Protocol.