package pt.up.fe.sdis.proj1;

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

public class Main {

    public static void hello(String... names) {
        Observable.from(names).subscribe(new Observer<String>() {

            @Override
            public void onCompleted() {
                // TODO Auto-generated method stub
                
            }

            @Override
            public void onError(Throwable arg0) {
                // TODO Auto-generated method stub
                
            }

            @Override
            public void onNext(String s) {
                System.out.println("Hello " + s + "!");
            }
        });
    }
    
    private static Runnable serverMockUp = new Runnable() {
        
        @Override
        public void run() {
            Message msg = new Message(Message.Type.DELETE);
            
            System.out.println("sender: " + Thread.currentThread().getId());
            
            while (!done) {
                ps.onNext(msg);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            
        }
    };
    private static boolean done = false;
    private static PublishSubject<Message> ps = PublishSubject.create();
    
    static class MyObserver implements Observer<Message> {

        public MyObserver(String name) {
            _name = name;
        }
        
        private final String _name;
        
        @Override
        public void onCompleted() {
            // TODO Auto-generated method stub
            
        }

        @Override
        public void onError(Throwable arg0) {
            // TODO Auto-generated method stub
            
        }

        @Override
        public void onNext(Message arg0) {
            System.out.println("obs '"+ _name +"': " + Thread.currentThread().getId());
            
        }
        
    }
    
    public static void main(String[] args) throws InterruptedException {
        System.out.println("main: " + Thread.currentThread().getId());
        
        Observable<Message> obsvb = ps.observeOn(Schedulers.io());
        
        new Thread(serverMockUp).start();
        
        
        for (int i = 0; i < 10; ++i) {
            Subscription s1 = obsvb.subscribe(new MyObserver("s" + Integer.toString(i + 1)));
            Subscription s2 = obsvb.subscribe(new MyObserver("s" + Integer.toString(i + 11)));
            Thread.sleep(200);
            s1.unsubscribe();
            s2.unsubscribe();
        }        
        Thread.sleep(200);
        System.out.println("Done!");
        done = true;
    }

}

@@ -27,2807 +27,74 @@
 1;%0A%0A
-import pt.up.fe.sdis.proj1.messages.Message;%0Aimport rx.Observable;%0Aimport rx.Observer;%0Aimport rx.Subscription;%0Aimport rx.schedulers.Schedulers;%0Aimport rx.subjects.PublishSubject;%0A%0Apublic class Main %7B%0A%0A    public static void hello(String... names) %7B%0A        Observable.from(names).subscribe(new Observer%3CString%3E() %7B%0A%0A            @Override%0A            public void onCompleted() %7B%0A                // TODO Auto-generated method stub%0A                %0A            %7D%0A%0A            @Override%0A            public void onError(Throwable arg0) %7B%0A                // TODO Auto-generated method stub%0A                %0A            %7D%0A%0A            @Override%0A            public void onNext(String s) %7B%0A                System.out.println(%22Hello %22 + s + %22!%22);%0A            %7D%0A        %7D);%0A    %7D%0A    %0A    private static Runnable serverMockUp = new Runnable() %7B%0A        %0A        @Override%0A        public void run() %7B%0A            Message msg = new Message(Message.Type.DELETE);%0A            %0A            System.out.println(%22sender: %22 + Thread.currentThread().getId());%0A            %0A            while (!done) %7B%0A                ps.onNext(msg);%0A                try %7B%0A                    Thread.sleep(100);%0A                %7D catch (InterruptedException e) %7B%0A                    e.printStackTrace();%0A                %7D%0A            %7D%0A            %0A        %7D%0A    %7D;%0A    private static boolean done = false;%0A    private static PublishSubject%3CMessage%3E ps = PublishSubject.create();%0A    %0A    static class MyObserver implements Observer%3CMessage%3E %7B%0A%0A        public MyObserver(String name) %7B%0A            _name = name;%0A        %7D%0A        %0A        private final String _name;%0A        %0A        @Override%0A        public void onCompleted() %7B%0A            // TODO Auto-generated method stub%0A            %0A        %7D%0A%0A        @Override%0A        public void onError(Throwable arg0) %7B%0A            // TODO Auto-generated method stub%0A            %0A        %7D%0A%0A        @Override%0A        public void onNext(Message arg0) %7B%0A            System.out.println(%22obs '%22+ _name +%22': %22 + Thread.currentThread().getId());%0A            %0A        %7D%0A        %0A    %7D%0A    %0A    public static void main(String%5B%5D args) throws InterruptedException %7B%0A        System.out.println(%22main: %22 + Thread.currentThread().getId());%0A        %0A        Observable%3CMessage%3E obsvb = ps.observeOn(Schedulers.io());%0A        %0A        new Thread(serverMockUp).start();%0A        %0A        %0A        for (int i = 0; i %3C 10; ++i) %7B%0A            Subscription s1 = obsvb.subscribe(new MyObserver(%22s%22 + Integer.toString(i + 1)));%0A            Subscription s2 = obsvb.subscribe(new MyObserver(%22s%22 + Integer.toString(i + 11)));%0A            Thread.sleep(200);%0A            s1.unsubscribe();%0A            s2.unsubscribe();%0A        %7D        %0A        Thread.sleep(200);%0A        System.out.println(%22Done!%22);%0A        done = true;
+public class Main %7B%0A%0A    public static void main(String%5B%5D args) %7B%0A
 %0A   

Clear Main.java.