package pt.up.fe.sdis.proj1.utils;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;

import pt.up.fe.sdis.proj1.messages.Message;
import rx.Observable;
import rx.Observer;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;

public class MulticastChannelMesssagePublisher extends Thread {
    private static final int MAX_UDP_SIZE = 65536;

    public MulticastChannelMesssagePublisher(String mCastAddr, int mCastPort)
            throws IOException {
        _mCastSocket = new MulticastSocket(mCastPort);
        _mCastSocket.joinGroup(InetAddress.getByName(mCastAddr));
    }

    public MulticastChannelMesssagePublisher(String mCastAddr, int mCastPort,
            int packetSize) throws IOException {
        _packetSize = packetSize;
        _mCastSocket = new MulticastSocket(mCastPort);
        _mCastSocket.joinGroup(InetAddress.getByName(mCastAddr));
    }

    @Override
    public void run() {
        byte[] buffer = new byte[_packetSize];
        DatagramPacket dp = new DatagramPacket(buffer, _packetSize);
        while (true) {
            buffer = new byte[_packetSize];
            try {
                _mCastSocket.receive(dp);
                _subject.onNext(dp.getData());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public Observable<Message> getObservable() {
        return _observable;
    }

    private PublishSubject<byte[]> _subject = PublishSubject.create();
    private Observable<Message> _observable = _subject
            .observeOn(Schedulers.io()).map(new Func1<byte[], Message>() {
                @Override
                public Message call(byte[] arg0) {
                    try {
                        Message result = Message.fromByteArray(arg0);
                        return result;
                    } catch (IOException e) {
                        // e.printStackTrace();
                        return null;
                    }
                }
            }).filter(new Func1<Message, Boolean>() {
                @Override
                public Boolean call(Message arg0) {
                    return arg0 != null;
                }
            });

    private MulticastSocket _mCastSocket;
    private int _packetSize = MAX_UDP_SIZE;

    public static void main(String[] args) {
        String addr = "230.0.0.1";
        int port = 11099;

        try {
            MulticastChannelMesssagePublisher mcmp = new MulticastChannelMesssagePublisher(
                    addr, port);

            mcmp.start();

            mcmp.getObservable()/*.filter(new Func1<Message, Boolean>() {

                @Override
                public Boolean call(Message arg0) {
                    return arg0.type == Message.Type.CHUNK;
                }
            })*/.subscribe(new Observer<Message>() {

                long i = 0L;

                @Override
                public void onCompleted() {
                    System.out.println("Done");
                }

                @Override
                public void onError(Throwable arg0) {
                    arg0.printStackTrace();
                }

                @Override
                public void onNext(Message arg0) {
                    System.out.println(Thread.currentThread().getId() + " : "
                            + i++ + " : " + arg0.type);
                }
            });
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

@@ -1260,16 +1260,101 @@
 ve(dp);%0A
+                // System.out.println(%22Something received from %22 + dp.getAddress());%0A
         
@@ -1389,16 +1389,24 @@
 etData()
+.clone()
 );%0A     
@@ -1962,16 +1962,83 @@
 (arg0);%0A
+                        // System.out.println(%22Message received%22);%0A
         
@@ -2097,18 +2097,16 @@
  catch (
-IO
 Exceptio
@@ -2442,16 +2442,52 @@
       %7D)
+.observeOn(Schedulers.computation())
 ;%0A%0A    p
@@ -2638,19 +2638,21 @@
 dr = %2223
-0.0
+9.255
 .0.1%22;%0A 
@@ -2677,16 +2677,25 @@
 11099;%0A%0A
+        %0A
         
@@ -3440,32 +3440,918 @@
              %7D%0A%0A
+                @Override%0A                public void onNext(Message arg0) %7B%0A                    long j = i++;%0A                    System.out.println(Thread.currentThread().getId() + %22 : %22%0A                            + j + %22 : %22 + arg0.type);%0A                %7D%0A            %7D);%0A            %0A            /*mcmp.getObservable().filter(new Func1%3CMessage, Boolean%3E() %7B%0A%0A                @Override%0A                public Boolean call(Message arg0) %7B%0A                    return arg0.type == Message.Type.PUTCHUNK;%0A                %7D%0A            %7D).subscribe(new Observer%3CMessage%3E() %7B%0A%0A                long i = 0L;%0A%0A                @Override%0A                public void onCompleted() %7B%0A                    System.out.println(%22Done%22);%0A                %7D%0A%0A                @Override%0A                public void onError(Throwable arg0) %7B%0A                    arg0.printStackTrace();%0A                %7D%0A%0A
                 
@@ -4570,32 +4570,34 @@
 %0A            %7D);
+*/
 %0A        %7D catch

Fix bugs in MulticastChannelMessagePublisher