NetworkOnmainThreadException android+Rxjava

Question

I have code that does write byte stream into a file and return the file but I am getting networkonmainthreadexception on writing input stream to file

accessTokenValidationWithResponseBodyForPDF(LinkApi.downloadFile(fileUrl))
                .flatMap(new Function<ResponseBody, ObservableSource<File>>() {
                    @Override
                    public ObservableSource<File> apply(@io.reactivex.annotations.NonNull ResponseBody responseBody) throws Exception {
                        return Observable.just(writeResponseBodyToDisk(responseBody))
                                .onErrorReturn(new Function<Throwable, File>() {
                                    @Override
                                    public File apply(@io.reactivex.annotations.NonNull Throwable throwable) throws Exception {
                                        return new File(AppConstants.not_valid);
                                    }
                                });
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new DisposableObserver<File>() {
                    @Override
                    public void onNext(@NotNull File resultObject) {
                        try {
                            hideProgress();
                            if (resultObject.getPath().equals(AppConstants.not_valid)) {

                                showAlertDialog(getString(R.string.something_went_wrong), true);

                            } else {

                                File pdfFile = (File) resultObject;
                                downloadedFile = pdfFile;
                                if (pdfFile != null) {
                                    displayPdf(pdfFile);
                                }

                            }
                        } catch (Exception e) {
                        }

                    }

                    @Override
                    public void onError(Throwable throwable) {
                        hideProgress();
                        
                    }

                    @Override
                    public void onComplete() {

                    }
                });

     private File writeResponseBodyToDisk(ResponseBody body) {

        
        String extStorageDirectory = getApplicationContext().getFilesDir().toString();

        File folder = new File(extStorageDirectory, ".Link");
        if (!folder.exists())
            folder.mkdir();

        SecureRandom secureRandom = new SecureRandom();
        int random = secureRandom.nextInt();

        File futureStudioIconFile = new File(extStorageDirectory + File.separator + ".Link" + File.separator + "link" + random + ".pdf");

        InputStream inputStream = null;
        OutputStream outputStream = null;

        try {
            byte[] fileReader = new byte[4096];

            long fileSize = body.contentLength();
            long fileSizeDownloaded = 0;

            inputStream = body.byteStream();
            outputStream = new FileOutputStream(futureStudioIconFile);

            while (true) {
                int read = inputStream.read(fileReader);//going to exception with networkonmainthreadexception

                if (read == -1) {
                    break;
                }

                outputStream.write(fileReader, 0, read);

                fileSizeDownloaded += read;

            }

//                outputStream.flush();

        } catch (Exception e) {

            Log.i("PDF", e.getMessage());
        } finally {
            safeClose(inputStream);
            safeClose(outputStream);
        }


        return futureStudioIconFile;

    }

Answer 1

This happens because you call writeResponseBodyToDisk then turn its result, into an Observable. You have to make the method call itself happen on a background thread, for example via

Observable.fromCallable(() -> writeResponseBodyToDisk(responseBody))
.subscribeOn(Schedulers.io())
.onErrorReturn(...)

Note that having .subscribeOn(Schedulers.io()) in your original code does not reach into the flatMap because the function runs on the emitter thread of accessTokenValidationWithResponseBodyForPDF. You have to put subscribeOns as close to the work and often repeat it many times at many places.