map的作用：
Returns an Observable that applies a specified function to each item emitted by the source ObservableSource and emits the results of these function applications.
对被观察者发送的每1个事件都通过指定的函数 处理，从而变换成另外一种事件 即， 将被观察者发送的事件转换为任意的类型事件。
image.png
demo:
Observable
.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
Log.d(TAG, "subscribe " + Thread.currentThread());
emitter.onComplete();
}
})
.map(new Function<String, Integer>() {
@Override
public Integer apply(String s) throws Exception {
return Integer.parseInt(s);
}
})
.subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "开始采用subscribe连接");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, "对Next事件作出响应" + integer);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "对Error事件作出响应");
}
@Override
public void onComplete() {
Log.d(TAG, "对Complete事件作出响应");
}
});
在map方法中我们传入了一个Function并且重写了apply方法 来定义如何对每个事件进行转换。这里我们调用了Integer.parseInt将String对象转换成了Integer对象。
下面我们通过源码看下 Observable在向Observer传递事件的时候 map 是如何进行了事件转换的。
1.进入map方法：
//Observable.java
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, "mapper is null");
//这里的this是指demo中自定义的Observable，mapper是demo中定义的事件转换Function
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}
map方法的返回值是Observable!这样就实现了在map方法后面可以直接调用subscribe来传入Observer，实现才能保证最爽的流式调用。仅仅如此吗？
那能不能说我们定义的Observer实际上直接观察的是map返回的Observable呢？那这个Observable又如何和我们最终的Observable关联呢？
2.查看ObservableMap具体实现
//ObservableMap.java
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
//根据上一节的知识我们知道：在调用Observable的subscribe方法后最终会调用到Observable实现类的
subscribeActual
//这里的t就是我们demo中自定义的Observer,source为我们自定义的Observable
public void subscribeActual(Observer<? super U> t) {
//创建了一个新的Observer: MapObserver 来订阅demo中定义的Observable,这样Observable中的消息会先传递到MapObserver
//我们猜测MapObserver充当了一个类似代理的东西，MapObserver在收到消息后经过处理肯定是要在传递到demo中定义的Observer（就是这个的参数t）的。
source.subscribe(new MapObserver<T, U>(t, function));
}
3.MapObserver如何将事件传递到demo中自定义的Observer呢？
//ObservableMap.java
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
//这里的actual为demo自定义的Observer,调用super最终会赋值给父类BasicFuseableObserver的成员变量downstream,被认为是MapObserver的下游Observer.
super(actual);
this.mapper = mapper;
}
@Override
//MapObserver中的onNext是在demo中自定义的Observable中调用emitter的onNext后执行的。
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downstream.onNext(null);
return;
}
U v;
try {
//mapper.apply(t)是指demo中调用map方法是传入的Fuction,通过调用该方法完成事件转换。这里是map原理的关键。
v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
//事件转换后在通知下游的Observer,也就是demo中实现的Observer,这样就完成了一个完整的时间传递。
downstream.onNext(v);
}
