diff options
Diffstat (limited to 'libtransport/src/protocols/byte_stream_reassembly.cc')
-rw-r--r-- | libtransport/src/protocols/byte_stream_reassembly.cc | 84 |
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 |