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