9981127987

Committed 17 Jul 2024 08:36PM UTC coverage: 94.416% (-5.6%) from 100.0%

Build # 9981127987

Build Type

Pull #425

github

Committed by

web-flow

Commit Message

Merge 99cf534b7 into f1c1636ca

Pull Request Pull Request #425: v10

Run Details

465 of 546 branches covered (85.16%)

363 of 377 new or added lines in 10 files covered. (96.29%)

40 existing lines in 5 files now uncovered.

913 of 967 relevant lines covered (94.42%)

786.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.22

/src/index.js

const EventEmitter = require('events')
const plans = require('./plans')
const Attorney = require('./attorney')
const Contractor = require('./contractor')
const Manager = require('./manager')
const Timekeeper = require('./timekeeper')
const Boss = require('./boss')
const Db = require('./db')
const { delay } = require('./tools')

const events = {
  error: 'error',
  stopped: 'stopped'
}
class PgBoss extends EventEmitter {
  #stoppingOn
  #stopped
  #starting
  #started
  #config
  #db
  #boss
  #contractor
  #manager
  #timekeeper

  static getConstructionPlans (schema) {
    return Contractor.constructionPlans(schema)
  }

  static getMigrationPlans (schema, version) {
    return Contractor.migrationPlans(schema, version)
  }

  static getRollbackPlans (schema, version) {
    return Contractor.rollbackPlans(schema, version)
  }

  constructor (value) {
    super()

    this.#stoppingOn = null
    this.#stopped = true

    const config = Attorney.getConfig(value)
    this.#config = config

    const db = this.getDb()
    this.#db = db

    if (db.isOurs) {
      this.#promoteEvents(db)
    }

    const contractor = new Contractor(db, config)

    const manager = new Manager(db, config)
    const bossConfig = { ...config, manager }

    const boss = new Boss(db, bossConfig)

    const timekeeper = new Timekeeper(db, bossConfig)
    manager.timekeeper = timekeeper

    this.#promoteEvents(manager)
    this.#promoteEvents(boss)
    this.#promoteEvents(timekeeper)

    this.#promoteFunctions(boss)
    this.#promoteFunctions(contractor)
    this.#promoteFunctions(manager)
    this.#promoteFunctions(timekeeper)

    this.#boss = boss
    this.#contractor = contractor
    this.#manager = manager
    this.#timekeeper = timekeeper
  }

  getDb () {
    if (this.#db) {
      return this.#db
    }

    if (this.#config.db) {
      return this.#config.db
    }

    const db = new Db(this.#config)
    db.isOurs = true
    return db
  }

  #promoteEvents (emitter) {
    for (const event of Object.values(emitter?.events)) {
      emitter.on(event, arg => this.emit(event, arg))
    }
  }

  #promoteFunctions (obj) {
    for (const func of obj?.functions) {
      this[func.name] = (...args) => func.apply(obj, args)
    }
  }

  async start () {
    if (this.#starting || this.#started) {
      return
    }

    this.#starting = true

    if (this.#db.isOurs && !this.#db.opened) {
      await this.#db.open()
    }

    if (this.#config.migrate) {
      await this.#contractor.start()
    } else {
      await this.#contractor.check()
    }

    this.#manager.start()

    if (this.#config.supervise) {
      await this.#boss.supervise()
    }

    if (this.#config.monitorStateIntervalSeconds) {
      await this.#boss.monitor()
    }

    if (this.#config.schedule) {
      await this.#timekeeper.start()
    }

    this.#starting = false
    this.#started = true
    this.#stopped = false

    return this
  }

  async stop (options = {}) {
    if (this.#stoppingOn || this.#stopped) {
      return
    }

    let { destroy = false, graceful = true, timeout = 30000, wait = true } = options

    timeout = Math.max(timeout, 1000)

    this.#stoppingOn = Date.now()

    await this.#manager.stop()
    await this.#timekeeper.stop()
    await this.#boss.stop()

    await new Promise((resolve, reject) => {
      const shutdown = async () => {
        try {
          if (this.#config.__test__throw_shutdown) {
            throw new Error(this.#config.__test__throw_shutdown)
          }

          await this.#manager.failWip()

          if (this.#db.isOurs && this.#db.opened && destroy) {
            await this.#db.close()
          }

          this.#stopped = true
          this.#stoppingOn = null
          this.#started = false

          this.emit(events.stopped)
          resolve()
        } catch (err) {
          this.emit(events.error, err)
          reject(err)
        }
      }

      if (!graceful) {
        return shutdown()
      }

      if (!wait) {
        resolve()
      }

      setImmediate(async () => {
        try {
          if (this.#config.__test__throw_stop_monitor) {
            throw new Error(this.#config.__test__throw_stop_monitor)
          }

          const isWip = () => this.#manager.getWipData({ includeInternal: false }).length > 0

          while ((Date.now() - this.#stoppingOn) < timeout && isWip()) {
            await delay(500)
          }

          await shutdown()
        } catch (err) {
          reject(err)
          this.emit(events.error, err)
        }
      })
    })
  }
}

module.exports = PgBoss
module.exports.states = plans.states

1	const EventEmitter = require('events')	4✔
2	const plans = require('./plans')	4✔
3	const Attorney = require('./attorney')	4✔
4	const Contractor = require('./contractor')	4✔
5	const Manager = require('./manager')	4✔
6	const Timekeeper = require('./timekeeper')	4✔
7	const Boss = require('./boss')	4✔
8	const Db = require('./db')	4✔
9	const { delay } = require('./tools')	4✔
10
11	const events = {	4✔
12	error: 'error',
13	stopped: 'stopped'
14	}
15	class PgBoss extends EventEmitter {
16	#stoppingOn
17	#stopped
18	#starting
19	#started
20	#config
21	#db
22	#boss
23	#contractor
24	#manager
25	#timekeeper
26
27	static getConstructionPlans (schema) {
28	return Contractor.constructionPlans(schema)	1✔
29	}
30
31	static getMigrationPlans (schema, version) {
UNCOV 32	return Contractor.migrationPlans(schema, version)	×
33	}
34
35	static getRollbackPlans (schema, version) {
UNCOV 36	return Contractor.rollbackPlans(schema, version)	×
37	}
38
39	constructor (value) {
40	super()	183✔
41
42	this.#stoppingOn = null	183✔
43	this.#stopped = true	183✔
44
45	const config = Attorney.getConfig(value)	183✔
46	this.#config = config	182✔
47
48	const db = this.getDb()	182✔
49	this.#db = db	182✔
50
51	if (db.isOurs) {	182✔
52	this.#promoteEvents(db)	181✔
53	}
54
55	const contractor = new Contractor(db, config)	182✔
56
57	const manager = new Manager(db, config)	182✔
58	const bossConfig = { ...config, manager }	182✔
59
60	const boss = new Boss(db, bossConfig)	182✔
61
62	const timekeeper = new Timekeeper(db, bossConfig)	182✔
63	manager.timekeeper = timekeeper	182✔
64
65	this.#promoteEvents(manager)	182✔
66	this.#promoteEvents(boss)	182✔
67	this.#promoteEvents(timekeeper)	182✔
68
69	this.#promoteFunctions(boss)	182✔
70	this.#promoteFunctions(contractor)	182✔
71	this.#promoteFunctions(manager)	182✔
72	this.#promoteFunctions(timekeeper)	182✔
73
74	this.#boss = boss	182✔
75	this.#contractor = contractor	182✔
76	this.#manager = manager	182✔
77	this.#timekeeper = timekeeper	182✔
78	}
79
80	getDb () {
81	if (this.#db) {	185✔
82	return this.#db	3✔
83	}
84
85	if (this.#config.db) {	182✔
86	return this.#config.db	1✔
87	}
88
89	const db = new Db(this.#config)	181✔
90	db.isOurs = true	181✔
91	return db	181✔
92	}
93
94	#promoteEvents (emitter) {
95	for (const event of Object.values(emitter?.events)) {	727✔
96	emitter.on(event, arg => this.emit(event, arg))	1,455✔
97	}
98	}
99
100	#promoteFunctions (obj) {
101	for (const func of obj?.functions) {	728✔
102	this[func.name] = (...args) => func.apply(obj, args)	6,188✔
103	}
104	}
105
106	async start () {
107	if (this.#starting \|\| this.#started) {	184✔
108	return	2✔
109	}
110
111	this.#starting = true	182✔
112
113	if (this.#db.isOurs && !this.#db.opened) {	182✔
114	await this.#db.open()	180✔
115	}
116
117	if (this.#config.migrate) {	182!
118	await this.#contractor.start()	182✔
119	} else {
NEW 120	await this.#contractor.check()	×
121	}
122
123	this.#manager.start()	181✔
124
125	if (this.#config.supervise) {	181✔
126	await this.#boss.supervise()	10✔
127	}
128
129	if (this.#config.monitorStateIntervalSeconds) {	181✔
130	await this.#boss.monitor()	3✔
131	}
132
133	if (this.#config.schedule) {	181✔
134	await this.#timekeeper.start()	11✔
135	}
136
137	this.#starting = false	181✔
138	this.#started = true	181✔
139	this.#stopped = false	181✔
140
141	return this	181✔
142	}
143
144	async stop (options = {}) {	×
145	if (this.#stoppingOn \|\| this.#stopped) {	187✔
146	return	7✔
147	}
148
149	let { destroy = false, graceful = true, timeout = 30000, wait = true } = options	180✔
150
151	timeout = Math.max(timeout, 1000)	180✔
152
153	this.#stoppingOn = Date.now()	180✔
154
155	await this.#manager.stop()	180✔
156	await this.#timekeeper.stop()	180✔
157	await this.#boss.stop()	180✔
158
159	await new Promise((resolve, reject) => {	180✔
160	const shutdown = async () => {	180✔
161	try {	179✔
162	if (this.#config.__test__throw_shutdown) {	179✔
163	throw new Error(this.#config.__test__throw_shutdown)	1✔
164	}
165
166	await this.#manager.failWip()	178✔
167
168	if (this.#db.isOurs && this.#db.opened && destroy) {	178✔
169	await this.#db.close()	2✔
170	}
171
172	this.#stopped = true	178✔
173	this.#stoppingOn = null	178✔
174	this.#started = false	178✔
175
176	this.emit(events.stopped)	178✔
177	resolve()	178✔
178	} catch (err) {
179	this.emit(events.error, err)	1✔
180	reject(err)	1✔
181	}
182	}
183
184	if (!graceful) {	180✔
185	return shutdown()	27✔
186	}
187
188	if (!wait) {	153✔
189	resolve()	3✔
190	}
191
192	setImmediate(async () => {	153✔
193	try {	153✔
194	if (this.#config.__test__throw_stop_monitor) {	153✔
195	throw new Error(this.#config.__test__throw_stop_monitor)	1✔
196	}
197
198	const isWip = () => this.#manager.getWipData({ includeInternal: false }).length > 0	171✔
199
200	while ((Date.now() - this.#stoppingOn) < timeout && isWip()) {	152✔
201	await delay(500)	22✔
202	}
203
204	await shutdown()	152✔
205	} catch (err) {
206	reject(err)	1✔
207	this.emit(events.error, err)	1✔
208	}
209	})
210	})
211	}
212	}
213
214	module.exports = PgBoss	4✔
215	module.exports.states = plans.states	4✔

timgit / pg-boss / 9981127987

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous