<commit_before>import { List, Map, Set } from 'immutable'
import { EventEmitter } from 'events'

function defaultDataGetter() {}

export class UserSocketGroup extends EventEmitter {
  constructor(nydus, session, initSocket) {
    super()
    this.nydus = nydus
    this.name = session.userName
    this.session = session
    this.sockets = new Set()
    this.subscriptions = new Map()

    if (initSocket) {
      this.add(initSocket)
    }
  }

  add(socket) {
    const newSockets = this.sockets.add(socket)
    if (newSockets !== this.sockets) {
      this.sockets = newSockets
      socket.once('close', () => this.delete(socket))
      this._applySubscriptions(socket)
      this.emit('connection', this, socket)
    }
  }

  delete(socket) {
    this.sockets = this.sockets.delete(socket)
    if (this.sockets.isEmpty()) {
      this.emit('close', this)
      this._applyCleanups()
    }
  }

  // Adds a subscription to all sockets for this user, including any sockets that may connect
  // after this. `initialDataGetter` should either be null/undefined, or a function(user, socket)
  // that returns the initialData to use for a subscribe call. `cleanup` should either be
  // null/undefined, or a function(user) that will be called when the user has disconnected
  // completely.
  subscribe(path, initialDataGetter, cleanup) {
    if (this.subscriptions.has(path)) {
      throw new Error('duplicate persistent subscription: ' + path)
    }

    this.subscriptions = this.subscriptions.set(path, {
      getter: initialDataGetter || defaultDataGetter,
      cleanup,
    })
    for (const socket of this.sockets) {
      this.nydus.subscribeClient(socket, path, initialDataGetter(this, socket))
    }
  }

  _applySubscriptions(socket) {
    for (const [path, { getter }] of this.subscriptions.entries()) {
      this.nydus.subscribeClient(socket, path, getter(this, socket))
    }

    // Give the client a message so they know we're done subscribing them to things
    this.nydus.subscribeClient(socket, this.getUserPath(), { type: 'subscribed' })
  }

  _applyCleanups() {
    for (const [/* path */, { cleanup }] of this.subscriptions.entries()) {
      if (cleanup) cleanup(this)
    }
  }

  unsubscribe(path) {
    const updated = this.subscriptions.delete(path)
    if (updated === this.subscriptions) return
<commit_msg>Fix subscribe method to not crash if initialDataGetter is not supplied. (#115)<commit_after>import { List, Map, Set } from 'immutable'
import { EventEmitter } from 'events'

function defaultDataGetter() {}

export class UserSocketGroup extends EventEmitter {
  constructor(nydus, session, initSocket) {
    super()
    this.nydus = nydus
    this.name = session.userName
    this.session = session
    this.sockets = new Set()
    this.subscriptions = new Map()

    if (initSocket) {
      this.add(initSocket)
    }
  }

  add(socket) {
    const newSockets = this.sockets.add(socket)
    if (newSockets !== this.sockets) {
      this.sockets = newSockets
      socket.once('close', () => this.delete(socket))
      this._applySubscriptions(socket)
      this.emit('connection', this, socket)
    }
  }

  delete(socket) {
    this.sockets = this.sockets.delete(socket)
    if (this.sockets.isEmpty()) {
      this.emit('close', this)
      this._applyCleanups()
    }
  }

  // Adds a subscription to all sockets for this user, including any sockets that may connect
  // after this. `initialDataGetter` should either be undefined, or a function(user, socket)
  // that returns the initialData to use for a subscribe call. `cleanup` should either be
  // null/undefined, or a function(user) that will be called when the user has disconnected
  // completely.
  subscribe(path, initialDataGetter = defaultDataGetter, cleanup) {
    if (this.subscriptions.has(path)) {
      throw new Error('duplicate persistent subscription: ' + path)
    }

    this.subscriptions = this.subscriptions.set(path, {
      getter: initialDataGetter,
      cleanup,
    })
    for (const socket of this.sockets) {
      this.nydus.subscribeClient(socket, path, initialDataGetter(this, socket))
    }
  }

  _applySubscriptions(socket) {
    for (const [path, { getter }] of this.subscriptions.entries()) {
      this.nydus.subscribeClient(socket, path, getter(this, socket))
    }

    // Give the client a message so they know we're done subscribing them to things
    this.nydus.subscribeClient(socket, this.getUserPath(), { type: 'subscribed' })
  }

  _applyCleanups() {
    for (const [/* path */, { cleanup }] of this.subscriptions.entries()) {
      if (cleanup) cleanup(this)
    }
  }

  unsubscribe(path) {
    const updated = this.subscriptions.delete(path)
    if (updated === this.subscriptions) return

{'fraction_non_alphanumeric': 0.11605824013504959, 'fraction_numerical': 0.0006330449461911796, 'mean_word_length': 3.262589928057554, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5196021', 'n_tokens_mistral': 1439, 'n_tokens_neox': 1420, 'n_words': 383}