@@ -0,0 +1,2212 @@
+package pt.up.fe.sdis.proj1.utils;%0A%0Aimport java.io.IOException;%0Aimport java.net.DatagramPacket;%0Aimport java.net.InetAddress;%0Aimport java.net.MulticastSocket;%0A%0Aimport pt.up.fe.sdis.proj1.messages.Message;%0Aimport rx.Observable;%0Aimport rx.functions.Func1;%0Aimport rx.schedulers.Schedulers;%0Aimport rx.subjects.PublishSubject;%0A%0Apublic class MulticastChannelMesssagePublisher extends Thread %7B%0A    private static final int MAX_UDP_SIZE = 65536;%0A    %0A    public MulticastChannelMesssagePublisher(String mCastAddr, int mCastPort) throws IOException %7B%0A        _mCastSocket = new MulticastSocket(mCastPort);%0A        _mCastSocket.joinGroup(InetAddress.getByName(mCastAddr));%0A    %7D%0A    %0A    public MulticastChannelMesssagePublisher(String mCastAddr, int mCastPort, int packetSize) throws IOException %7B%0A        _packetSize = packetSize;%0A        _mCastSocket = new MulticastSocket(mCastPort);%0A        _mCastSocket.joinGroup(InetAddress.getByName(mCastAddr));%0A    %7D%0A    %0A    @Override%0A    public void run() %7B%0A        byte%5B%5D buffer = new byte%5B_packetSize%5D;%0A        DatagramPacket dp = new DatagramPacket(buffer, _packetSize);%0A        %0A        while (true) %7B%0A            try %7B%0A                _mCastSocket.receive(dp);%0A                _subject.onNext(dp.getData());%0A            %7D catch (IOException e) %7B%0A            %7D%0A        %7D%0A    %7D    %0A    %0A    public Observable%3CMessage%3E getObservable() %7B%0A        return _observable;%0A    %7D%0A    %0A    private PublishSubject%3Cbyte%5B%5D%3E _subject = PublishSubject.create();%0A    private Observable%3CMessage%3E _observable = _subject%0A            .observeOn(Schedulers.io())%0A            .map(new Func1%3Cbyte%5B%5D, Message%3E() %7B%0A                @Override%0A                public Message call(byte%5B%5D arg0) %7B%0A                    try %7B%0A                        return Message.fromByteArray(arg0);%0A                    %7D catch (IOException e) %7B%0A                        return null;%0A                    %7D%0A                %7D%0A            %7D).filter(new Func1%3CMessage, Boolean%3E() %7B%0A                @Override%0A                public Boolean call(Message arg0) %7B%0A                    return arg0 != null;%0A                %7D%0A            %7D);%0A    %0A    private MulticastSocket _mCastSocket;%0A    private int _packetSize = MAX_UDP_SIZE;%0A%7D%0A

Implement MulticastChannelMessagePublisher class to handle Messages received via a multicast socket.