сжимайте и загружайте несколько файлов с прогрессом с помощью RxWorker

Question

Я пытаюсь загрузить несколько изображений на сервер, а до этого я хочу сжать свои изображения. для сжатия части я использую AdvancedLuban Library в моем проекте для сжатия выбранных изображений пользователей, а затем для загрузки части я использую Retrofit в Rx Java способе. Я делаю все в своем классе докладчиков, и все это работает правильно. Моя проблема в том, что я хочу сделать все это в моем классе UploadWorker, который унаследован от RxWorker, и при загрузке показывать прогресс в уведомлении. Написанный мной код не работает должным образом, и ход уведомлений не обновлялся правильно.

Вот мой UploadWorker.class

public class UploadWorker extends RxWorker {

    private static final String TAG = UploadWorker.class.getSimpleName();

    public static final String KEY_STRING_DATA = "string_data";

    private static final int COMPRESS_MAX_SIZE = 200;
    private static final int PROGRESS_MAX = 100;

    private Context context;

    private FileUploader uploader;


    public UploadWorker(@NonNull Context context, @NonNull WorkerParameters workerParams) {
        super(context, workerParams);
        this.context = context;
        ApiService apiService = ServiceBuilder.buildService(ApiService.class);
        uploader = new FileUploader(apiService);
    }

    @NonNull
    @Override
    public Single<Result> createWork() {


        Data data = getInputData();

        String strData = data.getString(KEY_STRING_DATA);

        List<String> stringList = deserializeFromJson(strData);

        List<File> files = new ArrayList<>();

        for (String path : stringList) {
            File f = new File(path);
            files.add(f);
        }

        return Single.fromObservable(Luban.compress(context,files)
                .setMaxSize(COMPRESS_MAX_SIZE)
                .putGear(Luban.CUSTOM_GEAR)
                .setCompressFormat(Bitmap.CompressFormat.JPEG)
                .asListObservable()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .map(new Function<List<File>, ArrayList<String>>() {
                    @Override
                    public ArrayList<String> apply(List<File> files) throws Exception {

                        ArrayList<String> filListPath = new ArrayList<>();

                        for (File file:files) {
                            filListPath.add(file.getAbsolutePath());
                        }
                        return filListPath ;
                    }
                })
                .map(new Function<ArrayList<String>, Disposable>() {
                    @Override
                    public Disposable apply(ArrayList<String> strings) throws Exception {
                        HashMap<String, RequestBody> map = new HashMap<>();
                        return getUploadObserver(map,strings);
                    }
                }).map(new Function<Disposable, Result>() {
                    @Override
                    public Result apply(Disposable disposable) throws Exception {
                        return Result.success();
                    }
                })
        );
    }


    private Disposable getUploadObserver(HashMap<String, RequestBody> map, ArrayList<String> files) {
        return uploader.uploadMultiImage(map, files)
                .subscribeOn(Schedulers.computation())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Double>() {
                    @Override
                    public void accept(Double progress) throws Exception {
                        notifyUpload((int) (100 * progress));
                        Log.d(TAG, "accept: " + (int) (100 * progress));
                    }
                });
    }

    public void notifyUpload(int progress) {

        

        NotificationManagerCompat notificationManagerCompat = NotificationManagerCompat.from(context);

           NotificationCompat.Builder  builder = new NotificationCompat.Builder(context, Config.NOTIFICATION_CHANNEL);

            builder.setSmallIcon(R.drawable.ic_notification_icon)
                    .setContentTitle("Upload")
                    .setContentText("Uploading in progress")
                    .setPriority(NotificationCompat.PRIORITY_LOW)
                    .setAutoCancel(true);

        if (progress < PROGRESS_MAX) {
            builder.setProgress(PROGRESS_MAX, progress, false);
        }else {
            builder.setContentText("Upload complete")
                    .setProgress(0,0,false);
        }
        notificationManagerCompat.notify(200, builder.build());
    }

    public static List<String> deserializeFromJson(String jsonString){
        Gson gson = new Gson();
        Type listOf = new TypeToken<ArrayList<String>>() {}.getType();
        return gson.fromJson(jsonString,listOf);
    }

}

FileUploader.class

public class FileUploader implements FileUploaderContract{

    private final ApiService service;

    private static final String TAG = FileUploaderModel.class.getSimpleName();

    public FileUploaderModel(ApiService service) {
        this.service = service;
    }
    @Override
    public Flowable<Double> uploadMultiImage(HashMap<String,RequestBody> map, ArrayList<String> filePaths) {
        return Flowable.create(new FlowableOnSubscribe<Double>() {
            @Override
            public void subscribe(FlowableEmitter<Double> emitter) throws Exception {
                try {

                    List<MultipartBody.Part> myPart = new ArrayList<>();
                    for (String path:filePaths) {
                        myPart.add(createMultipartBody(path, emitter));

                    }
                    ResponseBody response = service.postMultipleImage(map,myPart).blockingGet();
                    Log.d(TAG, "subscribe: " + response);
                    emitter.onComplete();
                } catch (Exception e) {
                    emitter.tryOnError(e);
                }
            }
        }, BackpressureStrategy.LATEST);
    }

    @NonNull
    private RequestBody createPartFromString(String descriptionString) {
        return RequestBody.create(MultipartBody.FORM, descriptionString);
    }

    private MultipartBody.Part createMultipartBody(String filePath, FlowableEmitter<Double> emitter) {
        File file = new File(filePath);
        return MultipartBody.Part.createFormData("image", file.getName(), createCountingRequestBody(file, emitter));
    }



    private RequestBody createCountingRequestBody(File file, FlowableEmitter<Double> emitter) {
        RequestBody requestBody = createRequestBody(file);
        return new CountingRequestBody(requestBody, (bytesWritten, contentLength) -> {
            double progress = (1.0 * bytesWritten) / contentLength;
            emitter.onNext(progress);
        });
    }
}

И вызовите мой UploadWorker в моем MainActivity , как показано ниже

String strData = serializeToJson(mAdapter.getImageList());
        Data data = new Data.Builder()
                .putString(UploadWorker.KEY_STRING_DATA,strData)
                .build();


        WorkRequest mRequestWork = new OneTimeWorkRequest.Builder(UploadWorker.class)
                .setInitialDelay(1, TimeUnit.SECONDS)
                .setInputData(data)
                .build();

WorkManager.getInstance(getContext()).enqueue(mRequestWork)