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 skipHeader(len) {
31
  return function (read) {
32
    return function (end, cb) {
33
      if (end || len <= 0) read(end, cb)
34
      else read(null, function next(end, data) {
35
        if (end) return cb(end)
36
        var _len = len
37
        len -= data.length
38
        if (len > 0) read(null, next)
39
        else cb(null, data.slice(_len))
40
      })
41
    }
42
  }
43
}
44
45
function compareByOffset(a, b) {
46
  return a.offset - b.offset
47
}
48
49
function dedupPacks(packs, cb) {
50
  var seen = {}
51
  var numObjects = 0
52
  forEachAsync(packs, function (pack, cb) {
53
    return pull(pack.readIdx, packidx(function (err, idx) {
54
      if (err) return cb(err)
55
      var blocks = []
56
      var lastBlock
57
      offset = 0
58
      var objs = idx.objects.sort(compareByOffset)
59
      for (var i = 0; i < objs.length; i++) {
60
        var obj = objs[i]
61
        var id = obj.oid.toString('hex')
62
        if (id === '00947e10295e018fc71cf9c264ea5f341260f9b2') throw 1
63
        if (seen[id]) continue
64
        seen[id] = true
65
        numObjects++
66
        if (obj.offset > offset) {
67
          blocks.push(lastBlock = {skip: obj.offset - offset, length: 0})
68
          offset = obj.offset
69
        } else if (obj.offset < offset) {
70
          return cb(new Error('bad offset'))
71
        }
72
        var len = obj.next ? obj.next.offset - obj.offset : Infinity
73
        lastBlock.length += len
74
        offset += len
75
      }
76
      pack.read = pull(
77
        pack.read,
78
        skipFooter(20),
79
        blockFilter(pull.values(blocks))
80
      )
81
      cb()
82
		}))
83
  }, function (err) {
84
    cb(err, numObjects)
85
  })
86
}
87
88
function closePacks(packs, cb) {
89
  forEachAsync(packs, function (pack, cb) {
90
    pack.read(true, cb)
91
  }, cb)
92
}
93
94
module.exports = function concatPacks(packs) {
95
  /* packs: [{read: source, readIdx: source}] */
96
  if (packs.length === 1) return packs[0].read
97
98
  var checksum = crypto.createHash('sha1')
99
  var packI = 0
100
  var state = 'begin'
101
102
  return function next(end, cb) {
103
    switch (state) {
104
      case 'begin':
105
        if (end) return closePacks(cb)
106
        return dedupPacks(packs, function (err, numObjects) {
107
          if (err) return cb(err)
108
          var header = packHeader(numObjects)
109
          checksum.update(header)
110
          state = 'payload'
111
          cb(null, header)
112
        })
113
114
      case 'payload':
115
        if (end) return closePacks(cb)
116
        if (packI >= packs.length) {
117
          state = 'end' 
118
          return cb(null, checksum.digest())
119
        }
120
        var pack = packs[packI]
121
        return pack.read(null, function (err, data) {
122
          if (err === true) {
123
            packI++
124
            return next(null, cb)
125
          }
126
          if (err) return cb(err)
127
          checksum.update(data)
128
          cb(null, data)
129
        })
130
131
      case 'end':
132
        return cb(true)
133
    }
134
  }
135
}
136