Retrofit RxJava Adapter

Retrofit 的 RxJava Adapter 用於將 Retrofit 的服務呼叫結果轉換為 RxJava 的 Observable、Single、Maybe 或 Completable 等反應式類別。在進行網路請求時，能充分利用 RxJava 的功能，如流控制、非同步處理、錯誤處理等。

https://reqres.in/ 有一個 List User 的 GET method API，以下以這個 API 實作

pom.xml

廚了 Retrofit2 集 RxJava3 以外，增加 adapter-rxjava3

        <!--retrofit-->
        <dependency>
            <groupId>com.squareup.retrofit2</groupId>
            <artifactId>retrofit</artifactId>
            <version>2.11.0</version>
        </dependency>
        <dependency>
            <groupId>com.squareup.retrofit2</groupId>
            <artifactId>converter-gson</artifactId>
            <version>2.11.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.squareup.retrofit2/adapter-rxjava3 -->
        <dependency>
            <groupId>com.squareup.retrofit2</groupId>
            <artifactId>adapter-rxjava3</artifactId>
            <version>2.11.0</version>
        </dependency>

        <!-- rxjava -->
        <dependency>
            <groupId>io.reactivex.rxjava3</groupId>
            <artifactId>rxjava</artifactId>
            <version>3.1.8</version>
        </dependency>

Retrofit

透過網頁服務的 API 定義 service 及發送，回傳的資料

UserData.java

public class UserData {
    private long id;
    private String first_name;
    private String last_name;
    private String email;
// getter, setter, toString
}

UserList.java

public class UserList {
    private int page;
    private int per_page;
    private int total;
    private int total_pages;

    private List<UserData> data;
// getter, setter, toString
}

UserListService.java

注意，這邊的回傳結果改為 Observable<UserList>

public interface UserListService {
    @GET("/api/users")
    Observable<UserList> listUser(@Query("page") int pageno);
}

Retrofit RxJava 測試

RetrofitRxJavaTest.java

public class RetrofitRxJavaTest {
    public static void sync() {
        int maxIdleConnections = 10;
        int keepAliveDurationMills = 1000;
        OkHttpClient.Builder httpClient = new OkHttpClient.Builder()
                .connectionPool(new ConnectionPool(maxIdleConnections, keepAliveDurationMills, TimeUnit.MILLISECONDS));
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("https://reqres.in/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava3CallAdapterFactory.create())
                .client(httpClient.build())
                .build();

        UserListService service = retrofit.create(UserListService.class);

        Observable<UserList> observable = service.listUser(2);
        observable.subscribeOn(Schedulers.io())
                    .observeOn(Schedulers.single())
//                    .subscribe(
//                            System.out::println
//                    );
                    .subscribe(new DefaultObserver<UserList>() {
                        @Override
                        public void onNext(UserList response) {
                            System.out.println(response);
                        }

                        @Override
                        public void onError(@NonNull Throwable e) {

                        }

                        @Override
                        public void onComplete() {

                        }
                    });
    }

    public static void main(String[] args) {
        System.out.println(new Date());
        sync();

        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

如果程式放在 Android APP，則可以改為在主執行緒，也就是主畫面的執行緒上觀察結果

service.listUser(2).subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())