git ssb

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