1 | 'use strict'
|
---|
2 |
|
---|
3 | const { InvalidArgumentError } = require('./core/errors')
|
---|
4 | const { kClients, kRunning, kClose, kDestroy, kDispatch, kInterceptors } = require('./core/symbols')
|
---|
5 | const DispatcherBase = require('./dispatcher-base')
|
---|
6 | const Pool = require('./pool')
|
---|
7 | const Client = require('./client')
|
---|
8 | const util = require('./core/util')
|
---|
9 | const createRedirectInterceptor = require('./interceptor/redirectInterceptor')
|
---|
10 | const { WeakRef, FinalizationRegistry } = require('./compat/dispatcher-weakref')()
|
---|
11 |
|
---|
12 | const kOnConnect = Symbol('onConnect')
|
---|
13 | const kOnDisconnect = Symbol('onDisconnect')
|
---|
14 | const kOnConnectionError = Symbol('onConnectionError')
|
---|
15 | const kMaxRedirections = Symbol('maxRedirections')
|
---|
16 | const kOnDrain = Symbol('onDrain')
|
---|
17 | const kFactory = Symbol('factory')
|
---|
18 | const kFinalizer = Symbol('finalizer')
|
---|
19 | const kOptions = Symbol('options')
|
---|
20 |
|
---|
21 | function defaultFactory (origin, opts) {
|
---|
22 | return opts && opts.connections === 1
|
---|
23 | ? new Client(origin, opts)
|
---|
24 | : new Pool(origin, opts)
|
---|
25 | }
|
---|
26 |
|
---|
27 | class Agent extends DispatcherBase {
|
---|
28 | constructor ({ factory = defaultFactory, maxRedirections = 0, connect, ...options } = {}) {
|
---|
29 | super()
|
---|
30 |
|
---|
31 | if (typeof factory !== 'function') {
|
---|
32 | throw new InvalidArgumentError('factory must be a function.')
|
---|
33 | }
|
---|
34 |
|
---|
35 | if (connect != null && typeof connect !== 'function' && typeof connect !== 'object') {
|
---|
36 | throw new InvalidArgumentError('connect must be a function or an object')
|
---|
37 | }
|
---|
38 |
|
---|
39 | if (!Number.isInteger(maxRedirections) || maxRedirections < 0) {
|
---|
40 | throw new InvalidArgumentError('maxRedirections must be a positive number')
|
---|
41 | }
|
---|
42 |
|
---|
43 | if (connect && typeof connect !== 'function') {
|
---|
44 | connect = { ...connect }
|
---|
45 | }
|
---|
46 |
|
---|
47 | this[kInterceptors] = options.interceptors && options.interceptors.Agent && Array.isArray(options.interceptors.Agent)
|
---|
48 | ? options.interceptors.Agent
|
---|
49 | : [createRedirectInterceptor({ maxRedirections })]
|
---|
50 |
|
---|
51 | this[kOptions] = { ...util.deepClone(options), connect }
|
---|
52 | this[kOptions].interceptors = options.interceptors
|
---|
53 | ? { ...options.interceptors }
|
---|
54 | : undefined
|
---|
55 | this[kMaxRedirections] = maxRedirections
|
---|
56 | this[kFactory] = factory
|
---|
57 | this[kClients] = new Map()
|
---|
58 | this[kFinalizer] = new FinalizationRegistry(/* istanbul ignore next: gc is undeterministic */ key => {
|
---|
59 | const ref = this[kClients].get(key)
|
---|
60 | if (ref !== undefined && ref.deref() === undefined) {
|
---|
61 | this[kClients].delete(key)
|
---|
62 | }
|
---|
63 | })
|
---|
64 |
|
---|
65 | const agent = this
|
---|
66 |
|
---|
67 | this[kOnDrain] = (origin, targets) => {
|
---|
68 | agent.emit('drain', origin, [agent, ...targets])
|
---|
69 | }
|
---|
70 |
|
---|
71 | this[kOnConnect] = (origin, targets) => {
|
---|
72 | agent.emit('connect', origin, [agent, ...targets])
|
---|
73 | }
|
---|
74 |
|
---|
75 | this[kOnDisconnect] = (origin, targets, err) => {
|
---|
76 | agent.emit('disconnect', origin, [agent, ...targets], err)
|
---|
77 | }
|
---|
78 |
|
---|
79 | this[kOnConnectionError] = (origin, targets, err) => {
|
---|
80 | agent.emit('connectionError', origin, [agent, ...targets], err)
|
---|
81 | }
|
---|
82 | }
|
---|
83 |
|
---|
84 | get [kRunning] () {
|
---|
85 | let ret = 0
|
---|
86 | for (const ref of this[kClients].values()) {
|
---|
87 | const client = ref.deref()
|
---|
88 | /* istanbul ignore next: gc is undeterministic */
|
---|
89 | if (client) {
|
---|
90 | ret += client[kRunning]
|
---|
91 | }
|
---|
92 | }
|
---|
93 | return ret
|
---|
94 | }
|
---|
95 |
|
---|
96 | [kDispatch] (opts, handler) {
|
---|
97 | let key
|
---|
98 | if (opts.origin && (typeof opts.origin === 'string' || opts.origin instanceof URL)) {
|
---|
99 | key = String(opts.origin)
|
---|
100 | } else {
|
---|
101 | throw new InvalidArgumentError('opts.origin must be a non-empty string or URL.')
|
---|
102 | }
|
---|
103 |
|
---|
104 | const ref = this[kClients].get(key)
|
---|
105 |
|
---|
106 | let dispatcher = ref ? ref.deref() : null
|
---|
107 | if (!dispatcher) {
|
---|
108 | dispatcher = this[kFactory](opts.origin, this[kOptions])
|
---|
109 | .on('drain', this[kOnDrain])
|
---|
110 | .on('connect', this[kOnConnect])
|
---|
111 | .on('disconnect', this[kOnDisconnect])
|
---|
112 | .on('connectionError', this[kOnConnectionError])
|
---|
113 |
|
---|
114 | this[kClients].set(key, new WeakRef(dispatcher))
|
---|
115 | this[kFinalizer].register(dispatcher, key)
|
---|
116 | }
|
---|
117 |
|
---|
118 | return dispatcher.dispatch(opts, handler)
|
---|
119 | }
|
---|
120 |
|
---|
121 | async [kClose] () {
|
---|
122 | const closePromises = []
|
---|
123 | for (const ref of this[kClients].values()) {
|
---|
124 | const client = ref.deref()
|
---|
125 | /* istanbul ignore else: gc is undeterministic */
|
---|
126 | if (client) {
|
---|
127 | closePromises.push(client.close())
|
---|
128 | }
|
---|
129 | }
|
---|
130 |
|
---|
131 | await Promise.all(closePromises)
|
---|
132 | }
|
---|
133 |
|
---|
134 | async [kDestroy] (err) {
|
---|
135 | const destroyPromises = []
|
---|
136 | for (const ref of this[kClients].values()) {
|
---|
137 | const client = ref.deref()
|
---|
138 | /* istanbul ignore else: gc is undeterministic */
|
---|
139 | if (client) {
|
---|
140 | destroyPromises.push(client.destroy(err))
|
---|
141 | }
|
---|
142 | }
|
---|
143 |
|
---|
144 | await Promise.all(destroyPromises)
|
---|
145 | }
|
---|
146 | }
|
---|
147 |
|
---|
148 | module.exports = Agent
|
---|