stream-management: Debounce ack requests (#1062)

So that they don't send after every stanza goes out but only when idle.

Author: singpolyma

packages/stream-management/index.js

@@ -30,6 +30,7 @@ export default function streamManagement({
 }) {
   let timeoutTimeout = null;
   let requestAckTimeout = null;
+  let requestAckDebounce = null;
 
   const sm = new EventEmitter();
   Object.assign(sm, {
@@ -42,6 +43,7 @@ export default function streamManagement({
     max: null,
     timeout: 60_000,
     requestAckInterval: 30_000,
+    requestAckDebounce: 250,
   });
 
   async function sendAck() {
@@ -53,6 +55,7 @@ export default function streamManagement({
   entity.on("disconnect", () => {
     clearTimeout(timeoutTimeout);
     clearTimeout(requestAckTimeout);
+    clearTimeout(requestAckDebounce);
     sm.enabled = false;
   });
 
@@ -160,6 +163,7 @@ export default function streamManagement({
   function requestAck() {
     clearTimeout(timeoutTimeout);
     clearTimeout(requestAckTimeout);
+    clearTimeout(requestAckDebounce);
 
     if (!sm.enabled) return;
 
@@ -181,7 +185,9 @@ export default function streamManagement({
 
     sm.outbound_q.push({ stanza, stamp: datetime() });
     // Debounce requests so we send only one after a big run of stanza together
-    queueMicrotask(requestAck);
+    clearTimeout(requestAckTimeout);
+    clearTimeout(requestAckDebounce);
+    requestAckDebounce = setTimeout(requestAck, sm.requestAckDebounce);
 
     return next();
   });

packages/stream-management/stream-features.test.js

@@ -309,6 +309,39 @@ test("resume - failed with something in queue", async () => {
   expect(entity.streamManagement.outbound_q).toBeEmpty();
 });
 
+test("sends an <r/> after stanzas, debounced", async () => {
+  const { entity } = mockClient();
+
+  entity.streamManagement.enabled = true;
+
+  let r = 0;
+  const onSend = (stanza) => {
+    if (stanza.name === "r") r++;
+  };
+  entity.on("send", onSend);
+
+  jest.useFakeTimers();
+
+  let promise = entity.send(<message id="a" />);
+  jest.advanceTimersByTime(50);
+  await promise;
+  expect(r).toBe(0);
+
+  promise = entity.send(<message id="b" />);
+  jest.advanceTimersByTime(50);
+  await promise;
+  expect(r).toBe(0);
+
+  jest.advanceTimersByTime(1000);
+  jest.useRealTimers();
+  await tick();
+
+  expect(r).toBe(1);
+
+  entity.removeListener("send", onSend);
+  await entity.disconnect();
+});
+
 test("sends an <a/> element before closing", async () => {
   const { entity, streamManagement } = mockClient();
   streamManagement.enabled = true;