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