Angular 7 上传到带有进度指示器的 blob 存储

Question

我正在尝试创建一项服务，使我能够:

1. 将文件上传到 Azure blob 存储
2. 返回进度
3. 失败重试
4. 成功后调用我的服务保存文件路径

为了实现这一目标，我开始关注 this tutorial on Medium .我已经能够在存储中保存文件并返回当前进度。

当我想使用保存的文件路径调用我的服务时，我的问题就来了。

我查看了以下内容以尝试找出如何实现此目标但没有成功 making multiple http requests , the rxjs docs making subsequent http requests .

我很难理解如何将这些示例变成我可以使用的东西。

请注意，我正在尝试使 azureBlobStorageService 可重用，因此我不会在 azure 服务中进行第二次 HTTP 调用，这将由调用者负责。

下面是我的代码和注释，我在其中尝试添加 mergeMap 或 flatMap 等，但没有成功。我已经删除了这些引用，因为我尝试了很多变体，我觉得评论更清楚地描述了我想要实现的目标

上传组件

this.uploadProgress$ = from(inputNode.files as FileList).pipe(
  map(file => this._uploadService.updateCertificate(file)),
  combineAll()
);

上传服务

// this is where I would like to call my REST api when the file has uploaded to azure
updateCertificate(file: File): Observable<IUploadProgress> {
      return this._azureBlobStorage
        .uploadCertificateToBlobStorage(file, this.group)
        .pipe(
          map(
            progress => this.mapProgress(file, progress)
          ),
          //  flatMap(x => this._httpClient.post('xcv', JSON.Stringify(sasToken.filename))) <--fail 1
        )
        .pipe(flatMap(x => this._httpClient.post('', JSON.stringify('')))); <-- fail 2
  } // also tried merge map and a couple of others

  private mapProgress(file: File, progress: number): IUploadProgress {
    return {
      filename: file.name,
      progress: progress
    };
  }

Azure BlobStorage 服务

uploadCertificateToBlobStorage(file: File, group: string): Observable<number> 
{
  this.populateSasToken('/cert/' + group + '/' + file.name);
  return this.uploadToBlobStorage(this.sasToken, file);
}

private populateSasToken(filename: string): void {
    //create sasToken stuff
  }

private uploadToBlobStorage(sasToken: ISasToken, file: File): Observable<number> {
  const customBlockSize = this.getBlockSize(file);
  const options = { blockSize: customBlockSize };
  const blobService = this.createBlobService(sasToken.storageAccessToken, sasToken.storageUri);

  blobService.singleBlobPutThresholdInBytes = customBlockSize;

  return this.uploadFile(blobService, sasToken, file, options);
}

  private createBlobService(sasToken: string, blobUri: string): IBlobService {
    return this._blobStorage
      .createBlobServiceWithSas(blobUri, sasToken)
      .withFilter(new this._blobStorage.ExponentialRetryPolicyFilter());
  }

// Need to change this to return a custom object with number and the sasToken.filename
// but when I change this return type and the return of the associated methods I errors, I can't see what i'm missing
private uploadFile(
    blobService: IBlobService,
    sasToken: ISasToken,
    file: File,
    options: { blockSize: number }
  ): Observable<number> {
    return new Observable<number>(observer => {
      const speedSummary = blobService.createBlockBlobFromBrowserFile(
        sasToken.container,
        sasToken.filename,
        file,
        options,
        error => this.callback(error, observer)
      );
      speedSummary.on('progress', () => this.getProgress(speedSummary, observer, sasToken.filename));
    }).pipe(
      startWith(0),
      distinctUntilChanged()
      // retry(4) I think this will allow me to retry failed called to azure. 
    );
  }

  private getProgress(speedSummary: ISpeedSummary, observer: Subscriber<number>, fileName: string): void {
    const progress = parseInt(speedSummary.getCompletePercent(2), 10);
    observer.next(progress === 100 ? 99 : progress);
  }

  private callback(error: any, observer: Subscriber<number>): void {
    if (error) {
      console.log(error);
      observer.error(error);
    } else {
      observer.next(100);
      observer.complete();
    }
  }

================================

修改上传文件

以下原因

Type Observable is not assignable to type Observable

================================

export class Xxx {
  y: number;
  x: string;
}




private uploadFile(
    blobService: IBlobService,
    sasToken: ISasToken,
    file: File,
    options: { blockSize: number }
  ): Observable<Xxx> {
    return new Observable<Xxx>(observer => {
      const speedSummary = blobService.createBlockBlobFromBrowserFile(
        sasToken.container,
        sasToken.filename,
        file,
        options,
        error => this.callback(error, observer)
      );
      speedSummary.on('progress', () => this.getProgress(speedSummary, observer, sasToken.filename));
    }).pipe(
      startWith(0),
      distinctUntilChanged(),
      retry(4)
    );
  }

  private getProgress(speedSummary: ISpeedSummary, observer: Subscriber<Xxx>, fileName: string): void {
    const progress = parseInt(speedSummary.getCompletePercent(2), 10);
    // observer.next(progress === 100 ? 99 : progress);
    observer.next(new Xxx());
  }

  private callback(error: any, observer: Subscriber<Xxx>): void {
    if (error) {
      console.log(error);
      observer.error(error);
    } else {
      // observer.next(100);
      observer.next(new Xxx());
      observer.complete();
    }
  }

Answer 1

我用了https://npmjs.com/package/angular-progress-http

我已经有一段时间没看过这段代码了，但这里有一些代码片段可能会有所帮助

文件.服务.ts

import * as FileSaver from 'file-saver';
import { Injectable } from '@angular/core';
import { ProgressHttp, Progress } from "angular-progress-http";
import { RequestOptions, Headers, Response, ResponseContentType } from '@angular/http';
import { AuthHttp } from 'angular2-jwt';

import { Observable } from 'rxjs/Observable';

import { environment } from '../environments/environment';

@Injectable()
export class FileService {

  constructor(private http: ProgressHttp, private authHttp: AuthHttp) { }

  upload(url: string, files: File[], listener: (progress: Progress) => void): Observable<Response> {
    let formData: FormData = new FormData();
    files.forEach(file => {
      if (file) {
        formData.append('uploadFile', file, file.name);
      }
    });
    let headers = new Headers();
    headers.append('Authorization', `Bearer ${localStorage.getItem('token')}`);
    let options = new RequestOptions({ headers: headers });
    return this.http.withUploadProgressListener(listener).post(url, formData, options);
  }

  download(url: string, filename: string) {
    let options = new RequestOptions(new Headers({ 'Content-Type': 'application/json' }));
    options.responseType = ResponseContentType.Blob;

    this.authHttp.get(url, options).subscribe(r => {
        this.saveFileContent(r, filename);
    });
  }

  private saveFileContent(res: Response, filename: string) {
    let fileBlob = res.blob();
    let blob = new Blob([fileBlob]);
    FileSaver.saveAs(blob, filename);
  }
}

和 api 端点操作。

    [Authorize(Roles = "Administrator"), HttpPost("AddFile/{id}")]
    public async Task<IActionResult> AddFile(int id)
    {
        var files = Request.Form.Files;
        if (files.Count > 0)
        {
            var sectionId = dbContext.Articles.Where(a => a.Id == id).Select(a => a.SectionId).Single();
            using (var fileStream = files[0].OpenReadStream())
            {
                await fileService.SaveAsync($"sections/{sectionId}/articles/{id}/{files[0].FileName}", fileStream);
            }
        }
        return Content("Ok");
    }

和文件服务

using ContactManager.API.Models;
using Microsoft.Extensions.Options;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Blob;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;

namespace ContactManager.API.Storage
{
    public class AzureFileService : IFileService
    {
        AppSettings appSettings;

        CloudStorageAccount storageAccount = null;

        CloudStorageAccount StorageAccount
        {
            get
            {
                if (storageAccount == null)
                {
                    storageAccount = new CloudStorageAccount(new Microsoft.WindowsAzure.Storage.Auth.StorageCredentials(this.appSettings.AzureStorage.Account, this.appSettings.AzureStorage.Key), true);
                }
                return storageAccount;
            }
        }

        CloudBlobClient blobClient = null;

        CloudBlobClient BlobClient
        {
            get
            {
                if (blobClient == null)
                {
                    blobClient = StorageAccount.CreateCloudBlobClient();
                }
                return blobClient;
            }
        }

        private CloudBlobContainer GetContainerReference(Permission permission)
        {
            return BlobClient.GetContainerReference(permission == Permission.Public ?  appSettings.AzureStorage.PublicFolder : appSettings.AzureStorage.PrivateFolder);
        }

        public AzureFileService(IOptions<AppSettings> appSettings)
        {
            this.appSettings = appSettings.Value;
        }

        public async Task SaveAsync(string path, Stream stream, Permission permission = Permission.Public)
        {
            var container = GetContainerReference(permission);
            var blockBlob = container.GetBlockBlobReference(path);
            await blockBlob.UploadFromStreamAsync(stream);
        }

        public void Delete(string path, Permission permission = Permission.Public)
        {
            var container = GetContainerReference(permission);
            var blockBlob = container.GetBlockBlobReference(path);
            blockBlob.DeleteAsync();
        }

        public async Task<Stream> GetAsync(string path, Permission permission = Permission.Public)
        {
            var container = GetContainerReference(permission);
            var blockBlob = container.GetBlockBlobReference(path);
            var stream = new MemoryStream();
            await blockBlob.DownloadToStreamAsync(stream);
            stream.Position = 0;
            return stream;
        }
    }
}

希望对您有所帮助。