git ssb

1
var pull       = require('pull-stream')
2
var muxrpc     = require('muxrpc')
3
var address    = require('ssb-address')
4
var ws         = require('pull-ws-server')
5
var Serializer = require('pull-serializer')
6
var ssbKeys    = require('ssb-keys')
7
var loadManf   = require('ssb-manifest/load')
8
var createMsg  = require('secure-scuttlebutt/message')(require('secure-scuttlebutt/defaults'))
9
10
function isFunction (f) {
11
  return 'function' === typeof f
12
}
13
14
function throwIfError(err) {
15
  if(err) throw err
16
}
17
18
module.exports = function (keys, config) {
19
  var manifest
20
  //if we are in the browser
21
  config = config || {}
22
  config.host = config.host || 'localhost'
23
  var client = muxrpc(loadManf(config), false, serialize)()
24
  client.keys = keys
25
26
  var wsStream
27
  var rpcStream
28
29
  client.connect = function(addr, cb) {
30
    if(isFunction(addr))
31
      cb = addr, addr = null
32
33
    addr = address(addr || config)
34
    if (wsStream) {
35
      wsStream.close()
36
      client._emit('reconnecting')
37
    }
38
39
    var called = false
40
41
    client.addr = addr
42
43
    //if auth is not the first method called,
44
    //then the other methods will get auth errors.
45
    //since rpc calls are queued, we can just do it here.
46
    client.auth(function (err, authed) {
47
      if (err)
48
        client._emit('error', err)
49
      else
50
        client._emit('authed', authed)
51
      if(called) return
52
      called = true; cb && cb(err, authed)
53
    })
54
55
    wsStream = ws.connect(addr, {
56
      onOpen: function() {
57
        client._emit('connect')
58
        //cb is called after auth, just above
59
      },
60
      onClose: function() {
61
        client._emit('close')
62
        //rpcStream will detect close on it's own.
63
        if(called) return
64
        called = true; cb && cb(err, authed)
65
      }
66
    })
67
68
    rpcStream = client.createStream()
69
    pull(wsStream, rpcStream, wsStream)
70
71
    return client
72
  }
73
74
  client.close = function(cb) {
75
    wsStream.close()
76
    rpcStream.close(cb ? cb : throwIfError)
77
    return client
78
  }
79
80
  client.reconnect = function(opts) {
81
    opts = opts || {}
82
    client.close(function() {
83
      if (opts.wait)
84
        setTimeout(client.connect.bind(client, client.addr), opts.wait)
85
      else
86
        client.connect(client.addr)
87
    })
88
    return client
89
  }
90
91
  client.publish = function (content, cb) {
92
    client.getLatest(client.keys.id, function (err, prev) {
93
      if (!prev) {
94
        var init = createMsg(client.keys, null, { type: 'init', public: client.keys.public }, null)
95
        client.add(init, function (err, res) {
96
          if (err)
97
            return cb(err)
98
          prev = res.value
99
          next()
100
        })
101
      } else
102
        next()
103
104
      function next () {
105
        var msg = createMsg(client.keys, null, content, prev||null)
106
        client.add(msg, cb)
107
      }
108
    })
109
    return client
110
  }
111
112
  var auth_ = client.auth
113
  client.auth = function (cb) {
114
    var authReq = ssbKeys.signObj(client.keys, {
115
      role: 'client',
116
      ts: Date.now(),
117
      public: client.keys.public
118
    })
119
    auth_.call(client, authReq, cb)
120
    return client
121
  }
122
123
  return client
124
}
125
126
function serialize (stream) {
127
  return Serializer(stream, JSON, {split: '\n\n'})
128
}
129