packagecom.jaydenxiao.common.baserx;
importjava.util.HashMap;
importjava.util.Map;
importrx.Observable;
importrx.Subscription;
importrx.android.schedulers.AndroidSchedulers;
importrx.functions.Action1;
importrx.subscriptions.CompositeSubscription;
/**
* 用于管理单个presenter的RxBus的事件和Rxjava相关代码的生命周期处理
* Created by xsf
* on 2016.08.14:50
*/
public classRxManager {
publicRxBusmRxBus= RxBus.getInstance();
//管理rxbus订阅
privateMap>mObservables=newHashMap<>();
/*管理Observables 和 Subscribers订阅*/
privateCompositeSubscriptionmCompositeSubscription=newCompositeSubscription();
/**
* RxBus注入监听
*@parameventName
*@paramaction1
*/
publicvoidon(String eventName,Action1 action1) {
Observable mObservable =mRxBus.register(eventName);
mObservables.put(eventName,mObservable);
/*订阅管理*/
mCompositeSubscription.add(mObservable.observeOn(AndroidSchedulers.mainThread())
.subscribe(action1, newAction1() {
@Override
public voidcall(Throwable throwable) {
throwable.printStackTrace();
}
}));
}
/**
* 单纯的Observables 和 Subscribers管理
*@paramm
*/
public voidadd(Subscription m) {
/*订阅管理*/
mCompositeSubscription.add(m);
}
/**
* 单个presenter生命周期结束，取消订阅和所有rxbus观察
*/
public voidclear() {
mCompositeSubscription.unsubscribe();// 取消所有订阅
for(Map.Entry> entry :mObservables.entrySet()) {
mRxBus.unregister(entry.getKey(),entry.getValue());// 移除rxbus观察
}
}
//发送rxbus
public voidpost(Object tag,Object content) {
mRxBus.post(tag,content);
}
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
packagecom.jaydenxiao.common.baserx;
importandroid.support.annotation.NonNull;
importcom.jaydenxiao.common.commonutils.LogUtils;
importjava.util.ArrayList;
importjava.util.Collection;
importjava.util.List;
importjava.util.concurrent.ConcurrentHashMap;
importrx.Observable;
importrx.Subscriber;
importrx.android.schedulers.AndroidSchedulers;
importrx.functions.Action1;
importrx.subjects.PublishSubject;
importrx.subjects.Subject;
/**
* 用RxJava实现的EventBus
* Created by xsf
* on 2016.08.14:50
*/
public classRxBus {
private staticRxBusinstance;
public static synchronizedRxBusgetInstance() {
if(null==instance) {
instance=newRxBus();
}
returninstance;
}
privateRxBus() {
}
@SuppressWarnings("rawtypes")
privateConcurrentHashMap>subjectMapper=newConcurrentHashMap>();
/**
* 订阅事件源
*
*@parammObservable
*@parammAction1
*@return
*/
publicRxBusOnEvent(Observable mObservable,Action1 mAction1) {
mObservable.observeOn(AndroidSchedulers.mainThread()).subscribe(mAction1, newAction1() {
@Override
public voidcall(Throwable throwable) {
throwable.printStackTrace();
}
});
returngetInstance();
}
/**
* 注册事件源
*
*@paramtag
*@return
*/
@SuppressWarnings({"rawtypes"})
public Observableregister(@NonNullObject tag) {
List subjectList =subjectMapper.get(tag);
if(null== subjectList) {
subjectList =newArrayList();
subjectMapper.put(tag,subjectList);
}
Subject subject;
subjectList.add(subject = PublishSubject.create());
LogUtils.logd("register"+tag +" size:"+ subjectList.size());
returnsubject;
}
@SuppressWarnings("rawtypes")
public voidunregister(@NonNullObject tag) {
List subjects =subjectMapper.get(tag);
if(null!= subjects) {
subjectMapper.remove(tag);
}
}
/**
* 取消监听
*
*@paramtag
*@paramobservable
*@return
*/
@SuppressWarnings("rawtypes")
publicRxBusunregister(@NonNullObject tag,
@NonNullObservable observable) {
if(null== observable)
returngetInstance();
List subjects =subjectMapper.get(tag);
if(null!= subjects) {
subjects.remove((Subject) observable);
if(isEmpty(subjects)) {
subjectMapper.remove(tag);
LogUtils.logd("unregister"+ tag +" size:"+ subjects.size());
}
}
returngetInstance();
}
public voidpost(@NonNullObject content) {
post(content.getClass().getName(),content);
}
/**
* 触发事件
*
*@paramcontent
*/
@SuppressWarnings({"unchecked","rawtypes"})
public voidpost(@NonNullObject tag,@NonNullObject content) {
LogUtils.logd("post"+"eventName: "+ tag);
List subjectList =subjectMapper.get(tag);
if(!isEmpty(subjectList)) {
for(Subject subject : subjectList) {
subject.onNext(content);
LogUtils.logd("onEvent"+"eventName: "+ tag);
}
}
}
@SuppressWarnings("rawtypes")
public static booleanisEmpty(Collection collection) {
return null== collection || collection.isEmpty();
}
}
