git ssb

1
'use strict'
2
var pull = require('pull-stream')
3
var Notify = require('pull-notify')
4
var mdm = require('mdmanifest')
5
var valid = require('scuttlebot/lib/validators')
6
var apidoc = require('scuttlebot/lib/apidocs').gossip
7
var u = require('scuttlebot/lib/util')
8
var ref = require('ssb-ref')
9
var ping = require('pull-ping')
10
var stats = require('statistics')
11
var Schedule = require('./schedule')
12
var Init = require('./init')
13
var AtomicFile = require('atomic-file')
14
var path = require('path')
15
var deepEqual = require('deep-equal')
16
17
function isFunction (f) {
18
  return 'function' === typeof f
19
}
20
21
function stringify(peer) {
22
  return [peer.host, peer.port, peer.key].join(':')
23
}
24
25
/*
26
Peers : [{
27
  key: id,
28
  host: ip,
29
  port: int,
30
  //to be backwards compatible with patchwork...
31
  announcers: {length: int}
32
  source: 'pub'|'manual'|'local'
33
}]
34
*/
35
36
37
module.exports = {
38
  name: 'gossip',
39
  version: '1.0.0',
40
  manifest: mdm.manifest(apidoc),
41
  permissions: {
42
    anonymous: {allow: ['ping']}
43
  },
44
  init: function (server, config) {
45
    var notify = Notify()
46
    var conf = config.gossip || {}
47
    var home = ref.parseAddress(server.getAddress())
48
49
    var stateFile = AtomicFile(path.join(config.path, 'gossip.json'))
50
51
    //Known Peers
52
    var peers = []
53
54
    function getPeer(id) {
55
      return u.find(peers, function (e) {
56
        return e && e.key === id
57
      })
58
    }
59
60
    var timer_ping = 5*6e4
61
62
    var gossip = {
63
      wakeup: 0,
64
      peers: function () {
65
        return peers
66
      },
67
      get: function (addr) {
68
        addr = ref.parseAddress(addr)
69
        return u.find(peers, function (a) {
70
          return (
71
            addr.port === a.port
72
            && addr.host === a.host
73
            && addr.key === a.key
74
          )
75
        })
76
      },
77
      connect: valid.async(function (addr, cb) {
78
        addr = ref.parseAddress(addr)
79
        if (!addr || typeof addr != 'object')
80
          return cb(new Error('first param must be an address'))
81
82
        if(!addr.key) return cb(new Error('address must have ed25519 key'))
83
        // add peer to the table, incase it isn't already.
84
        gossip.add(addr, 'manual')
85
        var p = gossip.get(addr)
86
        if(!p) return cb()
87
88
        p.stateChange = Date.now()
89
        p.state = 'connecting'
90
        server.connect(p, function (err, rpc) {
91
          if (err) {
92
            p.state = undefined
93
            p.failure = (p.failure || 0) + 1
94
            p.stateChange = Date.now()
95
            notify({ type: 'connect-failure', peer: p })
96
            server.emit('log:info', ['SBOT', p.host+':'+p.port+p.key, 'connection failed', err.message || err])
97
            p.duration = stats(p.duration, 0)
98
            return (cb && cb(err))
99
          }
100
          else {
101
            p.state = 'connected'
102
            p.failure = 0
103
          }
104
          cb && cb(null, rpc)
105
        })
106
107
      }, 'string|object'),
108
109
      disconnect: valid.async(function (addr, cb) {
110
        var peer = this.get(addr)
111
112
        peer.state = 'disconnecting'
113
        peer.stateChange = Date.now()
114
        if(!peer || !peer.disconnect) cb && cb()
115
        else peer.disconnect(true, function (err) {
116
          peer.stateChange = Date.now()
117
        })
118
119
      }, 'string|object'),
120
121
      changes: function () {
122
        return notify.listen()
123
      },
124
      //add an address to the peer table.
125
      add: valid.sync(function (addr, source) {
126
        addr = ref.parseAddress(addr)
127
        if(!ref.isAddress(addr))
128
          throw new Error('not a valid address:' + JSON.stringify(addr))
129
        // check that this is a valid address, and not pointing at self.
130
131
        if(addr.key === home.key) return
132
        if(addr.host === home.host && addr.port === home.port) return
133
134
        var f = gossip.get(addr)
135
136
        if(!f) {
137
          // new peer
138
          addr.source = source
139
          addr.announcers = 1
140
          addr.duration = addr.duration || null
141
          peers.push(addr)
142
          notify({ type: 'discover', peer: addr, source: source || 'manual' })
143
          return addr
144
        } else if (source === 'friends' || source === 'local') {
145
          // this peer is a friend or local, override old source to prioritize gossip
146
          f.source = source
147
        }
148
        //don't count local over and over
149
        else if(f.source != 'local')
150
          f.announcers ++
151
152
        return f
153
      }, 'string|object', 'string?'),
154
      delete: function (addr) {
155
        var peer = gossip.get(addr)
156
        var index = peers.indexOf(peer)
157
        if (~index) {
158
          peers.splice(index, 1)
159
        }
160
      },
161
      ping: function (opts) {
162
        var timeout = config.timers && config.timers.ping || 5*60e3
163
        //between 10 seconds and 30 minutes, default 5 min
164
        timeout = Math.max(10e3, Math.min(timeout, 30*60e3))
165
        return ping({timeout: timeout})
166
      },
167
      reconnect: function () {
168
        for(var id in server.peers)
169
          if(id !== server.id) //don't disconnect local client
170
            server.peers[id].forEach(function (peer) {
171
              peer.close(true)
172
            })
173
        return gossip.wakeup = Date.now()
174
      }
175
    }
176
177
    Schedule (gossip, config, server)
178
    Init (gossip, config, server)
179
    //get current state
180
181
    server.on('rpc:connect', function (rpc, isClient) {
182
      var peer = getPeer(rpc.id)
183
      //don't track clients that connect, but arn't considered peers.
184
      //maybe we should though?
185
      if(!peer) return
186
      console.log('Connected', stringify(peer))
187
      //means that we have created this connection, not received it.
188
      peer.client = !!isClient
189
      peer.state = 'connected'
190
      peer.stateChange = Date.now()
191
      peer.disconnect = function (err, cb) {
192
        if(isFunction(err)) cb = err, err = null
193
        rpc.close(err, cb)
194
      }
195
196
      if(isClient) {
197
        //default ping is 5 minutes...
198
        var pp = ping({serve: true, timeout: timer_ping}, function (_) {})
199
        peer.ping = {rtt: pp.rtt, skew: pp.skew}
200
        pull(
201
          pp,
202
          rpc.gossip.ping({timeout: timer_ping}, function (err) {
203
            if(err.name === 'TypeError') peer.ping.fail = true
204
          }),
205
          pp
206
        )
207
      }
208
209
      rpc.on('closed', function () {
210
        console.log('Disconnected', stringify(peer))
211
        //track whether we have successfully connected.
212
        //or how many failures there have been.
213
        var since = peer.stateChange
214
        peer.stateChange = Date.now()
215
//        if(peer.state === 'connected') //may be "disconnecting"
216
        peer.duration = stats(peer.duration, peer.stateChange - since)
217
//        console.log(peer.duration)
218
        peer.state = undefined
219
        notify({ type: 'disconnect', peer: peer })
220
        server.emit('log:info', ['SBOT', rpc.id, 'disconnect'])
221
      })
222
223
      notify({ type: 'connect', peer: peer })
224
    })
225
226
    var last
227
    stateFile.get(function (err, ary) {
228
      last = ary || []
229
      if(Array.isArray(ary))
230
        ary.forEach(function (v) {
231
          delete v.state
232
          // don't add local peers (wait to rediscover)
233
          if(v.source !== 'local') {
234
            gossip.add(v, 'stored')
235
          }
236
        })
237
    })
238
239
    var int = setInterval(function () {
240
      var copy = JSON.parse(JSON.stringify(peers))
241
      copy.forEach(function (e) {
242
        delete e.state
243
      })
244
      if(deepEqual(copy, last)) return
245
      last = copy
246
      stateFile.set(copy, function(err) {
247
        if (err) console.log(err)
248
      })
249
    }, 10*1000)
250
251
    if(int.unref) int.unref()
252
253
    return gossip
254
  }
255
}
256