Fixed Duplicate deletion bug

davewang · davewang · commit 81dcb2edc90f · 2018-06-13T15:22:53.000+08:00
diff --git a/node-binance-api.js b/node-binance-api.js
@@ -25,6 +25,7 @@ let api = function Binance() {
     const HttpsProxyAgent = require('https-proxy-agent');
     const SocksProxyAgent = require('socks-proxy-agent');
     const stringHash = require('string-hash');
+    const async = require("async");
     const base = 'https://api.binance.com/api/';
     const wapi = 'https://api.binance.com/wapi/';
     const stream = 'wss://stream.binance.com:9443/ws/';
@@ -1666,26 +1667,39 @@ let api = function Binance() {
                     }
                 };
 
-                let getSymbolDepthSnapshot = function(symbol) {
-                    publicRequest(base+'v1/depth', { symbol:symbol, limit:limit }, function(error, json) {
-                        // Initialize depth cache from snapshot
-                        depthCache[symbol] = depthData(json);
-                        // Prepare depth cache context
-                        let context = depthCacheContext[symbol];
-                        context.snapshotUpdateId = json.lastUpdateId;
-                        context.messageQueue = context.messageQueue.filter(depth => depth.u > context.snapshotUpdateId);
-                        // Process any pending depth messages
-                        for ( let depth of context.messageQueue ) {
-
-                            /* Although sync errors shouldn't ever happen here, we catch and swallow them anyway
-                               just in case. The stream handler function above will deal with broken caches. */
-                            try {depthHandler(depth);} catch (err) {
-                                // do nothing
-                            }
+                let getSymbolDepthSnapshot = async function(symbol){
+                   return new Promise((resolve, reject) => {
+                       publicRequest(base+'v1/depth', { symbol:symbol, limit:limit }, function(error, json) {
+                           if (error) {
+                               return reject (error);
+                           }
+                           // Store symbol next use
+                           json.symbol_ = symbol;
+                           resolve(json);
+                       });
+                   })
+               };
+
+               let updateSymbolDepthCache = async function(json){
+                    // Get previous store symbol
+                    let symbol = json.symbol_;
+                    // Initialize depth cache from snapshot
+                    depthCache[symbol] = depthData(json);
+                    // Prepare depth cache context
+                    let context = depthCacheContext[symbol];
+                    context.snapshotUpdateId = json.lastUpdateId;
+                    context.messageQueue = context.messageQueue.filter(depth => depth.u > context.snapshotUpdateId);
+                    // Process any pending depth messages
+                    for ( let depth of context.messageQueue ) {
+
+                        /* Although sync errors shouldn't ever happen here, we catch and swallow them anyway
+                           just in case. The stream handler function above will deal with broken caches. */
+                        try {depthHandler(depth);} catch (err) {
+                            // do nothing
                         }
-                        delete context.messageQueue;
-                        if ( callback ) callback(symbol, depthCache[symbol]);
-                    });
+                    }
+                    delete context.messageQueue;
+                    if ( callback ) callback(symbol, depthCache[symbol]);
                 };
 
                 /* If an array of symbols are sent we use a combined stream connection rather.
@@ -1700,14 +1714,20 @@ let api = function Binance() {
                         return symbol.toLowerCase()+'@depth';
                     });
                     subscription = subscribeCombined(streams, handleDepthStreamData, reconnect, function() {
-                        symbols.forEach(getSymbolDepthSnapshot);
+                      async.mapLimit(symbols, symbols.length, getSymbolDepthSnapshot,(err, results) => {
+                         if (err) throw err
+                         results.forEach(updateSymbolDepthCache);
+                      });
                     });
                     symbols.forEach(s => assignEndpointIdToContext(s, subscription.endpoint));
                 } else {
                     let symbol = symbols;
                     symbolDepthInit(symbol);
                     subscription = subscribe(symbol.toLowerCase()+'@depth', handleDepthStreamData, reconnect, function() {
-                        getSymbolDepthSnapshot(symbol);
+                      async.mapLimit([symbol], 1, getSymbolDepthSnapshot,(err, results) => {
+                          if (err) throw err
+                          results.forEach(updateSymbolDepthCache);
+                      });
                     });
                     assignEndpointIdToContext(symbol, subscription.endpoint);
                 }
diff --git a/package.json b/package.json
@@ -4,6 +4,7 @@
   "description": "Binance API for node https://github.com/jaggedsoft/node-binance-api",
   "main": "node-binance-api.js",
   "dependencies": {
+    "async": "^2.6.1",
     "dns-sync": "^0.1.3",
     "fs": "0.0.1-security",
     "https-proxy-agent": "^2.2.1",
