鸿蒙教学Demo代码.zip

package com.example.demo.class2; import io.reactivex.rxjava3.core.BackpressureStrategy; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Scheduler; import io.reactivex.rxjava3.disposables.Disposable; import io.reactivex.rxjava3.functions.Consumer; import io.reactivex.rxjava3.functions.Function; import io.reactivex.rxjava3.functions.Predicate; import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.subjects.PublishSubject; import io.reactivex.rxjava3.subjects.Subject; import java.lang.reflect.Method; import java.util.*; /** * RxBus * Created by gorden on 2016/5/12. * update 2017/3/1 */ @SuppressWarnings("unused") public class RxBus { public static final String LOG_BUS = "RXBUS_LOG"; private static volatile RxBus defaultInstance; private Map<Class, List<Disposable>> subscriptionsByEventType = new HashMap<>(); private Map<Object, List<Class>> eventTypesBySubscriber = new HashMap<>(); private Map<Class, List<SubscriberMethod>> subscriberMethodByEventType = new HashMap<>(); private final Subject<Object> bus; private RxBus() { this.bus = PublishSubject.create().toSerialized(); } public static RxBus get() { RxBus rxBus = defaultInstance; if (defaultInstance == null) { synchronized (RxBus.class) { rxBus = defaultInstance; if (defaultInstance == null) { rxBus = new RxBus(); defaultInstance = rxBus; } } } return rxBus; } /** * 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者 * * @param eventType 事件类型 * @return return */ private <T> Flowable<T> toObservable(Class<T> eventType) { return bus.toFlowable(BackpressureStrategy.BUFFER).ofType(eventType); } /** * 根据传递的code和 eventType 类型返回特定类型(eventType)的 被观察者 * * @param code 事件code * @param eventType 事件类型 */ private <T> Flowable<T> toObservable(final int code, final Class<T> eventType) { return bus.toFlowable(BackpressureStrategy.BUFFER).ofType(Message.class) .filter(new Predicate<Message>() { @Override public boolean test(Message o) throws Exception { return o.getCode() == code && eventType.isInstance(o.getObject()); } }).map(new Function<Message, Object>() { @Override public Object apply(Message o) throws Exception { return o.getObject(); } }).cast(eventType); } /** * 注册 * * @param subscriber 订阅者 */ public void register(Object subscriber) { Class<?> subClass = subscriber.getClass(); Method[] methods = subClass.getDeclaredMethods(); for (Method method : methods) { if (method.isAnnotationPresent(Subscribe.class)) { //获得参数类型 Class[] parameterType = method.getParameterTypes(); //参数不为空 且参数个数为1 if (parameterType != null && parameterType.length == 1) { Class eventType = parameterType[0]; addEventTypeToMap(subscriber, eventType); Subscribe sub = method.getAnnotation(Subscribe.class); int code = sub.code(); ThreadMode threadMode = sub.threadMode(); SubscriberMethod subscriberMethod = new SubscriberMethod(subscriber, method, eventType, code, threadMode); addSubscriberToMap(eventType, subscriberMethod); addSubscriber(subscriberMethod); } else if (parameterType == null || parameterType.length == 0) { Class eventType = BusData.class; addEventTypeToMap(subscriber, eventType); Subscribe sub = method.getAnnotation(Subscribe.class); int code = sub.code(); ThreadMode threadMode = sub.threadMode(); SubscriberMethod subscriberMethod = new SubscriberMethod(subscriber, method, eventType, code, threadMode); addSubscriberToMap(eventType, subscriberMethod); addSubscriber(subscriberMethod); } } } } /** * 将event的类型以订阅中subscriber为key保存到map里 * * @param subscriber 订阅者 * @param eventType event类型 */ private void addEventTypeToMap(Object subscriber, Class eventType) { List<Class> eventTypes = eventTypesBySubscriber.get(subscriber); if (eventTypes == null) { eventTypes = new ArrayList<>(); eventTypesBySubscriber.put(subscriber, eventTypes); } if (!eventTypes.contains(eventType)) { eventTypes.add(eventType); } } /** * 将注解方法信息以event类型为key保存到map中 * * @param eventType event类型 * @param subscriberMethod 注解方法信息 */ private void addSubscriberToMap(Class eventType, SubscriberMethod subscriberMethod) { List<SubscriberMethod> subscriberMethods = subscriberMethodByEventType.get(eventType); if (subscriberMethods == null) { subscriberMethods = new ArrayList<>(); subscriberMethodByEventType.put(eventType, subscriberMethods); } if (!subscriberMethods.contains(subscriberMethod)) { subscriberMethods.add(subscriberMethod); } } /** * 将订阅事件以event类型为key保存到map,用于取消订阅时用 * * @param eventType event类型 * @param disposable 订阅事件 */ private void addSubscriptionToMap(Class eventType, Disposable disposable) { List<Disposable> disposables = subscriptionsByEventType.get(eventType); if (disposables == null) { disposables = new ArrayList<>(); subscriptionsByEventType.put(eventType, disposables); } if (!disposables.contains(disposable)) { disposables.add(disposable); } } /** * 用RxJava添加订阅者 * * @param subscriberMethod d */ @SuppressWarnings("unchecked") private void addSubscriber(final SubscriberMethod subscriberMethod) { Flowable flowable; if (subscriberMethod.code == -1) { flowable = toObservable(subscriberMethod.eventType); } else { flowable = toObservable(subscriberMethod.code, subscriberMethod.eventType); } Disposable subscription = postToObservable(flowable, subscriberMethod) .subscribe(new Consumer<Object>() { @Override public void accept(Object o) throws Exception { callEvent(subscriberMethod, o); } }); addSubscriptionToMap(subscriberMethod.subscriber.getClass(), subscription); } /** * 用于处理订阅事件在那个线程中执行 * * @param observable d * @param subscriberMethod d * @return Observable */ private Flowable postToObservable(Flowable observable, SubscriberMethod subscriberMethod) { Scheduler scheduler; switch (subscriberMethod.threadMode) { case MAIN: scheduler = HarmonySchedulers.mainThread(); break; case NEW_THREAD: scheduler = Schedulers.newThread(); break; case CURRENT_THREAD: scheduler = Schedulers.trampoline(); break; default: throw new