aboutsummaryrefslogtreecommitdiffstats
path: root/libtransport/src/protocols/byte_stream_reassembly.cc
diff options
context:
space:
mode:
Diffstat (limited to 'libtransport/src/protocols/byte_stream_reassembly.cc')
-rw-r--r--libtransport/src/protocols/byte_stream_reassembly.cc126
1 files changed, 126 insertions, 0 deletions
diff --git a/libtransport/src/protocols/byte_stream_reassembly.cc b/libtransport/src/protocols/byte_stream_reassembly.cc
new file mode 100644
index 000000000..c2996ebc1
--- /dev/null
+++ b/libtransport/src/protocols/byte_stream_reassembly.cc
@@ -0,0 +1,126 @@
+/*
+ * Copyright (c) 2017-2019 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:
+ *
+ * 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.
+ */
+
+#include <hicn/transport/interfaces/socket_consumer.h>
+#include <hicn/transport/utils/array.h>
+#include <hicn/transport/utils/membuf.h>
+
+#include <implementation/socket_consumer.h>
+#include <protocols/byte_stream_reassembly.h>
+#include <protocols/errors.h>
+#include <protocols/indexer.h>
+#include <protocols/protocol.h>
+
+namespace transport {
+
+namespace protocol {
+
+using namespace core;
+using ReadCallback = interface::ConsumerSocket::ReadCallback;
+
+ByteStreamReassembly::ByteStreamReassembly(
+ implementation::ConsumerSocket *icn_socket,
+ TransportProtocol *transport_protocol)
+ : Reassembly(icn_socket, transport_protocol),
+ index_(IndexManager::invalid_index),
+ download_complete_(false) {}
+
+void ByteStreamReassembly::reassemble(
+ std::unique_ptr<ContentObjectManifest> &&manifest) {
+ if (TRANSPORT_EXPECT_TRUE(manifest != nullptr)) {
+ received_packets_.emplace(
+ std::make_pair(manifest->getName().getSuffix(), nullptr));
+ assembleContent();
+ }
+}
+
+void ByteStreamReassembly::reassemble(ContentObject::Ptr &&content_object) {
+ if (TRANSPORT_EXPECT_TRUE(content_object != nullptr)) {
+ received_packets_.emplace(std::make_pair(
+ content_object->getName().getSuffix(), std::move(content_object)));
+ assembleContent();
+ }
+}
+
+void ByteStreamReassembly::assembleContent() {
+ if (TRANSPORT_EXPECT_FALSE(index_ == IndexManager::invalid_index)) {
+ index_ = index_manager_->getNextReassemblySegment();
+ if (index_ == IndexManager::invalid_index) {
+ return;
+ }
+ }
+
+ auto it = received_packets_.find((const unsigned int)index_);
+ while (it != received_packets_.end()) {
+ // Check if valid packet
+ if (it->second) {
+ copyContent(*it->second);
+ }
+
+ received_packets_.erase(it);
+ index_ = index_manager_->getNextReassemblySegment();
+ it = received_packets_.find((const unsigned int)index_);
+ }
+
+ if (!download_complete_ && index_ != IndexManager::invalid_index) {
+ transport_protocol_->onReassemblyFailed(index_);
+ }
+}
+
+void ByteStreamReassembly::copyContent(const ContentObject &content_object) {
+ auto a = content_object.getPayload();
+ auto payload_length = a->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;
+
+ std::memcpy(read_buffer_->writableTail(), a->data(), write_size);
+ read_buffer_->append(write_size);
+
+ if (!read_buffer_->tailroom()) {
+ notifyApplication();
+ std::memcpy(read_buffer_->writableTail(), a->data() + write_size,
+ additional_bytes);
+ read_buffer_->append(additional_bytes);
+ }
+
+ download_complete_ =
+ index_manager_->getFinalSuffix() == content_object.getName().getSuffix();
+
+ if (TRANSPORT_EXPECT_FALSE(download_complete_)) {
+ notifyApplication();
+ transport_protocol_->onContentReassembled(
+ make_error_code(protocol_error::success));
+ }
+}
+
+void ByteStreamReassembly::reInitialize() {
+ index_ = IndexManager::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());
+}
+
+} // namespace protocol
+
+} // namespace transport