* Websocket depth cache * @param {array/string} symbols - an array or string of symbols to query * @param {function} callback - callback function * @param {int} limit - the number of entries * @return {string} the websocket endpoint
(symbols: string[] | string, callback: Callback, limit = 500)
| 6452 | * @return {string} the websocket endpoint |
| 6453 | */ |
| 6454 | depthCacheStream(symbols: string[] | string, callback: Callback, limit = 500) { |
| 6455 | const reconnect = () => { |
| 6456 | if (this.Options.reconnect) this.depthCacheStream(symbols, callback, limit); |
| 6457 | }; |
| 6458 | |
| 6459 | const symbolDepthInit = symbol => { |
| 6460 | if (typeof this.depthCacheContext[symbol] === 'undefined') this.depthCacheContext[symbol] = {}; |
| 6461 | const context = this.depthCacheContext[symbol]; |
| 6462 | context.snapshotUpdateId = null; |
| 6463 | context.lastEventUpdateId = null; |
| 6464 | context.messageQueue = []; |
| 6465 | this.depthCache[symbol] = { bids: {}, asks: {}}; |
| 6466 | }; |
| 6467 | |
| 6468 | const assignEndpointIdToContext = (symbol, endpointId) => { |
| 6469 | if (this.depthCacheContext[symbol]) { |
| 6470 | const context = this.depthCacheContext[symbol]; |
| 6471 | context.endpointId = endpointId; |
| 6472 | } |
| 6473 | }; |
| 6474 | |
| 6475 | const handleDepthStreamData = depth => { |
| 6476 | const symbol = depth.s; |
| 6477 | const context = this.depthCacheContext[symbol]; |
| 6478 | if (context.messageQueue && !context.snapshotUpdateId) { |
| 6479 | context.messageQueue.push(depth); |
| 6480 | } else { |
| 6481 | try { |
| 6482 | this.depthHandler(depth); |
| 6483 | } catch (err) { |
| 6484 | return this.terminate(context.endpointId, true); |
| 6485 | } |
| 6486 | if (callback) callback(symbol, this.depthCache[symbol], context); |
| 6487 | } |
| 6488 | }; |
| 6489 | |
| 6490 | const getSymbolDepthSnapshot = async (symbol: string) => { |
| 6491 | const json = await this.publicSpotRequest('v3/depth', { symbol: symbol, limit: limit }); |
| 6492 | json.symbol = symbol; |
| 6493 | // cb(null, json); |
| 6494 | return json; |
| 6495 | }; |
| 6496 | |
| 6497 | const updateSymbolDepthCache = json => { |
| 6498 | // Get previous store symbol |
| 6499 | const symbol = json.symbol; |
| 6500 | // Initialize depth cache from snapshot |
| 6501 | this.depthCache[symbol] = this.depthData(json); |
| 6502 | // Prepare depth cache context |
| 6503 | const context = this.depthCacheContext[symbol]; |
| 6504 | context.snapshotUpdateId = json.lastUpdateId; |
| 6505 | context.messageQueue = context.messageQueue.filter(depth => depth.u > context.snapshotUpdateId); |
| 6506 | // Process any pending depth messages |
| 6507 | for (const depth of context.messageQueue) { |
| 6508 | /* Although sync errors shouldn't ever happen here, we catch and swallow them anyway |
| 6509 | just in case. The stream handler function above will deal with broken caches. */ |
| 6510 | try { |
| 6511 | this.depthHandler(depth); |
no test coverage detected