@@ -0,0 +1,1208 @@
+package com.sync.changsha.rxjava;%0A%0Aimport java.util.concurrent.TimeUnit;%0Aimport org.junit.Test;%0Aimport rx.Observable;%0Aimport rx.Observer;%0Aimport rx.functions.Action1;%0Aimport rx.functions.Func1;%0A%0A/**%0A * %3Cp%3E%0A * %E8%BD%AC%E6%8D%A2%E6%B5%8B%E8%AF%95%0A * %3C/p%3E%0A * Created by Administrator on 2017/7/9 0009.%0A */%0Apublic class Transforming %7B%0A%0A  @Test%0A  public void concatMap() %7B%0A%0A    Observable.just(10, 20, 30)%0A        .concatMap(new Func1%3CInteger, Observable%3CInteger%3E%3E() %7B%0A          @Override public Observable%3CInteger%3E call(Integer integer) %7B%0A            // 10%E7%9A%84%E5%BB%B6%E8%BF%9F%E6%89%A7%E8%A1%8C%E6%97%B6%E9%97%B4%E4%B8%BA200%E6%AF%AB%E7%A7%92%E3%80%8120%E5%92%8C30%E7%9A%84%E5%BB%B6%E8%BF%9F%E6%89%A7%E8%A1%8C%E6%97%B6%E9%97%B4%E4%B8%BA180%E6%AF%AB%E7%A7%92%0A            int delay = 200;%0A            if (integer %3E 10) %7B%0A              delay = 180;%0A            %7D%0A            return Observable.from(new Integer%5B%5D %7Binteger, integer / 2%7D).delay(delay, TimeUnit.MILLISECONDS);%0A          %7D%0A        %7D).subscribe(new Observer%3CInteger%3E() %7B%0A      @Override public void onCompleted() %7B%0A        System.out.println(%22 onCompleted %22);%0A      %7D%0A%0A      @Override public void onError(Throwable throwable) %7B%0A        System.out.println(%22 onError , throwable : %22 + throwable.toString());%0A      %7D%0A%0A      @Override public void onNext(Integer integer) %7B%0A        System.out.println(%22 onNext , result : %22 + integer);%0A      %7D%0A    %7D);%0A  %7D%0A%0A%0A%7D%0A

test concatMap