client/src/assets/player/peertube-chunk-store.ts

   1 // From https://github.com/MinEduTDF/idb-chunk-store
   2 // We use temporary IndexDB (all data are removed on destroy) to avoid RAM issues
   3 // Thanks @santiagogil and @Feross
   4
   5 import { EventEmitter } from 'events'
   6 import Dexie from 'dexie'
   7
   8 class ChunkDatabase extends Dexie {
   9   chunks: Dexie.Table<{ id: number, buf: Buffer }, number>
  10
  11   constructor (dbname: string) {
  12     super(dbname)
  13
  14     this.version(1).stores({
  15       chunks: 'id'
  16     })
  17   }
  18 }
  19
  20 class ExpirationDatabase extends Dexie {
  21   databases: Dexie.Table<{ name: string, expiration: number }, number>
  22
  23   constructor () {
  24     super('webtorrent-expiration')
  25
  26     this.version(1).stores({
  27       databases: 'name,expiration'
  28     })
  29   }
  30 }
  31
  32 export class PeertubeChunkStore extends EventEmitter {
  33   private static readonly BUFFERING_PUT_MS = 1000
  34   private static readonly CLEANER_INTERVAL_MS = 1000 * 60 // 1 minute
  35   private static readonly CLEANER_EXPIRATION_MS = 1000 * 60 * 5 // 5 minutes
  36
  37   chunkLength: number
  38
  39   private pendingPut: { id: number, buf: Buffer, cb: Function }[] = []
  40   // If the store is full
  41   private memoryChunks: { [ id: number ]: Buffer | true } = {}
  42   private databaseName: string
  43   private putBulkTimeout
  44   private cleanerInterval
  45   private db: ChunkDatabase
  46   private expirationDB: ExpirationDatabase
  47   private readonly length: number
  48   private readonly lastChunkLength: number
  49   private readonly lastChunkIndex: number
  50
  51   constructor (chunkLength: number, opts) {
  52     super()
  53
  54     this.databaseName = 'webtorrent-chunks-'
  55
  56     if (!opts) opts = {}
  57     if (opts.torrent && opts.torrent.infoHash) this.databaseName += opts.torrent.infoHash
  58     else this.databaseName += '-default'
  59
  60     this.setMaxListeners(100)
  61
  62     this.chunkLength = Number(chunkLength)
  63     if (!this.chunkLength) throw new Error('First argument must be a chunk length')
  64
  65     this.length = Number(opts.length) || Infinity
  66
  67     if (this.length !== Infinity) {
  68       this.lastChunkLength = (this.length % this.chunkLength) || this.chunkLength
  69       this.lastChunkIndex = Math.ceil(this.length / this.chunkLength) - 1
  70     }
  71
  72     this.db = new ChunkDatabase(this.databaseName)
  73     // Track databases that expired
  74     this.expirationDB = new ExpirationDatabase()
  75
  76     this.runCleaner()
  77   }
  78
  79   put (index: number, buf: Buffer, cb: Function) {
  80     const isLastChunk = (index === this.lastChunkIndex)
  81     if (isLastChunk && buf.length !== this.lastChunkLength) {
  82       return this.nextTick(cb, new Error('Last chunk length must be ' + this.lastChunkLength))
  83     }
  84     if (!isLastChunk && buf.length !== this.chunkLength) {
  85       return this.nextTick(cb, new Error('Chunk length must be ' + this.chunkLength))
  86     }
  87
  88     // Specify we have this chunk
  89     this.memoryChunks[index] = true
  90
  91     // Add it to the pending put
  92     this.pendingPut.push({ id: index, buf, cb })
  93     // If it's already planned, return
  94     if (this.putBulkTimeout) return
  95
  96     // Plan a future bulk insert
  97     this.putBulkTimeout = setTimeout(async () => {
  98       const processing = this.pendingPut
  99       this.pendingPut = []
 100       this.putBulkTimeout = undefined
 101
 102       try {
 103         await this.db.transaction('rw', this.db.chunks, () => {
 104           return this.db.chunks.bulkPut(processing.map(p => ({ id: p.id, buf: p.buf })))
 105         })
 106       } catch (err) {
 107         console.log('Cannot bulk insert chunks. Store them in memory.', { err })
 108
 109         processing.forEach(p => this.memoryChunks[ p.id ] = p.buf)
 110       } finally {
 111         processing.forEach(p => p.cb())
 112       }
 113     }, PeertubeChunkStore.BUFFERING_PUT_MS)
 114   }
 115
 116   get (index: number, opts, cb) {
 117     if (typeof opts === 'function') return this.get(index, null, opts)
 118
 119     // IndexDB could be slow, use our memory index first
 120     const memoryChunk = this.memoryChunks[index]
 121     if (memoryChunk === undefined) return cb(null, new Buffer(0))
 122     // Chunk in memory
 123     if (memoryChunk !== true) return cb(null, memoryChunk)
 124
 125     // Chunk in store
 126     this.db.transaction('r', this.db.chunks, async () => {
 127       const result = await this.db.chunks.get({ id: index })
 128       if (result === undefined) return cb(null, new Buffer(0))
 129
 130       const buf = result.buf
 131       if (!opts) return this.nextTick(cb, null, buf)
 132
 133       const offset = opts.offset || 0
 134       const len = opts.length || (buf.length - offset)
 135       return cb(null, buf.slice(offset, len + offset))
 136     })
 137     .catch(err => {
 138       console.error(err)
 139       return cb(err)
 140     })
 141   }
 142
 143   close (db) {
 144     return this.destroy(db)
 145   }
 146
 147   async destroy (cb) {
 148     try {
 149       if (this.pendingPut) {
 150         clearTimeout(this.putBulkTimeout)
 151         this.pendingPut = null
 152       }
 153       if (this.cleanerInterval) {
 154         clearInterval(this.cleanerInterval)
 155         this.cleanerInterval = null
 156       }
 157
 158       if (this.expirationDB) {
 159         await this.expirationDB.close()
 160         this.expirationDB = null
 161       }
 162
 163       if (this.db) {
 164         console.log('Destroying IndexDB database %s.', this.databaseName)
 165         await this.db.close()
 166         await this.db.delete()
 167       }
 168
 169       return cb()
 170     } catch (err) {
 171       console.error('Cannot destroy peertube chunk store.', err)
 172       return cb(err)
 173     }
 174   }
 175
 176   private runCleaner () {
 177     this.checkExpiration()
 178
 179     this.cleanerInterval = setInterval(async () => {
 180       this.checkExpiration()
 181     }, PeertubeChunkStore.CLEANER_INTERVAL_MS)
 182   }
 183
 184   private checkExpiration () {
 185     this.expirationDB.transaction('rw', this.expirationDB.databases, async () => {
 186       // Update our database expiration since we are alive
 187       await this.expirationDB.databases.put({
 188         name: this.databaseName,
 189         expiration: new Date().getTime() + PeertubeChunkStore.CLEANER_EXPIRATION_MS
 190       })
 191
 192       const now = new Date().getTime()
 193       const databasesToDeleteInfo = await this.expirationDB.databases.where('expiration').below(now).toArray()
 194
 195       for (const databaseToDeleteInfo of databasesToDeleteInfo) {
 196         await this.dropDatabase(databaseToDeleteInfo.name)
 197
 198         await this.expirationDB.databases.where({ name: databaseToDeleteInfo.name }).delete()
 199       }
 200     }).catch(err => console.error('Cannot check expiration.', err))
 201   }
 202
 203   private dropDatabase (databaseName: string) {
 204     const dbToDelete = new ChunkDatabase(databaseName)
 205
 206     console.log('Deleting %s.', databaseName)
 207     return dbToDelete.delete()
 208       .catch(err => console.error('Cannot delete %s.', databaseName))
 209   }
 210
 211   private nextTick (cb, err, val?) {
 212     process.nextTick(() => cb(err, val), undefined)
 213   }
 214 }