dont send expired messages

Author: mshaposhnikov@w3-edge.com

client.js

@@ -42,10 +42,11 @@ class Client {
     this._clientId = p.clientId || 1;
 
     // build protocol bytes object
+    const timeoutMs = p.timeoutMs || 30000;
     this._protocolBytes = new ProtocolBytes();
     this._rateLimiter = new RateLimiter((data) => {
       return this._protocolBytes.sendFieldset(data);
-    }, 45, 1000, 5000);
+    }, 45, 1000, timeoutMs);
 
     this._protocolBytes.on('message_fieldset', (o) => {
       this._onMessageFieldset(o);
@@ -68,7 +69,7 @@ class Client {
 
     // attach messages handler
     this._incomeHandler = new IncomeFieldsetHandler({
-      timeoutMs: p.timeoutMs || 30000,
+      timeoutMs,
       eventEmitter: this._emitter
     });
 
@@ -97,6 +98,12 @@ class Client {
 
 
 
+  _sendFieldsetExpirable(fields) {
+    this._rateLimiter.runExpirable(fields);
+  }
+
+
+
   _sendStartApi() {
     const START_API = 71;
     const VERSION = 2;
@@ -126,7 +133,7 @@ class Client {
 
   async getCurrentTime() {
     /* Asks the current system time on the server side. */
-    this._sendFieldsetRateLimited([
+    this._sendFieldsetExpirable([
       OutcomeMessageType.REQ_CURRENT_TIME,
       1 /* VERSION */
     ]);
@@ -136,11 +143,11 @@ class Client {
   }
 
 
+
   /*########################################################################
   ################## Market Data
   ##########################################################################*/
 
-
   /*
   Starts to stream market data
   see reqMktData for parameters
@@ -179,12 +186,13 @@ class Client {
     p.genericTickList = p.genericTickList || '';
     p.snapshot = true;
     p.regulatorySnapshot = p.regulatorySnapshot || false;
-    this._sendFieldsetRateLimited(request_mktData(this._serverVersion, p));
+    this._sendFieldsetExpirable(request_mktData(this._serverVersion, p));
 
     return await this._incomeHandler.awaitRequestId(p.requestId);
   }
 
 
+
   async reqMarketDataType(marketDataType) {
     /* The API can receive frozen market data from Trader
     Workstation. Frozen market data is the last data recorded in our system.
@@ -544,7 +552,7 @@ class Client {
   async cancelOrder(orderId) {
     /* cancel an order. */
 
-    this._sendFieldsetRateLimited([
+    this._sendFieldsetExpirable([
       OutcomeMessageType.CANCEL_ORDER,
       1 /* VERSION */,
       orderId
@@ -563,7 +571,7 @@ class Client {
     orderId will be generated. This association will persist over multiple
     API and TWS sessions.  */
 
-    this._sendFieldsetRateLimited([
+    this._sendFieldsetExpirable([
       OutcomeMessageType.REQ_OPEN_ORDERS,
       1 /* VERSION */
     ]);
@@ -600,7 +608,7 @@ class Client {
     Note:  No association is made between the returned orders and the
     requesting client. */
 
-    this._sendFieldsetRateLimited([
+    this._sendFieldsetExpirable([
       OutcomeMessageType.REQ_ALL_OPEN_ORDERS,
       1 /* VERSION */
     ]);
@@ -752,7 +760,7 @@ class Client {
 
     const VERSION = 1;
 
-    this._sendFieldsetRateLimited([OutcomeMessageType.REQ_POSITIONS, VERSION]);
+    this._sendFieldsetExpirable([OutcomeMessageType.REQ_POSITIONS, VERSION]);
     return await this._incomeHandler.awaitMessageType(IncomeMessageType.POSITION_END);
   }
 
@@ -1063,7 +1071,7 @@ class Client {
       flds.push(contract.secId);
     }
 
-    this._sendFieldsetRateLimited(flds);
+    this._sendFieldsetExpirable(flds);
     return await this._incomeHandler.awaitRequestId(requestId);
   }
 
@@ -1477,7 +1485,7 @@ class Client {
       flds.push(chartOptionsStr);
     }
 
-    this._sendFieldsetRateLimited(flds);
+    this._sendFieldsetExpirable(flds);
     return await this._incomeHandler.awaitRequestId(requestId);
   }
 
@@ -1502,10 +1510,7 @@ class Client {
 
   async getHeadTimeStamp(p) {
     /*
-    contract:,
-    whatToShow: str,
-    useRth: int,
-    formatDate: int
+    whatToShow: str, useRth: int, formatDate: int
 
     Note that formatData parameter affects intraday bars only
     1-day bars always return with date in YYYYMMDD format
@@ -1542,7 +1547,7 @@ class Client {
       formatDate
     ];
 
-    this._sendFieldsetRateLimited(flds);
+    this._sendFieldsetExpirable(flds);
     return await this._incomeHandler.awaitRequestId(requestId);
   }
 
@@ -1558,7 +1563,7 @@ class Client {
 
 
 
-  async reqHistogramData(p) {
+  async getHistogramData(p) {
     /*
     contract: Contract,
     useRth: bool,
@@ -1573,7 +1578,7 @@ class Client {
       throw new Error("It does not support histogram requests.");
     }
 
-    this._sendFieldsetRateLimited([
+    this._sendFieldsetExpirable([
       OutcomeMessageType.REQ_HISTOGRAM_DATA,
       requestId,
       contract.conId,
@@ -1666,7 +1671,7 @@ class Client {
 
     flds.push(miscOptionsString);
 
-    this._sendFieldsetRateLimited(flds);
+    this._sendFieldsetExpirable(flds);
     return await this._incomeHandler.awaitRequestId(requestId);
   }
 
@@ -1757,7 +1762,7 @@ class Client {
       }
     }
 
-    this._sendFieldsetRateLimited(flds);
+    this._sendFieldsetExpirable(flds);
     return await this._incomeHandler.awaitRequestId(requestId);
   }
 
@@ -2190,7 +2195,7 @@ class Client {
       throw new Error("It does not support security definition option request.");
     }
 
-    this._sendFieldsetRateLimited([OutcomeMessageType.REQ_SEC_DEF_OPT_PARAMS,
+    this._sendFieldsetExpirable([OutcomeMessageType.REQ_SEC_DEF_OPT_PARAMS,
       requestId,
       underlyingSymbol,
       futFopExchange,

rate-limiter.js

@@ -1,22 +1,45 @@
+import util from 'util';
+const debuglog = util.debuglog('ib-tws-api');
+
+
 
 export default class RateLimiter {
-  constructor(workload, callsPerSlot, slotIntervalMs, maxQueueLength = Infinity) {
+  constructor(workload, callsPerSlot, slotIntervalMs, timeoutMs) {
     this._slotIntervalMs = slotIntervalMs;
     this._callsPerSlot = callsPerSlot;
-    this._maxQueueLength = maxQueueLength;
+    this._timeoutMs = timeoutMs;
     this._queue = [];
     this._workload = workload;
   }
 
 
 
   run(item) {
-    if (this.queueLength >= this.maxQueueLength) {
-      throw new Error('Too many requests queued');
-    }
+    return new Promise((resolve, reject) => {
+      this._queue.push({
+        item,
+        resolve,
+        reject
+      });
 
+      if (this._queue.length == 1) {
+        process.nextTick(() => {
+          this._processQueue();
+        });
+      }
+    });
+  }
+
+
+
+  runExpirable(item) {
     return new Promise((resolve, reject) => {
-      this._queue.push({ item, resolve, reject });
+      this._queue.push({
+        item,
+        resolve,
+        reject,
+        expireDate: Date.now() + this._timeoutMs
+      });
 
       if (this._queue.length == 1) {
         process.nextTick(() => {
@@ -43,9 +66,15 @@ export default class RateLimiter {
 
       if (this._slot_remaining > 0) {
         const i = this._queue.shift();
-        this._slot_remaining--;
 
-        Promise.resolve(this._workload(i.item)).then(i.resolve, i.reject);
+        if (i.expireDate && i.expireDate < Date.now()) {
+          // expired requests just ignored since timeout handled by
+          // IncomeFieldsetHandler.awaitRequestId
+          debuglog('RateLimiter: expired item ignored ' + i.item);
+        } else {
+          this._slot_remaining--;
+          Promise.resolve(this._workload(i.item)).then(i.resolve, i.reject);
+        }
       } else {
         if (this._timer) {
           clearTimeout(this._timer);