<commit_before>import asap from 'asap'
import assign from 'lodash/assign'
import cookies from 'cookies-js'
import isEmpty from 'lodash/isEmpty'
import isEqual from 'lodash/isEqual'
import filter from 'lodash/filter'
import forEach from 'lodash/forEach'
import map from 'lodash/map'
import once from 'lodash/once'
import React from 'react'
import fpSortBy from 'lodash/fp/sortBy'
import request from 'superagent'
import size from 'lodash/size'
import sortBy from 'lodash/sortBy'
import throttle from 'lodash/throttle'
import Xo from 'xo-lib'
import { createBackoff } from 'jsonrpc-websocket-client'
import { resolve } from 'url'

import _ from '../intl'
import invoke from '../invoke'
import logError from '../log-error'
import { confirm } from '../modal'
import { error, info, success } from '../notification'
import { noop, rethrow, tap } from '../utils'
import {
  connected,
  disconnected,
  signedIn,
  signedOut,
  updateObjects,
  updatePermissions
} from '../store/actions'

// ===================================================================

export const XEN_DEFAULT_CPU_WEIGHT = 256
export const XEN_DEFAULT_CPU_CAP = 0

// ===================================================================

export const isSrWritable = sr => sr && sr.content_type !== 'iso' && sr.size > 0
export const isSrShared = sr => sr && sr.$PBDs.length > 1
export const isVmRunning = vm => vm && vm.power_state === 'Running'

// ===================================================================

export const signOut = () => {
  cookies.expire('token')
  window.location.reload(true)
}

export const connect = () => {
  xo.open(createBackoff()).catch(error => {
    logError(error, 'failed to connect to xo-server')
  })
}

const xo = invoke(() => {
  const token = cookies.get('token')
  if (!token) {
    signOut()
    throw new Error('no valid token')
  }

  const xo = new Xo({
    credentials: { token }
  })

  xo.on('scheduledAttempt', ({ delay }) => {
    console.warn('next attempt in %s ms', delay)
  })

  xo.on('closed', connect)

  return xo
})
connect()

const _signIn = new Promise(resolve => xo.once('authenticated', resolve))

const _call = (method, params) => {
  let promise = _signIn.then(() => xo.call(method, params))

  if (process.env.NODE_ENV !== 'production') {
    promise = promise::rethrow(error => {
      console.error('XO error', {
        method,
        params,
        code: error.code,
        message: error.message,
        data: error.data
      })
    })
  }

  return promise
}

// ===================================================================

export const connectStore = store => {
  let updates = {}
  const sendUpdates = throttle(() => {
    store.dispatch(updateObjects(updates))
    updates = {}
  }, 5e2)

  xo.on('open', () => store.dispatch(connected()))
  xo.on('closed', () => {
    store.dispatch(signedOut())
    store.dispatch(disconnected())
  })
  xo.on('authenticated', () => {
    store.dispatch(signedIn(xo.user))

    _call('xo.getAllObjects').then(objects => store.dispatch(updateObjects(objects)))
  })
  xo.on('notification', notification => {
    if (notification.method !== 'all') {
      return
    }

    assign(updates, notification.params.items)
    sendUpdates()
  })
  subscribePermissions(permissions => store.dispatch(updatePermissions(permissions)))
}

// -------------------------------------------------------------------

export const resolveUrl = invoke(
  xo._url, // FIXME: accessing private prop
  baseUrl => to => resolve(baseUrl, to)
)

// -------------------------------------------------------------------

const createSubscription = cb => {
  const delay = 5e3

  const subscribers = Object.create(null)
  let cache
  let n = 0
  let nextId = 0
  let timeout

  let running = false

  const loop = () => {
    if (running) {
      return
    }

    running = true
    _signIn.then(() => cb()).then(result => {
      running = false
      timeout = setTimeout(loop, delay)

      if (!isEqual(result, cache)) {
        cache = result

        /* FIXME: Edge case:
         * 1) MyComponent has a subscription with subscribers[1]
         * 2) subscribers[0] causes the MyComponent unmounting (and thus its unsubscription)
         * When subscribers[1] will be executed, it will no longer exist,
         * which will throw an error (Uncaught (in promise) TypeError: subscriber is not a function)
         */
        forEach(subscribers, subscriber => {
          subscriber(result)
        })
      }
    }, ::console.error)
  }

  const subscribe = cb => {
    const id = nextId++
    subscribers[id] = cb

    if (n++) {
      cache !== undefined && asap(() => cb(cache))
    } else {
      loop()
    }<commit_msg>fix(xo/subcriptions): mark as non-running on error<commit_after>import asap from 'asap'
import assign from 'lodash/assign'
import cookies from 'cookies-js'
import isEmpty from 'lodash/isEmpty'
import isEqual from 'lodash/isEqual'
import filter from 'lodash/filter'
import forEach from 'lodash/forEach'
import map from 'lodash/map'
import once from 'lodash/once'
import React from 'react'
import fpSortBy from 'lodash/fp/sortBy'
import request from 'superagent'
import size from 'lodash/size'
import sortBy from 'lodash/sortBy'
import throttle from 'lodash/throttle'
import Xo from 'xo-lib'
import { createBackoff } from 'jsonrpc-websocket-client'
import { resolve } from 'url'

import _ from '../intl'
import invoke from '../invoke'
import logError from '../log-error'
import { confirm } from '../modal'
import { error, info, success } from '../notification'
import { noop, rethrow, tap } from '../utils'
import {
  connected,
  disconnected,
  signedIn,
  signedOut,
  updateObjects,
  updatePermissions
} from '../store/actions'

// ===================================================================

export const XEN_DEFAULT_CPU_WEIGHT = 256
export const XEN_DEFAULT_CPU_CAP = 0

// ===================================================================

export const isSrWritable = sr => sr && sr.content_type !== 'iso' && sr.size > 0
export const isSrShared = sr => sr && sr.$PBDs.length > 1
export const isVmRunning = vm => vm && vm.power_state === 'Running'

// ===================================================================

export const signOut = () => {
  cookies.expire('token')
  window.location.reload(true)
}

export const connect = () => {
  xo.open(createBackoff()).catch(error => {
    logError(error, 'failed to connect to xo-server')
  })
}

const xo = invoke(() => {
  const token = cookies.get('token')
  if (!token) {
    signOut()
    throw new Error('no valid token')
  }

  const xo = new Xo({
    credentials: { token }
  })

  xo.on('scheduledAttempt', ({ delay }) => {
    console.warn('next attempt in %s ms', delay)
  })

  xo.on('closed', connect)

  return xo
})
connect()

const _signIn = new Promise(resolve => xo.once('authenticated', resolve))

const _call = (method, params) => {
  let promise = _signIn.then(() => xo.call(method, params))

  if (process.env.NODE_ENV !== 'production') {
    promise = promise::rethrow(error => {
      console.error('XO error', {
        method,
        params,
        code: error.code,
        message: error.message,
        data: error.data
      })
    })
  }

  return promise
}

// ===================================================================

export const connectStore = store => {
  let updates = {}
  const sendUpdates = throttle(() => {
    store.dispatch(updateObjects(updates))
    updates = {}
  }, 5e2)

  xo.on('open', () => store.dispatch(connected()))
  xo.on('closed', () => {
    store.dispatch(signedOut())
    store.dispatch(disconnected())
  })
  xo.on('authenticated', () => {
    store.dispatch(signedIn(xo.user))

    _call('xo.getAllObjects').then(objects => store.dispatch(updateObjects(objects)))
  })
  xo.on('notification', notification => {
    if (notification.method !== 'all') {
      return
    }

    assign(updates, notification.params.items)
    sendUpdates()
  })
  subscribePermissions(permissions => store.dispatch(updatePermissions(permissions)))
}

// -------------------------------------------------------------------

export const resolveUrl = invoke(
  xo._url, // FIXME: accessing private prop
  baseUrl => to => resolve(baseUrl, to)
)

// -------------------------------------------------------------------

const createSubscription = cb => {
  const delay = 5e3

  const subscribers = Object.create(null)
  let cache
  let n = 0
  let nextId = 0
  let timeout

  let running = false

  const loop = () => {
    if (running) {
      return
    }

    running = true
    _signIn.then(() => cb()).then(result => {
      running = false
      timeout = setTimeout(loop, delay)

      if (!isEqual(result, cache)) {
        cache = result

        /* FIXME: Edge case:
         * 1) MyComponent has a subscription with subscribers[1]
         * 2) subscribers[0] causes the MyComponent unmounting (and thus its unsubscription)
         * When subscribers[1] will be executed, it will no longer exist,
         * which will throw an error (Uncaught (in promise) TypeError: subscriber is not a function)
         */
        forEach(subscribers, subscriber => {
          subscriber(result)
        })
      }
    }, error => {
      running = false
      console.error(error)
    })
  }

  const subscribe = cb => {
    const id = nextId++
    subscribers[id] = cb

    if (n++) {
      cache !== undefined && asap(() => cb(cache))
    } else {
      loop()
    }
{'fraction_non_alphanumeric': 0.21414269153756982, 'fraction_numerical': 0.003583096216671936, 'mean_word_length': 2.929606625258799, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 66, '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': '774958', 'n_tokens_mistral': 2901, 'n_tokens_neox': 2749, 'n_words': 746}