THRIFT-4564: Correctly reset the transport in NodeJS

compiler/cpp/src/thrift/generate/t_js_generator.cc

@@ -2274,14 +2274,10 @@ void t_js_generator::generate_service_client(t_service* tservice) {
     indent_up();
     if (gen_node_) {
       f_service_ << indent() << "delete this._reqs[this.seqid()];" << '\n';
-      f_service_ << indent() << "if (typeof " << outputVar << ".reset === 'function') {" << '\n';
-      f_service_ << indent() << "  " << outputVar << ".reset();" << '\n';
-      f_service_ << indent() << "}" << '\n';
-    } else {
-      f_service_ << indent() << "if (typeof " << outputVar << ".getTransport().reset === 'function') {" << '\n';
-      f_service_ << indent() << "  " << outputVar << ".getTransport().reset();" << '\n';
-      f_service_ << indent() << "}" << '\n';
     }
+    f_service_ << indent() << "if (typeof " << outputVar << ".getTransport().reset === 'function') {" << '\n';
+    f_service_ << indent() << "  " << outputVar << ".getTransport().reset();" << '\n';
+    f_service_ << indent() << "}" << '\n';
     f_service_ << indent() << "throw e;" << '\n';
     indent_down();
     f_service_ << indent() << "}" << '\n';

lib/nodejs/lib/thrift/buffered_transport.js

@@ -43,6 +43,7 @@ TBufferedTransport.prototype.reset = function () {
   this.writeCursor = 0;
   this.outBuffers = [];
   this.outCount = 0;
+  this._seqid = null;
 };
 
 // Default upper bound for a single accumulated message, consistent with the

lib/nodejs/lib/thrift/framed_transport.js

@@ -34,6 +34,12 @@ function TFramedTransport(buffer, callback) {
 
 Object.setPrototypeOf(TFramedTransport.prototype, THeaderTransport.prototype);
 
+TFramedTransport.prototype.reset = function () {
+  this.outBuffers = [];
+  this.outCount = 0;
+  this._seqid = null;
+};
+
 // Default upper bound for a single frame, consistent with the maxFrameSize
 // default used across the other Thrift libraries.
 TFramedTransport.DEFAULT_MAX_LENGTH = 16384000;

lib/nodejs/test/testAll.sh

@@ -149,6 +149,7 @@ node ${DIR}/deep-constructor.test.js || TESTOK=1
 node ${DIR}/recursion_depth.test.js || TESTOK=1
 node ${DIR}/connection_transport.test.js || TESTOK=1
 node ${DIR}/transport_receiver.test.js || TESTOK=1
+node ${DIR}/transport_reset.test.js || TESTOK=1
 node ${DIR}/server_error_events.test.js || TESTOK=1
 node ${DIR}/web_server_ws.test.js || TESTOK=1
 node ${DIR}/uuid.test.js || TESTOK=1

lib/nodejs/test/transport_reset.test.js

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+"use strict";
+
+const test = require("tape");
+const thrift = require("thrift");
+const ThriftTest = require("./gen-nodejs/ThriftTest");
+
+test("generated clients reset the transport after serialization errors", function t(assert) {
+  const flushed = [];
+  const transport = new thrift.TBufferedTransport(null, function (message) {
+    flushed.push(message);
+  });
+  const client = new ThriftTest.Client(transport, thrift.TBinaryProtocol);
+
+  assert.throws(
+    function () {
+      client.testString(42, function () {});
+    },
+    /writeString called without a string\/Buffer argument/,
+    "the serialization error is rethrown",
+  );
+  assert.equal(transport.outCount, 0, "the partial message is discarded");
+  assert.equal(
+    transport._seqid,
+    null,
+    "the partial message sequence ID is discarded",
+  );
+
+  client.testVoid(function () {});
+
+  assert.equal(flushed.length, 1, "the next request is flushed by itself");
+  let input;
+  thrift.TBufferedTransport.receiver(function (transport) {
+    input = transport;
+  })(flushed[0]);
+  const protocol = new thrift.TBinaryProtocol(input);
+  assert.equal(
+    protocol.readMessageBegin().fname,
+    "testVoid",
+    "the next request is not prefixed by the partial message",
+  );
+  assert.end();
+});
+
+test("generated clients reset framed transports after serialization errors", function t(assert) {
+  const flushed = [];
+  const transport = new thrift.TFramedTransport(null, function (message) {
+    flushed.push(message);
+  });
+  const client = new ThriftTest.Client(transport, thrift.TBinaryProtocol);
+
+  assert.throws(
+    function () {
+      client.testString(42, function () {});
+    },
+    /writeString called without a string\/Buffer argument/,
+    "the serialization error is rethrown",
+  );
+  assert.equal(transport.outCount, 0, "the partial message is discarded");
+  assert.equal(
+    transport._seqid,
+    null,
+    "the partial message sequence ID is discarded",
+  );
+
+  client.testVoid(function () {});
+
+  assert.equal(flushed.length, 1, "the next request is flushed by itself");
+  let input;
+  thrift.TFramedTransport.receiver(function (transport) {
+    input = transport;
+  })(flushed[0]);
+  const protocol = new thrift.TBinaryProtocol(input);
+  assert.equal(
+    protocol.readMessageBegin().fname,
+    "testVoid",
+    "the next request is not prefixed by the partial message",
+  );
+  assert.end();
+});