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

import java.util.Random;
import java.util.concurrent.TimeUnit;

import pt.up.fe.sdis.proj1.BackupSystem;
import pt.up.fe.sdis.proj1.messages.Message;
import pt.up.fe.sdis.proj1.protocols.AbstractProtocol;
import pt.up.fe.sdis.proj1.protocols.initiator.SpaceReclaiming;
import pt.up.fe.sdis.proj1.utils.Communicator;
import pt.up.fe.sdis.proj1.utils.MessageFilter;
import rx.Scheduler;
import rx.functions.Action1;
import rx.schedulers.Schedulers;

public class PeerChunkBackup extends AbstractProtocol {

    public PeerChunkBackup(BackupSystem bs) {
        super(bs.Comm.MDB.Publisher);
        _comm = bs.Comm;
        _bs = bs;

        start(new MessageFilter(Message.Type.PUTCHUNK));
    }

    @Override
    protected void ProcessMessage(final Message msg) {
        if(_bs.getAvailableSpace() < msg.getBody().length)
            new SpaceReclaiming(_bs, true);

        if(_bs.getAvailableSpace() >= msg.getBody().length){
            Schedulers.io().schedule(new Action1<Scheduler.Inner>() {
                @Override
                public void call(Scheduler.Inner arg0) {
                    _comm.MC.Sender.Send(Message.makeStored(msg.getFileID(), msg.getChunkNo()));
                }
            }, rand.nextInt(401), TimeUnit.MILLISECONDS);
            
            _bs.writeChunk(msg);

            _bs.Files.addChunk(msg.getFileID(), msg.getChunkNo(), msg.getReplicationDeg());
        }
    }

    Communicator _comm;
    BackupSystem _bs;
    private static Random rand = new Random();
}

@@ -355,16 +355,66 @@
 icator;%0A
+import pt.up.fe.sdis.proj1.utils.CounterObserver;%0A
 import p
@@ -470,24 +470,48 @@
 .Scheduler;%0A
+import rx.Subscription;%0A
 import rx.fu
@@ -1060,243 +1060,898 @@
     
-Schedulers.io().schedule(new Action1%3CScheduler.Inner%3E() %7B%0A                @Override%0A                public void call(Scheduler.Inner arg0) %7B%0A                    _comm.MC.Sender.Send(Message.makeStored(msg.getFileID(), msg.getChunkNo())
+final CounterObserver replyCounter = new CounterObserver();%0A            final Subscription sub = _bs.Comm.MDR.Publisher.getObservable()%0A                    .filter(new MessageFilter(Message.Type.STORED, msg.getFileID(), msg.getChunkNo())).subscribe(replyCounter);%0A            %0A            Schedulers.io().schedule(new Action1%3CScheduler.Inner%3E() %7B%0A                @Override%0A                public void call(Scheduler.Inner arg0) %7B%0A                    int i = replyCounter.getNumReceived();%0A                    if (i %3C msg.getReplicationDeg()) %7B%0A                        _comm.MC.Sender.Send(Message.makeStored(msg.getFileID(), msg.getChunkNo()));%0A                    %7D else %7B%0A                        _bs.deleteChunk(msg.getFileID(), msg.getChunkNo());%0A                        _bs.Files.removeChunk(msg.getFileID(), msg.getChunkNo());%0A                    %7D%0A                    sub.unsubscribe(
 );%0A 

Implement replyCounter on PeerChunkBackup.