1 files changed, 44 insertions, 40 deletions

diff --git a/libtransport/src/protocols/byte_stream_reassembly.cc b/libtransport/src/protocols/byte_stream_reassembly.cc
index 6662bec3f..b9eaf3bec 100644
--- a/libtransport/src/protocols/byte_stream_reassembly.cc
+++ b/libtransport/src/protocols/byte_stream_reassembly.cc
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017-2019 Cisco and/or its affiliates.
+ * Copyright (c) 2021 Cisco and/or its affiliates.
  * Licensed 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:
@@ -20,7 +20,7 @@
 #include <protocols/byte_stream_reassembly.h>
 #include <protocols/errors.h>
 #include <protocols/indexer.h>
-#include <protocols/protocol.h>
+#include <protocols/transport_protocol.h>
 
 namespace transport {
 
@@ -33,31 +33,26 @@ ByteStreamReassembly::ByteStreamReassembly(
     implementation::ConsumerSocket *icn_socket,
     TransportProtocol *transport_protocol)
     : Reassembly(icn_socket, transport_protocol),
-      index_(IndexManager::invalid_index),
+      index_(Indexer::invalid_index),
       download_complete_(false) {}
 
-void ByteStreamReassembly::reassemble(
-    std::unique_ptr<ContentObjectManifest> &&manifest) {
-  if (TRANSPORT_EXPECT_TRUE(manifest != nullptr) && read_buffer_->capacity()) {
+void ByteStreamReassembly::reassemble(ContentObject &content_object) {
+  if (TRANSPORT_EXPECT_TRUE(read_buffer_->capacity())) {
     received_packets_.emplace(
-        std::make_pair(manifest->getName().getSuffix(), nullptr));
+        std::make_pair(content_object.getName().getSuffix(),
+                       content_object.shared_from_this()));
     assembleContent();
   }
 }
 
-void ByteStreamReassembly::reassemble(ContentObject::Ptr &&content_object) {
-  if (TRANSPORT_EXPECT_TRUE(content_object != nullptr) &&
-      read_buffer_->capacity()) {
-    received_packets_.emplace(std::make_pair(
-        content_object->getName().getSuffix(), std::move(content_object)));
-    assembleContent();
-  }
+void ByteStreamReassembly::reassemble(utils::MemBuf &buffer, uint32_t suffix) {
+  throw errors::NotImplementedException();
 }
 
 void ByteStreamReassembly::assembleContent() {
-  if (TRANSPORT_EXPECT_FALSE(index_ == IndexManager::invalid_index)) {
-    index_ = index_manager_->getNextReassemblySegment();
-    if (index_ == IndexManager::invalid_index) {
+  if (TRANSPORT_EXPECT_FALSE(index_ == Indexer::invalid_index)) {
+    index_ = indexer_verifier_->getNextReassemblySegment();
+    if (index_ == Indexer::invalid_index) {
       return;
     }
   }
@@ -72,37 +67,44 @@ void ByteStreamReassembly::assembleContent() {
     }
 
     received_packets_.erase(it);
-    index_ = index_manager_->getNextReassemblySegment();
+    index_ = indexer_verifier_->getNextReassemblySegment();
     it = received_packets_.find((const unsigned int)index_);
   }
 
-  if (!download_complete_ && index_ != IndexManager::invalid_index) {
+  if (!download_complete_ && index_ != Indexer::invalid_index) {
     transport_protocol_->onReassemblyFailed(index_);
   }
 }
 
-bool ByteStreamReassembly::copyContent(const ContentObject &content_object) {
+bool ByteStreamReassembly::copyContent(ContentObject &content_object) {
   bool ret = false;
 
-  auto payload = content_object.getPayloadReference();
-  auto payload_length = payload.second;
-  auto write_size = std::min(payload_length, read_buffer_->tailroom());
-  auto additional_bytes = payload_length > read_buffer_->tailroom()
-                              ? payload_length - read_buffer_->tailroom()
-                              : 0;
+  content_object.trimStart(content_object.headerSize());
 
-  std::memcpy(read_buffer_->writableTail(), payload.first, write_size);
-  read_buffer_->append(write_size);
+  utils::MemBuf *current = &content_object;
 
-  if (!read_buffer_->tailroom()) {
-    notifyApplication();
-    std::memcpy(read_buffer_->writableTail(), payload.first + write_size,
-                additional_bytes);
-    read_buffer_->append(additional_bytes);
-  }
+  do {
+    auto payload_length = current->length();
+    auto write_size = std::min(payload_length, read_buffer_->tailroom());
+    auto additional_bytes = payload_length > read_buffer_->tailroom()
+                                ? payload_length - read_buffer_->tailroom()
+                                : 0;
 
-  download_complete_ =
-      index_manager_->getFinalSuffix() == content_object.getName().getSuffix();
+    std::memcpy(read_buffer_->writableTail(), current->data(), write_size);
+    read_buffer_->append(write_size);
+
+    if (!read_buffer_->tailroom()) {
+      notifyApplication();
+      std::memcpy(read_buffer_->writableTail(), current->data() + write_size,
+                  additional_bytes);
+      read_buffer_->append(additional_bytes);
+    }
+
+    current = current->next();
+  } while (current != &content_object);
+
+  download_complete_ = indexer_verifier_->getFinalSuffix() ==
+                       content_object.getName().getSuffix();
 
   if (TRANSPORT_EXPECT_FALSE(download_complete_)) {
     ret = download_complete_;
@@ -115,17 +117,19 @@ bool ByteStreamReassembly::copyContent(const ContentObject &content_object) {
 }
 
 void ByteStreamReassembly::reInitialize() {
-  index_ = IndexManager::invalid_index;
+  index_ = Indexer::invalid_index;
   download_complete_ = false;
 
   received_packets_.clear();
 
   // reset read buffer
   ReadCallback *read_callback;
-  reassembly_consumer_socket_->getSocketOption(
-      interface::ConsumerCallbacksOptions::READ_CALLBACK, &read_callback);
 
-  read_buffer_ = utils::MemBuf::create(read_callback->maxBufferSize());
+  if (reassembly_consumer_socket_) {
+    reassembly_consumer_socket_->getSocketOption(
+        interface::ConsumerCallbacksOptions::READ_CALLBACK, &read_callback);
+    read_buffer_ = utils::MemBuf::create(read_callback->maxBufferSize());
+  }
 }
 
 }  // namespace protocol