hyper-tps / distribute / hypertps-win64 / lib / daemon-client.js

import Pipe from 'bare-pipe'
import FramedStream from 'framed-stream'
import b4a from 'b4a'
import fs from 'bare-fs'
import EventEmitter from 'bare-events'

const DEFAULT_SOCKET = '/tmp/keet-daemon.sock'
const RPC_TIMEOUT = 10000
const BASE_DELAY = 5000
const MAX_DELAY = 30000

export default class DaemonClient extends EventEmitter {
  constructor (opts = {}) {
    super()
    this.socketPath = opts.socketPath || DEFAULT_SOCKET
    this.connected = false

    this._pipe = null
    this._framed = null
    this._nextId = 1
    this._pending = new Map()
    this._reconnectTimer = null
    this._currentDelay = 0
    this._destroyed = false
  }

  connect () {
    if (this._destroyed) return
    this._clearReconnect()

    // Check socket exists before connecting (avoids noisy errors)
    try {
      fs.statSync(this.socketPath)
    } catch {
      this._scheduleReconnect()
      return
    }

    try {
      this._pipe = new Pipe(this.socketPath)
    } catch (err) {
      this._scheduleReconnect()
      return
    }

    this._framed = new FramedStream(this._pipe)

    this._pipe.on('error', () => {
      this._onDisconnect()
    })

    this._pipe.on('close', () => {
      this._onDisconnect()
    })

    this._framed.on('data', (buf) => {
      this._onMessage(buf)
    })

    // Pipe connected
    this.connected = true
    this._currentDelay = 0
    this.emit('connected')
  }

  async rpc (method, params = {}) {
    if (!this.connected || !this._framed) {
      throw new Error('Not connected')
    }

    const id = this._nextId++
    const msg = JSON.stringify({ id, method, params })

    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this._pending.delete(id)
        reject(new Error(`RPC timeout: ${method}`))
      }, RPC_TIMEOUT)

      this._pending.set(id, { resolve, reject, timer })
      this._framed.write(b4a.from(msg, 'utf-8'))
    })
  }

  destroy () {
    this._destroyed = true
    this._clearReconnect()

    // Fail all pending RPCs
    for (const [id, entry] of this._pending) {
      clearTimeout(entry.timer)
      entry.reject(new Error('Client destroyed'))
    }
    this._pending.clear()

    if (this._framed) {
      this._framed.destroy()
      this._framed = null
    }
    if (this._pipe) {
      this._pipe.destroy()
      this._pipe = null
    }
    this.connected = false
  }

  _onMessage (buf) {
    let msg
    try {
      msg = JSON.parse(b4a.toString(buf, 'utf-8'))
    } catch {
      return
    }

    // Push event (no id field)
    if (msg.event) {
      this.emit('event', msg)
      return
    }

    // RPC response
    if (msg.id != null && this._pending.has(msg.id)) {
      const entry = this._pending.get(msg.id)
      this._pending.delete(msg.id)
      clearTimeout(entry.timer)

      if (msg.ok === false || msg.error) {
        entry.reject(new Error(msg.error || 'RPC failed'))
      } else {
        entry.resolve(msg.data)
      }
    }
  }

  _onDisconnect () {
    if (!this.connected) return
    this.connected = false

    // Fail all pending RPCs
    for (const [id, entry] of this._pending) {
      clearTimeout(entry.timer)
      entry.reject(new Error('Disconnected'))
    }
    this._pending.clear()

    this._framed = null
    this._pipe = null

    this.emit('disconnected')

    if (!this._destroyed) {
      this._scheduleReconnect()
    }
  }

  _scheduleReconnect () {
    if (this._reconnectTimer || this._destroyed) return
    if (this._currentDelay === 0) {
      this._currentDelay = BASE_DELAY
    } else {
      this._currentDelay = Math.min(this._currentDelay * 1.5, MAX_DELAY)
    }
    this._reconnectTimer = setTimeout(() => {
      this._reconnectTimer = null
      this.connect()
    }, this._currentDelay)
  }

  _clearReconnect () {
    if (this._reconnectTimer) {
      clearTimeout(this._reconnectTimer)
      this._reconnectTimer = null
    }
  }
}