RxJava Multiple Network Requests with replay

Question

I have two API calls. The first API call returns a list of the most recent orders made with the order date and status. The second network call uses the unique id in the first call to return information about the order. I have the following:

public interface ServiceAPI
{
    @POST("buyer/recentOrders")
    Single<RecentOrdersJSON> getRecentOrders ();

    @POST("buyer/orderDetails/{ORDER_ID}")
    Single<OrderDetailsJSON> orderDetails(
            @Path("ORDER_ID") String ORDER_ID);
}

The RecentOrdersJSON Class:

public class RecentOrdersJSON {

    @SerializedName("orders")
    @Expose
    private List<Order> orders = null;
    @SerializedName("recordCount")
    @Expose
    private Integer recordCount;
    @SerializedName("msg")
    @Expose
    private String msg;
    @SerializedName("currency_symbol")
    @Expose
    private String currency_symbol;
    @SerializedName("status")
    @Expose
    private Integer status;

    public List<Order> getOrders() {
        return orders;
    }

    public void setOrders(List<Order> orders) {
        this.orders = orders;
    }

    public Integer getRecordCount() {
        return recordCount;
    }

    public void setRecordCount(Integer recordCount) {
        this.recordCount = recordCount;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public String getCurrency_symbol() {
        return currency_symbol;
    }

    public void setCurrency_symbol(String currency_symbol) {
        this.currency_symbol = currency_symbol;
    }

    public Integer getStatus() {
        return status;
    }

    public void setStatus(Integer status) {
        this.status = status;
    }

}

The OrderDetailsJSON class:

public class OrderDetailsJSON {

    @SerializedName("productDetails")
    @Expose
    private List<ProductDetail> productDetails = null;
    @SerializedName("orderStatusHistory")
    @Expose
    private List<OrderStatusHistory> orderStatusHistory = null;
    @SerializedName("orderInfo")
    @Expose
    private OrderInfo orderInfo;
    @SerializedName("billingAddress")
    @Expose
    private BillingAddress billingAddress;
    @SerializedName("unreadNotificationCount")
    @Expose
    private Integer unreadNotificationCount;
    @SerializedName("msg")
    @Expose
    private String msg;
    @SerializedName("currency_symbol")
    @Expose
    private String currency_symbol;
    @SerializedName("status")
    @Expose
    private Integer status;

    public List<ProductDetail> getProductDetails() {
        return productDetails;
    }

    public void setProductDetails(List<ProductDetail> productDetails) {
        this.productDetails = productDetails;
    }

    public List<OrderStatusHistory> getOrderStatusHistory() {
        return orderStatusHistory;
    }

    public void setOrderStatusHistory(List<OrderStatusHistory> orderStatusHistory) {
        this.orderStatusHistory = orderStatusHistory;
    }

    public OrderInfo getOrderInfo() {
        return orderInfo;
    }

    public void setOrderInfo(OrderInfo orderInfo) {
        this.orderInfo = orderInfo;
    }

    public BillingAddress getBillingAddress() {
        return billingAddress;
    }

    public void setBillingAddress(BillingAddress billingAddress) {
        this.billingAddress = billingAddress;
    }

    public Integer getUnreadNotificationCount() {
        return unreadNotificationCount;
    }

    public void setUnreadNotificationCount(Integer unreadNotificationCount) {
        this.unreadNotificationCount = unreadNotificationCount;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    public String getCurrency_symbol() {
        return currency_symbol;
    }

    public void setCurrency_symbol(String currency_symbol) {
        this.currency_symbol = currency_symbol;
    }

    public Integer getStatus() {
        return status;
    }

    public void setStatus(Integer status) {
        this.status = status;
    }

}

private Observable<RecentOrdersJSON> getOrders()
{
    return ServiceAPI.getRecentOrders()
            .toObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());
}


ConnectableObservable<RecentOrdersJSON> ordersObservable = (ConnectableObservable<RecentOrdersJSON>) getOrders().replay();

disposable.add(
        ordersObservable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new DisposableObserver<RecentOrdersJSON>()
                {
                    @Override
                    public void onNext(RecentOrdersJSON recentOrdersJSON)
                    {

                        // Refreshing List
                        orderList.clear();
                        orderList.addAll(recentOrdersJSON.getOrders());
                        orderlistAdapter.notifyDataSetChanged();
                    }

                    @Override
                    public void onError(Throwable e)
                    {

                    }

                    @Override
                    public void onComplete()
                    {

                    }
                }));


// Fetching individual orderInfo
disposable.add(ordersObservable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

                .flatMap(new Function<List<Order>, ObservableSource<Order>>()
               {
                  @Override
                    public ObservableSource<Order> apply(List<Order> orders)
                   {
                                                   return Observable.fromIterable(orders);
                    }
               })

);


private Observable<OrderDetailsJSON> getOrderObservable(final String Order_ID)
{
    return ServiceAPI.orderDetails(Order_ID)
            .toObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());

}

The Order POJO:

   @SerializedName("order_id")
    @Expose
    private String order_id;
    @SerializedName("order_date_added")
    @Expose
    private String order_date_added;
    @SerializedName("order_status")
    @Expose
    private Integer order_status;

    public String getOrder_id() {
        return order_id;
    }

    public void setOrder_id(String order_id) {
        this.order_id = order_id;
    }

    public String getOrder_date_added() {
        return order_date_added;
    }

    public void setOrder_date_added(String order_date_added) {
        this.order_date_added = order_date_added;
    }

    public Integer getOrder_status() {
        return order_status;
    }

    public void setOrder_status(Integer order_status) {
        this.order_status = order_status;
    }

When trying to use a flatmap to get the id from the first network call result I get the following message Observable cannot be applied.

Answer 1

First of all :

Your Order pojo need to have OrderDetailsJSON object

public class Order {

   ...

   OrderDetailsJSON orderDetailsJson;

   public void setOrderDetailsJson(OrderDetailsJSON orderDetailsJson) {
      this.orderDetailsJson = orderDetailsJson;
   }

   public OrderDetailsJSON getOrderDetailsJson(){
      return this.orderDetailsJson
   }
}

The replay

replay() operator is used to make an Observable emit the data on new subscriptions without re-executing the logic again i.e. without making the api call again(getOrders().replay()). Since you are using ConnectableObservable make following changes in you implementation to achieve what you want ...

Replace this

// Fetching individual orderInfo
disposable.add(ordersObservable
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

                .flatMap(new Function<List<Order>, ObservableSource<Order>>()
               {
                  @Override
                    public ObservableSource<Order> apply(List<Order> orders)
                   {
                                                   return Observable.fromIterable(orders);
                    }
               })

);

with

/**
 * Fetching individual order details
 * First FlatMap converts single List<Order> from RecentOrdersJSON to multiple emissions of orders
 * Second FlatMap makes HTTP call on each Order emission to fetch the details
 * */
disposable.add(
        ordersObservable
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            /**
             * Converting List<Order> emission to single Order emissions
             * */
            .flatMap(new Function<RecentOrdersJSON, ObservableSource<Order>>() {
                @Override
                public ObservableSource<Order> apply(RecentOrdersJSON recentOrderJson)
                {
                    return Observable.fromIterable(recentOrdersJSON.getOrders());
                }
            })
            /**
             * Fetching details on each order emission
             * */
            .flatMap(new Function<Order, ObservableSource<Order>>() {
                @Override
                public ObservableSource<Order> apply(Order order) throws Exception {
                    return getOrderDetailsObservable(order);
                }
            })
            .subscribeWith(new DisposableObserver<Order>() {

                @Override
                public void onNext(Order order) {
                    int position = orderList.indexOf(order);

                    if (position == -1) {
                        // Order not found in the list
                        // This shouldn't happen
                        return;
                    }

                    orderList.set(position, order);
                    orderlistAdapter.notifyItemChanged(position);
                }

                @Override
                public void onError(Throwable e) {
                    //show error
                }

                @Override
                public void onComplete() {

                }
}));

Also replace this

private Observable<OrderDetailsJSON> getOrderObservable(final String Order_ID)
{ 
    return ServiceAPI.orderDetails(Order_ID)
            .toObservable() 
            .subscribeOn(Schedulers.io()) 
            .observeOn(AndroidSchedulers.mainThread()); 

} 

with

private Observable<Order> getOrderDetailsObservable(final Order order) {
    return ServiceAPI
            .orderDetails(order.getOrder_id())
            .toObservable()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .map(new Function<OrderDetailsJSON, Order>() {
                @Override
                public Order apply(OrderDetailsJSON orderDetailsJSON) throws Exception {
                    order.setOrderDetailsJson(orderDetailsJSON);
                    return order;
                }
            });
}