Reapplying only boiler plate code from D8955216

Summary:
Original commit changeset: 8703161f6d79
The original diff D8955216 landed but caused memory regression and later was reverted in D9262496. In order to find the problematic piece of the original code,
I am splitting it to the boiler plate code and the rest. This diff contains only the boiler plate code change. It is adding a parameter ProtocolVersion to the interface method
ConnectionFactory::connect and all its implementations. No behavior is added or changed.

Reviewed By: phoad

Differential Revision: D9501071

fbshipit-source-id: 172d30fda0c08896d198c75d3f063dbe1e95f613

Author: lehecka

rsocket/ConnectionFactory.h

@@ -17,6 +17,7 @@
 #include <folly/Function.h>
 #include <folly/futures/Future.h>
 #include "rsocket/DuplexConnection.h"
+#include "rsocket/framing/ProtocolVersion.h"
 
 namespace folly {
 class EventBase;
@@ -59,6 +60,7 @@ class ConnectionFactory {
    * Resource creation depends on the particular implementation.
    */
   virtual folly::Future<ConnectedDuplexConnection> connect(
+      ProtocolVersion,
       ResumeStatus resume) = 0;
 };
 } // namespace rsocket

rsocket/RSocket.cpp

@@ -28,7 +28,7 @@ folly::Future<std::unique_ptr<RSocketClient>> RSocket::createConnectedClient(
     folly::EventBase* stateMachineEvb) {
   CHECK(resumeManager)
       << "provide ResumeManager::makeEmpty() instead of nullptr";
-
+  auto protocolVersion = setupParameters.protocolVersion;
   auto createRSC =
       [connectionFactory,
        setupParameters = std::move(setupParameters),
@@ -55,7 +55,7 @@ folly::Future<std::unique_ptr<RSocketClient>> RSocket::createConnectedClient(
             stateMachineEvb);
       };
 
-  return connectionFactory->connect(ResumeStatus::NEW_SESSION)
+  return connectionFactory->connect(protocolVersion, ResumeStatus::NEW_SESSION)
       .then(
           [createRSC = std::move(createRSC)](
               ConnectionFactory::ConnectedDuplexConnection connection) mutable {

rsocket/RSocketClient.cpp

@@ -74,7 +74,7 @@ folly::Future<folly::Unit> RSocketClient::resume() {
       << "The client was likely created without ConnectionFactory. Can't "
       << "resume";
 
-  return connectionFactory_->connect(ResumeStatus::RESUMING)
+  return connectionFactory_->connect(protocolVersion_, ResumeStatus::RESUMING)
       .then(
           [this](
               ConnectionFactory::ConnectedDuplexConnection connection) mutable {

rsocket/benchmarks/StreamThroughputMemory.cpp

@@ -137,6 +137,7 @@ class Factory : public ConnectionFactory {
   }
 
   folly::Future<ConnectedDuplexConnection> connect(
+      ProtocolVersion,
       ResumeStatus /* unused */) override {
     return folly::via(worker_.getEventBase(), [this] {
       return ConnectedDuplexConnection{std::move(connection_),

rsocket/framing/FrameHeader.h

@@ -18,12 +18,10 @@
 
 #include "rsocket/framing/FrameFlags.h"
 #include "rsocket/framing/FrameType.h"
+#include "rsocket/internal/Common.h"
 
 namespace rsocket {
 
-/// A unique identifier of a stream.
-using StreamId = uint32_t;
-
 /// Header that begins every RSocket frame.
 class FrameHeader {
  public:

rsocket/framing/FrameSerializer.cpp

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 #include "rsocket/framing/FrameSerializer.h"
-
 #include "rsocket/framing/FrameSerializer_v1_0.h"
 
 namespace rsocket {
@@ -50,4 +49,16 @@ folly::IOBufQueue FrameSerializer::createBufferQueue(size_t bufferSize) const {
   return queue;
 }
 
+folly::Optional<StreamId> FrameSerializer::peekStreamId(
+    const ProtocolVersion& protocolVersion,
+    const folly::IOBuf& frame) {
+  if (protocolVersion == FrameSerializerV1_0::Version) {
+    return FrameSerializerV1_0().peekStreamId(frame);
+  }
+
+  auto* msg = "unknown protocol version";
+  DCHECK(false) << msg;
+  return folly::none;
+}
+
 } // namespace rsocket

rsocket/framing/FrameSerializer.h

@@ -35,6 +35,10 @@ class FrameSerializer {
   static std::unique_ptr<FrameSerializer> createAutodetectedSerializer(
       const folly::IOBuf& firstFrame);
 
+  static folly::Optional<StreamId> peekStreamId(
+      const ProtocolVersion& protocolVersion,
+      const folly::IOBuf& frame);
+
   virtual FrameType peekFrameType(const folly::IOBuf& in) const = 0;
   virtual folly::Optional<StreamId> peekStreamId(
       const folly::IOBuf& in) const = 0;

rsocket/framing/FramedDuplexConnection.cpp

@@ -13,10 +13,7 @@
 // limitations under the License.
 
 #include "rsocket/framing/FramedDuplexConnection.h"
-
 #include <folly/io/Cursor.h>
-
-#include "rsocket/framing/FrameSerializer.h"
 #include "rsocket/framing/FrameSerializer_v1_0.h"
 #include "rsocket/framing/FramedReader.h"
 

rsocket/internal/Common.h

@@ -37,6 +37,9 @@ typedef Range<const char*> StringPiece;
 
 namespace rsocket {
 
+/// A unique identifier of a stream.
+using StreamId = uint32_t;
+
 constexpr std::chrono::seconds kDefaultKeepaliveInterval{5};
 
 constexpr int64_t kMaxRequestN = std::numeric_limits<int32_t>::max();

rsocket/test/RSocketClientServerTest.cpp

@@ -111,7 +111,9 @@ TEST(RSocketClientServer, ServerGetsGarbage) {
   auto factory =
       std::make_shared<TcpConnectionFactory>(*worker.getEventBase(), address);
 
-  auto result = factory->connect(ResumeStatus::NEW_SESSION).get();
+  auto result =
+      factory->connect(ProtocolVersion::Latest, ResumeStatus::NEW_SESSION)
+          .get();
   auto connection = std::move(result.connection);
   auto evb = &result.eventBase;