[FIX] remove flusher timers if we do an inline flush of the outbound

nats-base-client/protocol.ts

@@ -395,6 +395,7 @@ export class ProtocolHandler implements Dispatcher<ParserEvent> {
   servers: Servers;
   server!: ServerImpl;
   features: Features;
+  flusher?: number;
 
   constructor(options: ConnectionOptions, publisher: Publisher) {
     this._closed = false;
@@ -843,10 +844,17 @@ export class ProtocolHandler implements Dispatcher<ParserEvent> {
     this.outbound.fill(buf, ...payloads);
 
     if (len === 0) {
-      setTimeout(() => {
+      this.flusher = setTimeout(() => {
         this.flushPending();
       });
     } else if (this.outbound.size() >= this.pendingLimit) {
+      // if we have a flusher, clear it - otherwise in a bench
+      // type scenario where the main loop is dominated by a publisher
+      // we create many timers.
+      if (this.flusher) {
+        clearTimeout(this.flusher);
+        this.flusher = undefined;
+      }
       this.flushPending();
     }
   }
@@ -890,7 +898,7 @@ export class ProtocolHandler implements Dispatcher<ParserEvent> {
     let proto: string;
     if (options.headers) {
       if (options.reply) {
-        proto = `HPUB ${subject} ${options.reply} ${hlen} ${len}${CR_LF}`;
+        proto = `HPUB ${subject} ${options.reply} ${hlen} ${len}\r\n`;
       } else {
         proto = `HPUB ${subject} ${hlen} ${len}\r\n`;
       }
@@ -945,9 +953,9 @@ export class ProtocolHandler implements Dispatcher<ParserEvent> {
       return;
     }
     if (max) {
-      this.sendCommand(`UNSUB ${s.sid} ${max}${CR_LF}`);
+      this.sendCommand(`UNSUB ${s.sid} ${max}\r\n`);
     } else {
-      this.sendCommand(`UNSUB ${s.sid}${CR_LF}`);
+      this.sendCommand(`UNSUB ${s.sid}\r\n`);
     }
     s.max = max;
   }