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注入監(jiān)聽
*@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觀察
}
}
//發(fā)送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實現(xiàn)的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);
}
}
/**
* 取消監(jiān)聽
*
*@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);
}
/**
* 觸發(fā)事件
*
*@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();
}
}
