前言
整体变换简介
- 将一坨变换整合起来放在一起
- 用于固定的变换场景
整体变换体现了响应式编程思想：通过Transformer实现整体变化，变化后能向下传播。
1.RxJava1 整体变换实例
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
if (!subscriber.isUnsubscribed()) {
subscriber.onNext("test");
Log.d(TAG, "currentThread:" + Thread.currentThread());
subscriber.onCompleted();
}
}
}).compose(new Observable.Transformer<String, String>() {
@Override
public Observable<String> call(Observable<String> stringObservable) {
return stringObservable.
subscribeOn(Schedulers.newThread()).
observeOn(AndroidSchedulers.mainThread());
}
}).subscribe(new Observer<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
Log.d(TAG, "onNext:" + s);
Log.d(TAG, "currentThread:" + Thread.currentThread());
}
});
06-19 14:26:18.275 12255-12350/com.haocai.rxjavademo D/kpioneer: currentThread:Thread[RxNewThreadScheduler-1,5,main]
06-19 14:26:18.275 12255-12255/com.haocai.rxjavademo D/kpioneer: onNext:test
06-19 14:26:18.275 12255-12255/com.haocai.rxjavademo D/kpioneer: currentThread:Thread[main,5,main]
2.RxJava2 整体变换实例
/*---------无背压---------*/
Observable.
create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
if (!emitter.isDisposed()) {
Log.d(TAG, "无背压 currentThread:" + Thread.currentThread());
emitter.onNext("test");
emitter.onComplete();
}
}
}).
compose(new ObservableTransformer<String, String>() {
@Override
public ObservableSource<String> apply(Observable<String> upstream) {
return upstream.
subscribeOn(Schedulers.newThread()).
observeOn(AndroidSchedulers.mainThread());
}
}).
subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String o) {
Log.d(TAG, "无背压 onNext:" + o);
Log.d(TAG, "无背压 currentThread:" + Thread.currentThread());
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
/*---------有背压---------*/
Flowable.
create(new FlowableOnSubscribe<String>() {
@Override
public void subscribe(FlowableEmitter<String> emitter) throws Exception {
if (!emitter.isCancelled()) {
Log.d(TAG, "有背压 currentThread:" + Thread.currentThread());
emitter.onNext("test");
emitter.onComplete();
}
}
}, BackpressureStrategy.DROP).
compose(new FlowableTransformer<String, String>() {
@Override
public Publisher<String> apply(Flowable<String> upstream) {
return upstream.
subscribeOn(Schedulers.newThread()).
observeOn(AndroidSchedulers.mainThread());
}
}).
subscribe(new Subscriber<String>() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(String s) {
Log.d(TAG, "有背压 onNext:" + s);
Log.d(TAG, "有背压 currentThread:" + Thread.currentThread());
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
}
});
06-19 14:45:08.515 17301-17407/com.haocai.rxjavademo D/kpioneer: 无背压 currentThread:Thread[RxNewThreadScheduler-1,5,main]
06-19 14:45:08.515 17301-17301/com.haocai.rxjavademo D/kpioneer: 无背压 onNext:test
06-19 14:45:08.515 17301-17301/com.haocai.rxjavademo D/kpioneer: 无背压 currentThread:Thread[main,5,main]
06-19 14:45:08.525 17301-17408/com.haocai.rxjavademo D/kpioneer: 有背压 currentThread:Thread[RxNewThreadScheduler-2,5,main]
06-19 14:45:08.535 17301-17301/com.haocai.rxjavademo D/kpioneer: 有背压 onNext:test
06-19 14:45:08.535 17301-17301/com.haocai.rxjavademo D/kpioneer: 有背压 currentThread:Thread[main,5,main]
3.RxJava1 Transformer原理分析
- 继承自Func1接口，泛型参数是两个Observable
public interface Transformer<T, R> extends Func1<Observable<T>, Observable<R>> {
// cover for generics insanity
}
- 为compose方法的入参
public <R> Observable<R> compose(Transformer<? super T, ? extends R> transformer) {
return ((Transformer<T, R>) transformer).call(this);
}
- 传入一个Observable返回一个Observable
4.RxJava2 Transformer原理分析
4.1.RxJava2(无背压) ObservableTransformer
- 有一个apply方法
- 传入一个Observable返回一个新的Observable
- 为compose方法的入参
/**
* Interface to compose Observables.
*
* @param <Upstream> the upstream value type
* @param <Downstream> the downstream value type
*/
public interface ObservableTransformer<Upstream, Downstream> {
/**
* Applies a function to the upstream Observable and returns an ObservableSource with
* optionally different element type.
* @param upstream the upstream Observable instance
* @return the transformed ObservableSource instance
*/
@NonNull
ObservableSource<Downstream> apply(@NonNull Observable<Upstream> upstream);
}
/**
* Wraps an ObservableSource into an Observable if not already an Observable.
*
* <dl>
* <dt><b>Scheduler:</b></dt>
* <dd>{@code wrap} does not operate by default on a particular {@link Scheduler}.</dd>
* </dl>
*
* @param <T> the value type
* @param source the source ObservableSource instance
* @return the new Observable instance or the same as the source
* @throws NullPointerException if source is null
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Observable<T> wrap(ObservableSource<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
if (source instanceof Observable) {
return RxJavaPlugins.onAssembly((Observable<T>)source);
}
return RxJavaPlugins.onAssembly(new ObservableFromUnsafeSource<T>(source));
}
public final <R> Observable<R> compose(ObservableTransformer<? super T, ? extends R> composer) {
return wrap(((ObservableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
}
4.2.RxJava2(有背压) FlowableTransformer
- 有一个apply方法
- 传入一个Flowable返回一个新的Flowable
- 为compose方法的入参
public interface FlowableTransformer<Upstream, Downstream> {
/**
* Applies a function to the upstream Flowable and returns a Publisher with
* optionally different element type.
* @param upstream the upstream Flowable instance
* @return the transformed Publisher instance
*/
@NonNull
Publisher<Downstream> apply(@NonNull Flowable<Upstream> upstream);
}
@CheckReturnValue
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
@SuppressWarnings("unchecked")
public static <T> Flowable<T> fromPublisher(final Publisher<? extends T> source) {
if (source instanceof Flowable) {
return RxJavaPlugins.onAssembly((Flowable<T>)source);
}
ObjectHelper.requireNonNull(source, "publisher is null");
return RxJavaPlugins.onAssembly(new FlowableFromPublisher<T>(source));
}
@SuppressWarnings("unchecked")
@CheckReturnValue
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Flowable<R> compose(FlowableTransformer<? super T, ? extends R> composer) {
return fromPublisher(((FlowableTransformer<T, R>) ObjectHelper.requireNonNull(composer, "composer is null")).apply(this));
}
5.RxJava1 Transformer仿写
public interface Converter<T, R> extends Func1<Caller<T>, Caller<R>> {
}
public final <R> Caller<R> unify(Converter<T, R> converter) {
return converter.call(this);
}
实例
public class Lesson4_2Activity extends AppCompatActivity {
public static final String TAG = "kpioneer";
@Override
protected void onCreate(final Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_custom_test);
ButterKnife.bind(this);
}
@OnClick(R.id.testDo)
public void onViewClicked() {
Caller.
create(new Caller.OnCall<String>() {
@Override
public void call(Receiver<String> stringReceiver) {
stringReceiver.onReceive("test");
Log.d(TAG, "currentThread:" + Thread.currentThread());
}
}).
unify(new Caller.Converter<String, String>() {
@Override
public Caller<String> call(Caller<String> stringCaller) {
return stringCaller.
callOn(new NewThreadSwitcher()).
callbackOn(new LooperSwitcher(getMainLooper()));
}
}).
call(new Receiver<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable t) {
}
@Override
public void onReceive(String s) {
Log.d(TAG, "onNext:" + s);
Log.d(TAG, "currentThread:" + Thread.currentThread());
}
});
}
}
06-19 15:55:38.735 477-477/com.haocai.rxjavademo D/kpioneer: onNext:test
06-19 15:55:38.735 477-477/com.haocai.rxjavademo D/kpioneer: currentThread:Thread[main,5,main]
06-19 15:55:38.735 477-1180/com.haocai.rxjavademo D/kpioneer: currentThread:Thread[ NewThreadWorker,5,main]
6.RxJava2 Transformer仿写
6.1RxJava2 无背压
public <R> Caller<R> unify(CallerConverter<T,R> callerConverter){
return callerConverter.convert(this);
}
/**
* Created by Xionghu on 2018/6/19.
* Desc: 用于整体变换
*/
public interface TelephonerConverter<T,R> {
Telephoner<R> convert(Telephoner<T> telephoner);
}
6.2RxJava2 有背压
public <R> Telephoner<R> unify(TelephonerConverter<T, R> tTelephonerConverter) {
return tTelephonerConverter.convert(this);
}
/**
* Created by Xionghu on 2018/6/19.
* Desc: 用于整体变换
*/
public interface TelephonerConverter<T,R> {
Telephoner<R> convert(Telephoner<T> telephoner);
}
6.3实例
/**
* Created by Xionghu on 2018/6/11.
* Desc: RxJava2 整体变换仿写
*/
public class Lesson4_3Activity extends AppCompatActivity {
public static final String TAG = "kpioneer";
@Override
protected void onCreate(final Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_custom_test);
ButterKnife.bind(this);
}
@OnClick(R.id.testDo)
public void onViewClicked() {
Caller.
create(new CallerOnCall<String>() {
@Override
public void call(CallerEmitter<String> callerEmitter) {
callerEmitter.onReceive("test");
Log.d(TAG, "无背压 currentThread:" + Thread.currentThread());
}
}).
unify(new CallerConverter<String, String>() {
@Override
public Caller<String> convert(Caller<String> caller) {
return caller.
callOn(new NewThreadSwitcher()).
callbackOn(new LooperSwitcher(getMainLooper()));
}
}).
call(new Callee<String>() {
@Override
public void onCall(Release release) {
}
@Override
public void onReceive(String s) {
Log.d(TAG, "无背压 onReceive:" + s);
Log.d(TAG, "无背压 currentThread:" + Thread.currentThread());
}
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable t) {
}
});
Telephoner.
create(new TelephonerOnCall<String>() {
@Override
public void call(TelephonerEmitter<String> telephonerEmitter) {
telephonerEmitter.onReceive("test");
Log.d(TAG, "有背压 currentThread:" + Thread.currentThread());
}
}).
unify(new TelephonerConverter<String, String>() {
@Override
public Telephoner<String> convert(Telephoner<String> telephoner) {
return telephoner.
callOn(new NewThreadSwitcher()).
callbackOn(new LooperSwitcher(getMainLooper()));
}
}).
call(new Receiver<String>() {
@Override
public void onCall(Drop d) {
d.request(Long.MAX_VALUE);
}
@Override
public void onReceive(String s) {
Log.d(TAG, "有背压 onReceive:" + s);
Log.d(TAG, "有背压 currentThread:" + Thread.currentThread());
}
@Override
public void onError(Throwable t) {
}
@Override
public void onCompleted() {
}
});
}
}
06-19 16:28:23.215 11452-11698/com.haocai.rxjavademo D/kpioneer: 无背压 currentThread:Thread[NewThreadWorker,5,main]
06-19 16:28:23.235 11452-11703/com.haocai.rxjavademo D/kpioneer: 有背压 currentThread:Thread[NewThreadWorker,5,main]
06-19 16:28:23.235 11452-11452/com.haocai.rxjavademo D/kpioneer: 无背压 onReceive:test
06-19 16:28:23.235 11452-11452/com.haocai.rxjavademo D/kpioneer: 无背压 currentThread:Thread[main,5,main]
06-19 16:28:23.235 11452-11452/com.haocai.rxjavademo D/kpioneer: 有背压 onReceive:test
06-19 16:28:23.235 11452-11452/com.haocai.rxjavademo D/kpioneer: 有背压 currentThread:Thread[main,5,main]
