git ssb

1
var nest = require('depnest')
2
var pull = require('pull-stream')
3
var MutantPullReduce = require('mutant-pull-reduce')
4
5
exports.needs = nest({
6
  'sbot.pull.backlinks': 'first'
7
})
8
9
exports.gives = nest("backlinks.obs.filter", true)
10
11
/**
12
 * sbot.obs.filter returns an observable list of messages that link
13
 * back to the message with the given message ID (@id). Only messages that
14
 * pass the filter are added to the list.
15
 *
16
 * When a message arrives, if a filter function is given in the options (opts.filter)
17
 * and passing it to the filter function does not result in it returning
18
 * 'true' the message is not added to the observable list.
19
 *
20
 * A 'sync' observable property is also added to the returned observable
21
 * which is 'true' when all previously seen messages are caught up with.
22
 *
23
 * Note: Unlike backlinks.obs.for this does not cache the observable for
24
 * callers that supply the same arguments.
25
 */
26
exports.create = function(api) {
27
  function pullFilterReduceObs(id, opts) {
28
    if (!id || typeof(id) !== "string") {
29
      throw new Error("id must be a string.")
30
    }
31
32
    var sbotFilter = {
33
      $filter: {
34
        dest: id
35
      }
36
    }
37
38
    var msgBacklinks = api.sbot.pull.backlinks({
39
      query: [sbotFilter],
40
      sync: true
41
    })
42
43
    // If a filter function is supplied in the options, we use it to filter
44
    // the links stream, otherwise we use all the messages from the stream
45
    var filterFunction = opts && opts.filter ? opts.filter : () => true
46
47
    var filteredBacklinks = pull(
48
      msgBacklinks,
49
      pull.filter(filterFunction)
50
    )
51
52
    var backlinksObs = MutantPullReduce(filteredBacklinks, (state, msg) => {
53
      state.push(msg)
54
      return state;
55
    }, {
56
      startValue: [],
57
      nextTick: true,
58
      sync: true
59
    })
60
61
    return backlinksObs;
62
  }
63
64
  return nest({
65
    "backlinks.obs.filter": (id, opts) => pullFilterReduceObs(id, opts)
66
  })
67
}
68