git ssb

1
var pull = require('pull-stream')
2
var cat = require('pull-cat')
3
var loop = require('looper')
4
var multicb = require('multicb')
5
var crypto = require('crypto')
6
var skipFooter = require('pull-skip-footer')
7
var packidx = require('pull-git-packidx-parser')
8
var blockFilter = require('pull-block-filter')
9
10
function packHeader(numObjects) {
11
  var header = new Buffer(12)
12
  header.write('PACK')
13
  header.writeUInt32BE(2, 4)
14
  header.writeUInt32BE(numObjects, 8)
15
  return header
16
}
17
18
function forEachAsync(arr, fn, cb) {
19
  var i = 0
20
  loop(function (next) {
21
    if (i >= arr.length) return cb && cb()
22
    fn(arr[i++], function (err) {
23
      if (err) return cb && cb(err)
24
      next()
25
    })
26
  })
27
}
28
29
function compareByOffset(a, b) {
30
  return a.offset - b.offset
31
}
32
33
function dedupPacks(packs, cb) {
34
  var seen = {}
35
  var numObjects = 0
36
  forEachAsync(packs, function (pack, cb) {
37
    return pull(pack.readIdx, packidx(function (err, idx) {
38
      if (err) return cb(err)
39
      var blocks = []
40
      var lastBlock
41
      offset = 0
42
      var objs = idx.objects.sort(compareByOffset)
43
      for (var i = 0; i < objs.length; i++) {
44
        var obj = objs[i]
45
        var id = obj.oid.toString('hex')
46
        if (seen[id]) continue
47
        seen[id] = true
48
        numObjects++
49
        if (obj.offset > offset) {
50
          blocks.push(lastBlock = {skip: obj.offset - offset, length: 0})
51
          offset = obj.offset
52
        } else if (obj.offset < offset) {
53
          return cb(new Error('bad offset'))
54
        }
55
        var len = obj.next ? obj.next.offset - obj.offset : Infinity
56
        lastBlock.length += len
57
        offset += len
58
      }
59
      pack.read = pull(
60
        pack.read,
61
        skipFooter(20),
62
        blockFilter(pull.values(blocks))
63
      )
64
      cb()
65
    }))
66
  }, function (err) {
67
    cb(err, numObjects)
68
  })
69
}
70
71
function closePacks(packs, cb) {
72
  forEachAsync(packs, function (pack, cb) {
73
    pack.read(true, cb)
74
  }, cb)
75
}
76
77
module.exports = function concatPacks(packs) {
78
  /* packs: [{read: source, readIdx: source}] */
79
  if (packs.length === 1) return packs[0].read
80
81
  var checksum = crypto.createHash('sha1')
82
  var packI = 0
83
  var state = 'begin'
84
85
  return function next(end, cb) {
86
    switch (state) {
87
      case 'begin':
88
        if (end) return closePacks(cb)
89
        return dedupPacks(packs, function (err, numObjects) {
90
          if (err) return cb(err)
91
          var header = packHeader(numObjects)
92
          checksum.update(header)
93
          state = 'payload'
94
          cb(null, header)
95
        })
96
97
      case 'payload':
98
        if (end) return closePacks(cb)
99
        if (packI >= packs.length) {
100
          state = 'end' 
101
          return cb(null, checksum.digest())
102
        }
103
        var pack = packs[packI]
104
        return pack.read(null, function (err, data) {
105
          if (err === true) {
106
            packI++
107
            return next(null, cb)
108
          }
109
          if (err) return cb(err)
110
          checksum.update(data)
111
          cb(null, data)
112
        })
113
114
      case 'end':
115
        return cb(true)
116
    }
117
  }
118
}
119