{"size": 1277, "ext": "js", "max_stars_repo_path": "packages/@lyra/client/src/data/listen.js", "max_stars_repo_name": "benjamespoet/vega", "max_stars_repo_stars_event_min_datetime": "2018-08-15T04:41:24.000Z", "max_stars_repo_stars_event_max_datetime": "2022-02-02T15:41:52.000Z", "max_issues_repo_path": "packages/@lyra/client/src/data/listen.js", "max_issues_repo_name": "benjamespoet/vega", "max_issues_repo_issues_event_min_datetime": "2018-01-18T10:35:20.000Z", "max_issues_repo_issues_event_max_datetime": "2022-03-20T03:27:45.000Z", "max_forks_repo_path": "packages/@lyra/client/src/data/listen.js", "max_forks_repo_name": "benjamespoet/vega", "max_forks_repo_forks_event_min_datetime": "2019-05-21T14:25:04.000Z", "max_forks_repo_forks_event_max_datetime": "2020-05-09T17:58:30.000Z"}
{"max_stars_count": 15, "max_issues_count": 219, "max_forks_count": 5, "avg_line_length": 34.5135135135, "max_line_length": 79, "alphanum_fraction": 0.692247455}
const ws = require('ws')
const {filter, tap} = require('rxjs/operators')
const pick = require('../util/pick')
const defaults = require('../util/defaults')
const createWsClient = require('./wsClient')

const possibleOptions = ['includePreviousRevision', 'includeResult']
const defaultOptions = {
  includeResult: true
}

const clients = {}

module.exports = function listen(query, params, opts = {}) {
  const {url, token} = this.clientConfig
  const uri = `${url}${this.getDataUrl('channel')}`.replace(/^http/, 'ws')
  const key = `${uri}${token ? token.slice(0, 5) : ''}`
  const headers = token ? {Authorization: `Bearer ${token}`} : undefined

  // @todo improve this - observables?
  let client = clients[key]
  if (!client) {
    clients[key] = createWsClient(ws, uri, {headers})
    client = clients[key]
  }

  const options = defaults(opts, defaultOptions)
  const listenOpts = pick(options, possibleOptions)
  const {includePreviousRevision, includeResult} = listenOpts
  const listenFor = options.events ? options.events : ['mutation']
  const shouldEmitReconnect = listenFor.indexOf('reconnect') !== -1 // @todo

  return client
    .request('listen', {query, params, includePreviousRevision, includeResult})
    .pipe(filter(msg => listenFor.includes(msg.type)))
}
