This is an reactive implementation of an event bus, with a few convenient functions especially useful for handling events with activities, fragments and similar.
RxJava V1: If you are looking for a version for RxJava V1, check out my RXBus
- it allows you to send events to a bus
- it allows you to subscribe to special events wherever you want
- it allows you to queue events until an activity is resumed (to make sure views are accessable for example) and even to pause and resume events => for example, queue events while an activity is paused and emit them as soon as it get's resumed
- it's very lightweight
Gradle (via JitPack.io)
- add jitpack to your project's
build.gradle:
repositories {
maven { url "https://jitpack.io" }
}- add the compile statement to your module's
build.gradle:
dependencies {
implementation 'com.github.MFlisar.RxBus2:lib:<LATEST-VERSION>'
// optinonal:
// implementation 'com.github.MFlisar.RxBus2:lib-extension:<LATEST-VERSION>'
// alternatively, to include ALL modules at once
// implementation 'com.github.MFlisar:RxBus2:<LATEST-VERSION>'
}Content
- Demo
- Simple usage
- Sending an event
- Advanced usage - QUEUING AND BINDING
- Advanced usage - KEYS
- Advanced usage - TRANSFORM
- Advanced usage - SUB CLASSES
- Helper class -
RxDisposableManager
Just check out the DemoActivity, it will show the base usage and the difference between the default and the queued RxBus
Use the RxBusBuilder to create Subscriptions or simple Flowables. Just like following:
// Variant 1 - create a simple Flowable:
Flowable<TestEvent> simpleFlowable = RxBusBuilder.create(TestEvent.class).build();
// Variant 2 - subscribe with the BUILDER:
Disposable simpleDisposable = RxBusBuilder.create(TestEvent.class)
.subscribe(new Consumer<TestEvent>() {
@Override
public void accept(TestEvent event) {
// Event received, handle it...
}
});// Send an event to the bus - all observers that observe this class WITHOUT a key will receive this event
RxBus.get().send(new TestEvent());
// Send an event to the bus - only observers that observe the class AND key will receive this event
RxBus.get()
.withKey(R.id.observer_key_1)
.send(new TestEvent());
// Send an event to the bus - all observers that either observe the class or the class AND key will receive this event
RxBus.get()
.withKey(R.id.observer_key_1)
.withSendToDefaultBus()
.send(new TestEvent());
// Send an event to the bus and cast it to a specific class (a base class of multiple classes)
// This allows you to send casted objects to the bus
// so that all observers of the casted class will receive this event
// (of course only if the cast of the send event is possible, otherwise an exception is thrown!)
RxBus.get()
.withCast(TestEvent.class)
.send(new SubTestEvent());Of course you combine those functionalities as you want!
You can use this library to subscribe to events and only get them when your activity is resumed, so that you can be sure views are available. Or you can just pause and resume the bus based on any logic you want. Just like following:
RxBusBuilder.create(TestEvent.class)
// this enables the queuing mode! Passed object must implement IRxBusQueue interface, see the demo app for an example
.withQueuing(rxBusQueue)
.withOnNext(new Consumer<TestEvent>() {
@Override
public void accept(TestEvent s) {
// activity IS resumed, you can safely update your UI for example
}
})
.buildSubscription();Additionally, you can bind the subscription to an object and afterwards call RxDisposableManager.unsubscribe(boundObject); to unsubcribe ANY subscription that was bound to boundObject just like following:
// boundObject... can be for example your activity
RxBusBuilder.create(TestEvent.class)
// this enables the queuing mode! Passed object must implement IRXBusQueue interface, see the demo app for an example
.withQueuing(rxBusQueue)
.withBound(boundObject)
.subscribe(new Consumer<TestEvent>() {
@Override
public void accept(TestEvent s) {
// activity IS resumed, you can safely update your UI for example
}
});
// call this for example in your activities onDestroy or wherever appropriate to unsubscribe ALL subscriptions at once that are bound to the boundOBject
RxDisposableManager.unsubscribe(boundObject);You can use this library to subscribe to events of a typ and ONLY get them when it was send to the bus with a special key (and and additionally only when your activity is resumed, as this example shows via .withQueuing()), so that you can distinct event subscriptions of the same class based on a key (the key can be an Integer or a String). Just like following:
RxBusBuilder.create(TestEvent.class)
// this enables the binding to the key
.withKey(R.id.observer_key_1) // you can provide multiple keys as well
.withQueuing(rxBusQueue)
.withBound(boundObject)
.subscribe(new Consumer<TestEvent>() {
@Override
public void accept(TestEvent event) {
// activity IS resumed, you can safely update your UI for example
}
});You can pass in a FlowableTransformer to transform the observed event to whatever you want!
FlowableTransformer<TestEvent, TestEventTransformed> transformer = new FlowableTransformer<TestEvent, TestEventTransformed>() {
@Override
public Observable<TestEventTransformed> apply(Flowable<TestEvent> observable) {
return observable
.map(new Function<TestEvent, TestEventTransformed>() {
@Override
public TestEventTransformed apply(TestEvent event) {
return event.transform();
}
});
}
};
RxBusBuilder.create(TestEvent.class)
.withQueuing(rxBusQueue)
.withBound(boundObject)
.subscribe(new Consumer<TestEventTransformed>() {
@Override
public void accept(TestEventTransformed transformedEvent) {
}
}, transformer);By default, sub class handling is disabled and you must use RxBus.withCast(BaseClass.class) to send sub classes to base class observers (and only to the base class observers). You have two other options for handling this.
- Use
RxBus.get().withSendToSuperClasses(true).send(subClassEvent)- this will send the event to all super class observers of the send event as well - Enable above by default via
RxBusDefaults.get().setSendToSuperClassesAsWell(true)then all events that are send viaRxBus.get().sendEvent(subClassEvent)are send to all super classes of the send event as well
If you enable RxBusDefaults.get().setSendToSuperClassesAsWell(true), you can still overwrite this global default value for a single event by sending the event via RxBus.get().withSendToSuperClasses(false).send(subClassEvent) o a per event base.
This class helps to bind Disposables to objects and offers an easy way to unsubscribe all Disposables that are bound to an object at once. You can simply use this directly via RxDisposableManager.addDisposable(boundObject, flowable) or use it directly with the RxBusBuilder via the RxBusBuilder.withBound(boundObject) which will automatically add the Disposable to the RxDisposableManager as soon as you call RxBusBuilder.subscribe(...). This will automatically add the Disposable to the RxDisposableManager when you call RxBusBuilder.subscribe(...). Afterwards you unsubscribe via RxDisposableManager.unsubscribe(boundObject);. The bound object can be an Activity or Fragment for example, but any other object as well.
Example - Direct usage
Subscription subscription = RxBusBuilder.create(...).subscribe(...);
RxDisposableManager.addDisposable(activity, subscription);Example - RxBusBuilder
RxBusBuilder.create(...).withBound(activity)...;Now you only have to make sure to unsubscribe again like following:
RxDisposableManager.unsubscribe(activity);This will remove ANY subscription that is bound to activity and therefore this can be used in your activity's onDestroy method to make sure ALL subscriptions are unsubscribed at once so that you don't leak the activity.