Чтение файлов параллельно наблюдаемым

Question

Я пишу какое-то приложение для чтения выбранных файлов пользователем и преобразования их в base64. Я хочу получить уведомление, когда все файлы будут прочитаны в памяти. Для этого я использую Observable, где обрабатывают событие onloadFileReader и отправляют полное уведомление. Я использую forkJoin для параллельной работы.

См. Ниже код, где я создаю Observable и подписывается на него.

onChange($event: any) { 
    console.log('No of files selected: ' + $event.target.files.length); 
    var observableBatch : any = []; 

    var rawFiles = $event.target.files; 
    for (var i = rawFiles.length - 1; i >= 0; i--) { 

     var reader = new FileReader(); 
     var file = rawFiles[i]; 
     var myobservable = Observable.create((observer: any) => { 
     reader.onload = function (e: any) { 
      var data = e.target; 
      var imageSrc = data.result; 
      console.log('File loaded succesfully.'); 
      observer.next("File loaded"); 
      observer.complete(); 
     }; 
     }); 

     observableBatch.push(myobservable); 
     reader.readAsArrayBuffer(file); 

    } 

    Observable.forkJoin(observableBatch) 
    .subscribe(
     (m) => { 
     console.log(m); 
     }, 
     (e) => { 
     console.log(e); 
     }, 
    () => { 
     console.log("All file(s) loading completed!!!"); 
     } 
    ); 
} 

Полный пример кода доступен в plunkr

Когда я выбираю один файл, onload функция выполняется, и я получаю следующие журналы консольных

Однако, когда я выбираю несколько файлов, onload выполняется только один раз, и пакетная операция не завершена. Пожалуйста, смотрите следующие журналы консоли

Может кто-нибудь помочь мне понять, где я делаю ошибку?

Answer 1

Я нашел this Ответ на аналогичный вопрос. По-видимому, это связано с порядком, в котором выполняются цикл и обратные вызовы. Я думаю, что .forkJoin() ожидает, что столько Observables будет завершено, как было передано ему, но к тому времени, когда он получит все и подписаться, первый onload уже завершен, поэтому завершение Observable никогда не произойдет.

В любом случае, вы можете решить проблему, поставив код, в котором вы настроили обратный вызов FileReader, Observable и onload в свою собственную функцию. Here is the plunkr showing that it works.

export class AppComponent { 
    title = 'file reader'; 
    observableBatch : any = []; 

    onChange($event: any) { 
    console.log('No of files selected: ' + $event.target.files.length); 
    //Make sure to clear the observableBatch array before restarting the whole process. 
    this.observableBatch = []; 

    var rawFiles = $event.target.files; 
    for (var i = rawFiles.length - 1; i >= 0; i--) { 
     this.setUpFile(rawFiles[i]); 
    } 

    Observable.forkJoin(this.observableBatch) 
    .subscribe(
     (m) => { 
     console.log(m); 
     }, 
     (e) => { 
     console.log(e); 
     }, 
    () => { 
     console.log("All file(s) loading completed!!!"); 
     } 
    ); 
    } 


    setUpFile(file) { 
    var reader = new FileReader(file); 
    var myobservable = Observable.create((observer: any) => { 
     reader.onload = function (e: any) { 
     var data = e.target; 
     var imageSrc = data.result; 
     console.log('File loaded succesfully.'); 
     observer.next("File loaded"); 
     observer.complete(); 
     }; 
    }); 

    this.observableBatch.push(myobservable); 
    reader.readAsArrayBuffer(file); 
    } 
} 

Answer 2

Я использую этот код с помощью flatmap, чтобы убедиться, что все загружается

import {Injectable} from '@angular/core' 
import {Attachment} from './attachments.component' 
import {Inject} from '@angular/core' 
import {BehaviorSubject} from 'rxjs/BehaviorSubject' 
import {Observable} from "rxjs/Observable"; 
import {AttachmentBackendService} from './attachment.backend.service' 
import 'rxjs/add/observable/from' 
import 'rxjs/add/operator/mergeMap' 

@Injectable() 
export class AttachmentStore { 
    private _attachments: BehaviorSubject<Attachment[]> = new  BehaviorSubject<Attachment[]>([]) 
    private dataStore : { 
    attachments : Attachment[] 
    } 
    private storeId : string = '' 
    private attachmentId : number = 0 

    constructor(private attachmentBackendService: AttachmentBackendService) { 
    this.dataStore = { attachments : [] } 
    } 

    get attachments() { 
    return this._attachments.asObservable() 
    } 

    // public 
    addFiles(files: FileList) { 
    let fileArray = Array.from(files) 
    this.processFiles(
     fileArray[0], 
     fileArray.slice(1)) 
     .subscribe(
     (attachment) => { 
      this.storeAndSaveAttachment(attachment) 
      this._attachments.next(this.dataStore.attachments) 
     }, 
     (e) => { 
      console.log(e) 
     }, 
     () => { 
      console.log("file loading completed!!!") 
     }) 
    return this.storeId 
} 

removeFile(index: number) { 
    let attachment = this.dataStore.attachments[index] 
    this.attachmentBackendService.deleteAttachment(this.storeId, attachment.id) 
    this.dataStore.attachments.splice(index, 1) 
    this._attachments.next(this.dataStore.attachments) 
} 

// private 
private processFiles(file : File, fileArray : File[]) { 
    if (fileArray.length > 0) { 
    return this.processFiles(
        fileArray.slice(0,1)[0], 
        fileArray.slice(1)) 
       .flatMap((attachment) => { 
        this.storeAndSaveAttachment(attachment) 
        return this.fileReaderObs(file,this.attachmentId++) 
       }) 
    } else { 
    if (this.storeId == '') 
    { 
     this.storeId = this.attachmentBackendService.storeId 
    } 
    return this.fileReaderObs(file,this.attachmentId++) 
    } 
} 

private storeAndSaveAttachment(attachment : Attachment) { 
    this.dataStore.attachments.push(attachment) 
    this.attachmentBackendService.saveAttachment(this.storeId, attachment) 
} 

private fileReaderObs(file : File, attachmentId : number) { 
    let reader = new FileReader() 
    let fileReaderObs = Observable.create((observer: any) => { 
    reader.onload = function() { 
     let attachment : Attachment = { 
     id : attachmentId, 
     name : file.name, 
     data : btoa(reader.result) 
     } 
     observer.next(attachment) 
     observer.complete() 
    } 
    }) 
    reader.readAsBinaryString(file) 
    return fileReaderObs 
} 

}

Answer 3

Я хочу предложить это решение. Не стесняйтесь сказать мне, есть ли у вас проблемы с этим решением.

const files = Array.from(event.srcElement.files); 
 

 
Observable.from(files) 
 
    .map((file: File) => { 
 
    const reader = new FileReader(); 
 
    const load$ = Observable.fromEvent(reader, 'load').take(1); 
 
    const read$ = Observable.of(file).do(reader.readAsDataURL.bind(reader)); 
 
    return [load$, read$]; 
 
    }) 
 
    .toArray() 
 
    .switchMap((values: any) => { 
 
    const arrayObservables = values.reduce((acc, value) => acc.concat(value), []); 
 
    return Observable.forkJoin(...arrayObservables); 
 
    }) 
 
    .subscribe({ 
 
    next: console.log 
 
    });

Приветствия.