package com.github.kubode.rxeventbus;

import rx.Scheduler;
import rx.Subscription;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;

/**
 * An implementation of event bus using {@link PublishSubject}.
 * <p/>
 * MT-Safe.
 * <p/>
 * Not support generics.
 */
public class RxEventBus {
    private final Subject<Event, Event> subject = new SerializedSubject<>(PublishSubject.<Event>create());

    /**
     * Posts an {@link Event} to subscribed handlers.
     * <p/>
     * If no handlers have been subscribed for event's class, unhandled will be called with unhandled event.
     *
     * @param <E>       Type of event.
     * @param event     An event to post.
     * @param unhandled Will be called if event is not handled.
     *                  Note: If handler subscribed by using async {@link Scheduler}, it can't guarantee event is actually handled.
     */
    public <E extends Event> void post(E event, Action1<E> unhandled) {
        subject.onNext(event);
        if (event.handledCount == 0) {
            unhandled.call(event);
        }
    }

    /**
     * An overload method of {@link #post(Event, Action1)} that do nothing on unhandled.
     *
     * @see #post(Event, Action1)
     */
    public <E extends Event> void post(E event) {
        post(event, new Action1<E>() {
            @Override
            public void call(E e) {
            }
        });
    }

    /**
     * Subscribes handler to receive events type of specified class.
     * <p/>
     * You should call {@link Subscription#unsubscribe()} if you want to stop receiving events.
     *
     * @param <E>       Type of event.
     * @param clazz     Type of event that you want to receive.
     * @param handler   An event handler function that called if an event is posted.
     * @param scheduler handler will dispatched on this.
     */
    public <E extends Event> Subscription subscribe(Class<E> clazz, Action1<E> handler, Scheduler scheduler) {
        return subject
                .ofType(clazz)
                .doOnNext(new Action1<E>() {
                    @Override
                    public void call(E e) {
                        e.handledCount++;
                    }
                })
                .observeOn(scheduler)
                .subscribe(handler);
    }

    /**
     * An overload method of {@link #subscribe(Class, Action1, Scheduler)} that scheduler specified by {@link Schedulers#immediate()}
     *
     * @see #subscribe(Class, Action1, Scheduler)
     */
    public <E extends Event> Subscription subscribe(Class<E> clazz, Action1<E> handler) {
        return subscribe(clazz, handler, Schedulers.immediate());
    }
}

@@ -248,16 +248,19 @@
 %0A/**%0A * 
+%3Cp%3E
 An imple
@@ -315,28 +315,27 @@
 ct%7D.
+%3C/p%3E
 %0A * %3Cp
-/%3E%0A * 
+%3E
 MT-Safe.
 %0A * 
@@ -334,20 +334,19 @@
 afe.
-%0A * %3Cp/
+%3C/p
 %3E%0A * 
+%3Cp%3E
 Not 
@@ -362,16 +362,20 @@
 enerics.
+%3C/p%3E
 %0A */%0Apub
@@ -520,16 +520,19 @@
 %0A     * 
+%3Cp%3E
 Posts an
@@ -569,16 +569,20 @@
 andlers.
+%3C/p%3E
 %0A     * 
@@ -583,26 +583,17 @@
     * %3Cp
-/%3E%0A     * 
+%3E
 If no ha
@@ -681,24 +681,28 @@
 ndled event.
+%3C/p%3E
 %0A     *%0A    
@@ -1183,32 +1183,35 @@
 %0A    /**%0A     * 
+%3Cp%3E
 An overload meth
@@ -1271,24 +1271,28 @@
 n unhandled.
+%3C/p%3E
 %0A     *%0A    
@@ -1524,16 +1524,19 @@
 %0A     * 
+%3Cp%3E
 Subscrib
@@ -1588,16 +1588,20 @@
 d class.
+%3C/p%3E
 %0A     * 
@@ -1606,18 +1606,9 @@
 * %3Cp
-/%3E%0A     * 
+%3E
 You 
@@ -1691,16 +1691,20 @@
  events.
+%3C/p%3E
 %0A     *%0A
@@ -1887,24 +1887,24 @@
  is posted.%0A
-
      * @para
@@ -1948,16 +1948,80 @@
 n this.%0A
+     * @return A %7B@link Subscription%7D which can stop observing.%0A
      */%0A

fix javadoc