import { Injectable } from "@angular/core" ; import { environment } from "src/environments/environment" ; import { BehaviorSubject , Subscription , Subject , Observable } from "rxjs" ; import { VerifyResponse } from "../components/videotape/videotape.component" ; import { FileChunkService } from "./file-chunk.service" ; import { FileHashService } from "./file-hash.service" ; import { HttpClient } from "@angular/common/http" ; import { RequestService } from "./request.service" ; import { CHUNK_SIZE } from "../config" ; export interface ImagePayload { type : string ; hash ? : number ; } export interface UploadPayload { chunk : Blob ; type : string ; upload_type ? : "upload" | "merge" | "fileVerify" ; hash ? : number ; fileHash ? : number ; sort_index ? : number ; cutNum ? : number ; index ? : number ; } export interface QueueNode < T > { type : "chunk" | "file" ; size : number ; name : string ; mimeType : string ; payload ? : T ; blob ? : Blob ; percentage ? : number ; hash ? : number | string ; chunks ? : QueueNode < any > [ ] ; index ? : number ; chunksLength ? : number ; already ? : number ; path ? : string ; } export interface Chunk { file : Blob ; } @ Injectable ( { providedIn : "root" , } ) export class UploadQueueService { status : boolean = false ; isSuspend : boolean = false ; queue : Array < QueueNode < any > > = [ ] ; videoQueue : Array < QueueNode < any > > = [ ] ; alreadyUploadQueue : Array < QueueNode < any > > = [ ] ; onChangeUploadStatus : BehaviorSubject < boolean > = new BehaviorSubject ( this . isSuspend ) ; alreadyUpProgress : boolean = false ; getImgData : BehaviorSubject < { data : Blob ; elem : ImagePayload ; } > = new BehaviorSubject ( null ) ; globalImgCache : Subscription ; globalVideoCache : Subscription ; alreadyUploadPayload$ : Subject < QueueNode < ImagePayload > > = new Subject ( ) ; constructor ( private fileChunk , private fileHash , private http , private request ) { } num : number = 0 ; param ele push ( ele ) { this . queue . push ( ele ) ; ! this . status && this . run ( ) ; return true ; } /**
     * 队列去重
     */ duplicateRemoval ( ) { let hash = { } , data = this . queue ; data = data . reduce ( ( preVal , curVal ) => { hash [ curVal . hash ] ? "" : ( hash [ curVal . hash ] = true && preVal . push ( curVal ) ) ; return preVal ; } , [ ] ) ; return data ; } /**
     * 队列出列 -
     */ pop ( payload ? , notEmitImg ? ) { let qNode = this . queue . shift ( ) ; if ( ! qNode ) return ; qNode . path = payload ? payload . path : "" ; this . alreadyUploadQueue . push ( qNode ) ; ! notEmitImg && this . alreadyUploadPayload$ . next ( qNode ) ; return qNode ; } get front ( ) : QueueNode < any > { return this . queue [ 0 ] ; } get rear ( ) : QueueNode < any > { return this . queue [ this . queue . length - 1 ] ; } clear ( ) { this . queue = [ ] ; } get size ( ) : number { return this . queue . length ; } suspend ( ) { if ( ! this . requestList || ! this . requestList . length ) return ; this . status = false ; this . requestList [ 0 ] . abort ( ) ; this . isSuspend = true ; this . onChangeUploadStatus . next ( this . isSuspend ) ; } restart ( ) { this . queue = [ ... this . queue ] ; this . requestList = [ ] ; this . status = true ; this . isSuspend = false ; this . upload ( ) ; this . onChangeUploadStatus . next ( this . isSuspend ) ; } /**
     * run方法 自动上传
     */ run ( ) { this . status = true ; this . upload ( ) ; } add ( ele ) { if ( ele . size < CHUNK_SIZE ) { this . push ( ele ) ; } else this . packingFile ( ele ) ; } requestList : XMLHttpRequest [ ] = [ ] ; /**
     * 上传视频
     */ async packingFile ( ele ) { ele . payload = { } ; const fileChunkList = await this . fileChunk . handleFile ( ele . blob ) ; ele . blob = null ; const hash = await this . fileHash . initHashWorker ( fileChunkList ) ; ele . payload . hash = hash ; ele . type = "chunk" ; const { data : { uploadedList , shouldUpload } , } = await this . verifyUpload ( hash , ele . payload . path , ele . payload ) ; if ( ! shouldUpload ) { alert ( "文件已上传" ) ; return ; } else { this . chunkToQueueNode ( uploadedList ? uploadedList : [ ] , fileChunkList , ele ) ; } } /**
     * 网速判断
     */ networkLogic ( ) { let val = true ; return val ; } async driveUpload ( node ) { if ( ! node ) return true ; let formData = new FormData ( ) ; formData . append ( "type" , node . type ) ; formData . append ( "mimeType" , node . mimeType ) ; if ( node . type === "chunk" ) { formData . append ( "chunk" , ( node as any ) . chunk ) ; for ( let key in node . payload ) { formData . append ( key , node . payload [ key ] ) ; } } node . type === "file" && formData . append ( "file" , node . blob ) ; return await this . request . request ( { url : `${ environment . apiUrl }/upload` , requestList : this . requestList , onProgress : this . createProgressHandler ( this . front ) , data : formData , headers : { } , } ) ; } /**
     * 通用上传API
     * 递归
     */ async upload ( ) { let chunkNode = null ; if ( this . front . chunks && this . front . chunks . length ) { chunkNode = this . front . chunks ; } else { chunkNode = this . front ; } if ( chunkNode && ( chunkNode as Array < QueueNode < any > > ) . length ) { const { data , code } = await this . driveUpload ( chunkNode [ 0 ] ) ; if ( data && code === 200 ) { ( chunkNode as Array < QueueNode < any > > ) . shift ( ) ; if ( ( chunkNode as Array < QueueNode < any > > ) . length ) { this . upload ( ) ; return ; } else if ( data && data . canMerge === true ) { this . merge ( ) ; return ; } else { this . cancel ( ) ; return ; } } else { this . cancel ( ) ; alert ( "上传出错，请重新上传" ) ; } } console . log ( "切片上传不应该执行到这里" ) ; const { code , data } = await this . driveUpload ( chunkNode as QueueNode < any > ) ; if ( code === 200 ) { this . pop ( ) ; if ( this . size ) this . upload ( ) ; else this . status = false ; } } async merge ( ) { const { code , data } = await this . mergeRequest ( this . front ) ; if ( code === 200 && this . front ) { this . pop ( data [ 0 ] , data . status === "already" ) ; if ( this . size ) { this . upload ( ) ; } else this . status = false ; return ; } } createProgressHandler ( item ) { if ( ! item ) return ; try { return ( e ) => { if ( item . type === "file" ) { item . percentage = parseInt ( String ( ( e . loaded / e . total ) * 100 ) ) ; console . log ( e . loaded , e . total ) ; } else { ( item as any ) . current = parseInt ( String ( ( e . loaded / e . total ) * 100 ) ) ; if ( ( item as any ) . current === 100 ) { item . percentage = 100 - ( ( ( item . chunks . length - 1 ) / item . chunksLength ) as any ) . toFixed ( 2 ) * 100 ; console . log ( "--------------------- 切片完成 ---------------------" , item . percentage ) ; } } } ; } catch ( e ) { console . log ( e , item ) ; } } async chunkToQueueNode ( uploadedList = [ ] , fileChunkList , ele ) { const data = fileChunkList . map ( ( { file } , index ) => { let obj = { chunk : file , type : ele . type , payload : null , size : file . size , chunk_total : fileChunkList . length , percentage : uploadedList . includes ( index + "" ) ? 100 : 0 , } ; obj . payload = { cutNum : fileChunkList . length , fileHash : ele . payload . hash , hash : ele . payload . hash + "__" + index as any , index : index , upload_type : "upload" , } ; return obj ; } ) . filter ( ( { payload : { hash } } ) => ! uploadedList . includes ( hash as any ) ) ; ele . already = uploadedList . length ; ele . chunks = data ; /**
         *  此地需要判断一种情况
         *  当切片通过线上已上传的切片过滤后没有可传的切片时
         *  此时直接进行合并操作
         * */ if ( fileChunkList . length === uploadedList . length ) { let node = { type : null , size : null , payload : { type : ele . type , hash : ele . payload . hash as any , chunk : null , sort_index : ele . payload . sort_index , } , } ; this . mergeRequest ( node ) ; return ; } ele . chunksLength = fileChunkList . length ; ele . percentage = ( ele . already / ele . chunksLength ) * 100 ; this . push ( ele ) ; } async mergeRequest ( node ) { let data = { fileHash : node . payload . hash , type : node . payload . type , upload_type : "merge" , sort_index : node . payload . sort_index , mimeType : node . mimeType , } ; ! node . payload . sort_index && delete data . sort_index ; return this . http . post ( "http://127.0.0.1:3001/upload" , data ) . toPromise ( ) ; } param base64 doWorkerGetBlob ( base64 ) { let obs = new Observable ( ( observer ) => { const worker = new Worker ( "../assets/js/dataURItoBlob.js" ) ; worker . postMessage ( { res : base64 } ) ; worker . onmessage = ( e ) => { observer . next ( e ) ; } ; } ) ; return obs ; } param fileHash param filePath async verifyUpload ( fileHash , filePath , params ) { const data = await this . request . request ( { url : `${ environment . apiUrl }/upload` , headers : { "content-type" : "application/json" , } , data : JSON . stringify ( { fileHash : fileHash , filePath : filePath , type : params . type , upload_type : "fileVerify" , } ) , } ) ; return data ; } cancel ( ) { this . suspend ( ) ; this . queue . shift ( ) ; this . size && this . restart ( ) ; } } export function HashCode ( str ) { var hash = 0 , i , chr ; if ( str . length === 0 ) return hash ; for ( i = 0 ; i < str . length ; i ++ ) { chr = str . charCodeAt ( i ) ; hash = ( hash << 5 ) - hash + chr ; hash |= 0 ; } return hash ; }