git ssb

1
const Value = require('mutant/value')
2
const bus = require('./bus')
3
const ref = require('ssb-ref')
4
const traverse = require('traverse')
5
const pull = require('pull-stream')
6
7
function rateLimit(ms, f) {
8
  let lastTime = 0
9
  let timer
10
11
  return function fun() {
12
    if (timer) clearTimeout(timer)
13
    const now = Date.now()
14
    if (now - lastTime > ms) {
15
      f()
16
      lastTime = now
17
    } else {
18
      timer = setTimeout(fun, ms)
19
    }
20
  }
21
}
22
23
module.exports = function Blobs(ssb, blobs, blobBytes, blobRefs, blobsPresent) {
24
  let refs = 0
25
  let present = 0
26
  let totalSize = 0
27
  let knownBlobs = new Set() 
28
  let sizeObs = {}
29
  let synced = false
30
  let refList = []
31
32
  const updateStuff = rateLimit(230, ()=>{ 
33
    if (!synced) return
34
    blobBytes.set(totalSize)
35
    blobsPresent.set(present)
36
    blobRefs.set(refs)
37
38
    if (!window.frameElement) {
39
      const event = new CustomEvent('blobs-progress', { detail: present / refs }); 
40
      document.body.dispatchEvent(event)
41
    }
42
43
  })
44
45
  let getSize = function(blob) {
46
    ssb.blobs.size(blob, (err, size) => {
47
      if (err) return sizeObs[blob].set(err.message)
48
      sizeObs[blob].set(size || 'zero')
49
      totalSize += size
50
      ++present
51
      updateStuff()
52
    })
53
  }
54
55
  let blobsLs
56
  return pull(
57
    pull.asyncMap( (kv, cb) => {
58
      if (typeof blobsLs === 'undefined') {
59
        pull(
60
          ssb.blobs.ls(),
61
          pull.collect( (err, ls) =>{
62
            if (err) console.error(err)
63
            blobsLs = ls || []
64
            cb(null, kv)
65
          } )
66
        )
67
      } else {
68
        cb(null, kv)
69
      }
70
    }),
71
    pull.through( processBlobReferences )
72
  )
73
74
  function processBlobReferences(kv) {
75
  
76
    if (kv.sync) {
77
      if (!synced) {
78
        synced = true
79
        updateStuff()
80
        blobs.set(refList)
81
        console.log('BLOBS synced')
82
      }
83
      return
84
    }
85
    
86
    traverse(kv.value.content || {}).forEach( function(v) {
87
      if (ref.isBlob(v)) {
88
        let blob = v
89
        let newBlob = !knownBlobs.has(blob)
90
        if (newBlob) {
91
          knownBlobs.add(blob)
92
          refs++
93
          updateStuff()
94
          
95
          if (!sizeObs[blob]) {
96
            sizeObs[blob] = Value('...')
97
98
            if (blobsLs.includes(blob)) {
99
              getSize(blob)
100
            } else {
101
              sizeObs[blob].set('wanted ...')
102
              ssb.blobs.want(blob, err => {
103
                if (err) return sizeObs[blob].set(err.message)
104
                getSize(blob)
105
              })
106
            }
107
108
          }
109
        }
110
111
        refList.push({
112
          id: blob,
113
          size: sizeObs[blob],
114
          neededBy: {
115
            key: kv.key,
116
            type: kv.value.content.type,
117
            name: kv.value.content.name,
118
            path: this.path.join('.')
119
          }
120
        })
121
        if (synced) blobs.set(refList)
122
      }
123
    })
124
  }
125
}
126