amazon observable RxJava1.java

    private Observable<SentToS3Event> getAmazonObservable(String newFilename) {
        Observable<SentToS3Event> amazonObservable = Observable.defer(new Func0<Observable<SentToS3Event>>() {
            @Override
            public Observable<SentToS3Event> call() {
                return Observable.create((ObservableEmitter<SentToS3Event> subscriber) -> {
                    final PutObjectRequest putObjectRequest = new PutObjectRequest(AmazonS3FetchParams.MY_BUCKET_NAME, newFilename, new java.io.File(fileURI))
                            .withCannedAcl(CannedAccessControlList.PublicRead)
                            .withStorageClass(StorageClass.Standard);
                    putObjectRequest.setGeneralProgressListener(new S3ProgressListener() {
                        @Override
                        public void onPersistableTransfer(PersistableTransfer persistableTransfer) {

                        }

                        @Override
                        public void progressChanged(ProgressEvent progressEvent) {
                            if (progressEvent.getEventCode() == ProgressEvent.COMPLETED_EVENT_CODE) {
                                Logger.d("TRANSFER finished: " + progressEvent.getBytesTransferred() + " bytes");
                                SentToS3Event sentEvent = new SentToS3Event(fileName, fileURI, position,
                                        0, from, nrRetry, messageDbId, getAmazonImageUrl(fileName));
                                subscriber.onNext(sentEvent);
                                subscriber.onComplete();
                            } else if (progressEvent.getEventCode() == ProgressEvent.FAILED_EVENT_CODE) {
                                SentToS3Event sentEvent = new SentToS3Event(fileName, fileURI, position,
                                        -1, from, nrRetry, messageDbId, getAmazonImageUrl(fileName));
                                subscriber.onNext(sentEvent);
                                subscriber.onComplete();
                            }
                        }
                    });
                    PutObjectResult objectResult = s3Client.putObject(putObjectRequest);
                });
            }
        });

        return amazonObservable;
    }

    private Observable<SentToS3Event> getAmazonObservable(final String newFilename) {
        return Observable.create(observableEmitter -> {
            final PutObjectRequest putObjectRequest = new PutObjectRequest(AmazonS3FetchParams.MY_BUCKET_NAME, newFilename, new java.io.File(fileURI))
                    .withCannedAcl(CannedAccessControlList.PublicRead)
                    .withStorageClass(StorageClass.Standard);

            putObjectRequest.setGeneralProgressListener(new S3ProgressListener() {
                @Override
                public void onPersistableTransfer(PersistableTransfer persistableTransfer) {

                }

                @Override
                public void progressChanged(ProgressEvent progressEvent) {
                    if (progressEvent.getEventCode() == ProgressEvent.COMPLETED_EVENT_CODE) {
                        Logger.d("TRANSFER finished: " + progressEvent.getBytesTransferred() + " bytes");
                        SentToS3Event sentEvent = new SentToS3Event(fileName, fileURI, position,
                                0, from, nrRetry, messageDbId, getAmazonImageUrl(fileName));
                        observableEmitter.onNext(sentEvent);
                        observableEmitter.onComplete();
                    } else if (progressEvent.getEventCode() == ProgressEvent.FAILED_EVENT_CODE) {
                        SentToS3Event sentEvent = new SentToS3Event(fileName, fileURI, position,
                                -1, from, nrRetry, messageDbId, getAmazonImageUrl(fileName));
                        observableEmitter.onNext(sentEvent);
                        observableEmitter.onComplete();
                    }
                }
            });
            PutObjectResult objectResult = s3Client.putObject(putObjectRequest);
        });
    }