[HICN-534] Major rework on libtransport organization

Change-Id: I361b83a18b4fd59be136d5f0817fc28e17e89884
Signed-off-by: Mauro Sardara <msardara@cisco.com>

1 files changed, 1061 insertions, 0 deletions

diff --git a/libtransport/src/implementation/socket_producer.h b/libtransport/src/implementation/socket_producer.h
new file mode 100644
index 000000000..1f03fe53d
--- /dev/null
+++ b/libtransport/src/implementation/socket_producer.h
@@ -0,0 +1,1061 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <hicn/transport/security/signer.h>
+
+#include <implementation/socket.h>
+
+#include <utils/content_store.h>
+#include <utils/event_thread.h>
+#include <utils/suffix_strategy.h>
+
+#include <atomic>
+#include <cmath>
+#include <condition_variable>
+#include <mutex>
+#include <queue>
+#include <thread>
+
+#define REGISTRATION_NOT_ATTEMPTED 0
+#define REGISTRATION_SUCCESS 1
+#define REGISTRATION_FAILURE 2
+#define REGISTRATION_IN_PROGRESS 3
+
+namespace transport {
+namespace implementation {
+
+using namespace core;
+using namespace interface;
+
+class ProducerSocket : public Socket<BasePortal>,
+                       public BasePortal::ProducerCallback {
+ public:
+  explicit ProducerSocket(interface::ProducerSocket *producer_socket)
+      : producer_interface_(producer_socket),
+        portal_(std::make_shared<Portal>(io_service_)),
+        data_packet_size_(default_values::content_object_packet_size),
+        content_object_expiry_time_(default_values::content_object_expiry_time),
+        output_buffer_(default_values::producer_socket_output_buffer_size),
+        async_thread_(),
+        registration_status_(REGISTRATION_NOT_ATTEMPTED),
+        making_manifest_(false),
+        hash_algorithm_(HashAlgorithm::SHA_256),
+        suffix_strategy_(core::NextSegmentCalculationStrategy::INCREMENTAL),
+        on_interest_input_(VOID_HANDLER),
+        on_interest_dropped_input_buffer_(VOID_HANDLER),
+        on_interest_inserted_input_buffer_(VOID_HANDLER),
+        on_interest_satisfied_output_buffer_(VOID_HANDLER),
+        on_interest_process_(VOID_HANDLER),
+        on_new_segment_(VOID_HANDLER),
+        on_content_object_to_sign_(VOID_HANDLER),
+        on_content_object_in_output_buffer_(VOID_HANDLER),
+        on_content_object_output_(VOID_HANDLER),
+        on_content_object_evicted_from_output_buffer_(VOID_HANDLER),
+        on_content_produced_(VOID_HANDLER) {}
+
+  virtual ~ProducerSocket() {
+    stop();
+    if (listening_thread_.joinable()) {
+      listening_thread_.join();
+    }
+  }
+
+  interface::ProducerSocket *getInterface() {
+    return producer_interface_;
+  }
+
+  void setInterface(interface::ProducerSocket *producer_socket) {
+    producer_interface_ = producer_socket;
+  }
+
+  void connect() override {
+    portal_->connect(false);
+    listening_thread_ = std::thread(std::bind(&ProducerSocket::listen, this));
+  }
+
+  bool isRunning() override { return !io_service_.stopped(); };
+
+  virtual uint32_t produce(Name content_name, const uint8_t *buffer,
+                           size_t buffer_size, bool is_last = true,
+                           uint32_t start_offset = 0) {
+    return ProducerSocket::produce(
+        content_name, utils::MemBuf::copyBuffer(buffer, buffer_size), is_last,
+        start_offset);
+  }
+
+  virtual uint32_t produce(Name content_name,
+                           std::unique_ptr<utils::MemBuf> &&buffer,
+                           bool is_last = true, uint32_t start_offset = 0) {
+    if (TRANSPORT_EXPECT_FALSE(buffer->length() == 0)) {
+      return 0;
+    }
+
+    // Copy the atomic variables to ensure they keep the same value
+    // during the production
+    std::size_t data_packet_size = data_packet_size_;
+    uint32_t content_object_expiry_time = content_object_expiry_time_;
+    HashAlgorithm hash_algo = hash_algorithm_;
+    bool making_manifest = making_manifest_;
+    auto suffix_strategy = utils::SuffixStrategyFactory::getSuffixStrategy(
+        suffix_strategy_, start_offset);
+    std::shared_ptr<utils::Signer> signer;
+    getSocketOption(GeneralTransportOptions::SIGNER, signer);
+
+    auto buffer_size = buffer->length();
+    int bytes_segmented = 0;
+    std::size_t header_size;
+    std::size_t manifest_header_size = 0;
+    std::size_t signature_length = 0;
+    std::uint32_t final_block_number = start_offset;
+    uint64_t free_space_for_content = 0;
+
+    core::Packet::Format format;
+    std::shared_ptr<ContentObjectManifest> manifest;
+    bool is_last_manifest = false;
+
+    // TODO Manifest may still be used for indexing
+    if (making_manifest && !signer) {
+      TRANSPORT_LOGD("Making manifests without setting producer identity.");
+    }
+
+    core::Packet::Format hf_format = core::Packet::Format::HF_UNSPEC;
+    core::Packet::Format hf_format_ah = core::Packet::Format::HF_UNSPEC;
+    if (content_name.getType() == HNT_CONTIGUOUS_V4 ||
+        content_name.getType() == HNT_IOV_V4) {
+      hf_format = core::Packet::Format::HF_INET_TCP;
+      hf_format_ah = core::Packet::Format::HF_INET_TCP_AH;
+    } else if (content_name.getType() == HNT_CONTIGUOUS_V6 ||
+               content_name.getType() == HNT_IOV_V6) {
+      hf_format = core::Packet::Format::HF_INET6_TCP;
+      hf_format_ah = core::Packet::Format::HF_INET6_TCP_AH;
+    } else {
+      throw errors::RuntimeException("Unknown name format.");
+    }
+
+    format = hf_format;
+    if (making_manifest) {
+      manifest_header_size = core::Packet::getHeaderSizeFromFormat(
+          signer ? hf_format_ah : hf_format,
+          signer ? signer->getSignatureLength() : 0);
+    } else if (signer) {
+      format = hf_format_ah;
+      signature_length = signer->getSignatureLength();
+    }
+
+    header_size =
+        core::Packet::getHeaderSizeFromFormat(format, signature_length);
+    free_space_for_content = data_packet_size - header_size;
+    uint32_t number_of_segments = uint32_t(
+        std::ceil(double(buffer_size) / double(free_space_for_content)));
+    if (free_space_for_content * number_of_segments < buffer_size) {
+      number_of_segments++;
+    }
+
+    // TODO allocate space for all the headers
+    if (making_manifest) {
+      uint32_t segment_in_manifest = static_cast<uint32_t>(
+          std::floor(double(data_packet_size - manifest_header_size -
+                            ContentObjectManifest::getManifestHeaderSize()) /
+                     ContentObjectManifest::getManifestEntrySize()) -
+          1.0);
+      uint32_t number_of_manifests = static_cast<uint32_t>(
+          std::ceil(float(number_of_segments) / segment_in_manifest));
+      final_block_number += number_of_segments + number_of_manifests - 1;
+
+      manifest.reset(ContentObjectManifest::createManifest(
+          content_name.setSuffix(suffix_strategy->getNextManifestSuffix()),
+          core::ManifestVersion::VERSION_1, core::ManifestType::INLINE_MANIFEST,
+          hash_algo, is_last_manifest, content_name, suffix_strategy_,
+          signer ? signer->getSignatureLength() : 0));
+      manifest->setLifetime(content_object_expiry_time);
+
+      if (is_last) {
+        manifest->setFinalBlockNumber(final_block_number);
+      } else {
+        manifest->setFinalBlockNumber(utils::SuffixStrategy::INVALID_SUFFIX);
+      }
+    }
+
+    TRANSPORT_LOGD("--------- START PRODUCE ----------");
+    for (unsigned int packaged_segments = 0;
+         packaged_segments < number_of_segments; packaged_segments++) {
+      if (making_manifest) {
+        if (manifest->estimateManifestSize(2) >
+            data_packet_size - manifest_header_size) {
+          // Send the current manifest
+          manifest->encode();
+
+          // If identity set, sign manifest
+          if (signer) {
+            signer->sign(*manifest);
+          }
+
+          passContentObjectToCallbacks(manifest);
+          TRANSPORT_LOGD("Send manifest %u", manifest->getName().getSuffix());
+
+          // Send content objects stored in the queue
+          while (!content_queue_.empty()) {
+            passContentObjectToCallbacks(content_queue_.front());
+            TRANSPORT_LOGD("Send content %u",
+                           content_queue_.front()->getName().getSuffix());
+            content_queue_.pop();
+          }
+
+          // Create new manifest. The reference to the last manifest has been
+          // acquired in the passContentObjectToCallbacks function, so we can
+          // safely release this reference
+          manifest.reset(ContentObjectManifest::createManifest(
+              content_name.setSuffix(suffix_strategy->getNextManifestSuffix()),
+              core::ManifestVersion::VERSION_1,
+              core::ManifestType::INLINE_MANIFEST, hash_algo, is_last_manifest,
+              content_name, suffix_strategy_,
+              signer ? signer->getSignatureLength() : 0));
+
+          manifest->setLifetime(content_object_expiry_time);
+          manifest->setFinalBlockNumber(
+              is_last ? final_block_number
+                      : utils::SuffixStrategy::INVALID_SUFFIX);
+        }
+      }
+
+      auto content_suffix = suffix_strategy->getNextContentSuffix();
+      auto content_object = std::make_shared<ContentObject>(
+          content_name.setSuffix(content_suffix), format);
+      content_object->setLifetime(content_object_expiry_time);
+
+      auto b = buffer->cloneOne();
+      b->trimStart(free_space_for_content * packaged_segments);
+      b->trimEnd(b->length());
+
+      if (TRANSPORT_EXPECT_FALSE(packaged_segments == number_of_segments - 1)) {
+        b->append(buffer_size - bytes_segmented);
+        bytes_segmented += (int)(buffer_size - bytes_segmented);
+
+        if (is_last && making_manifest) {
+          is_last_manifest = true;
+        } else if (is_last) {
+          content_object->setRst();
+        }
+
+      } else {
+        b->append(free_space_for_content);
+        bytes_segmented += (int)(free_space_for_content);
+      }
+
+      content_object->appendPayload(std::move(b));
+
+      if (making_manifest) {
+        using namespace std::chrono_literals;
+        utils::CryptoHash hash = content_object->computeDigest(hash_algo);
+        manifest->addSuffixHash(content_suffix, hash);
+        content_queue_.push(content_object);
+      } else {
+        if (signer) {
+          signer->sign(*content_object);
+        }
+        passContentObjectToCallbacks(content_object);
+        TRANSPORT_LOGD("Send content %u",
+                       content_object->getName().getSuffix());
+      }
+    }
+
+    if (making_manifest) {
+      if (is_last_manifest) {
+        manifest->setFinalManifest(is_last_manifest);
+      }
+
+      manifest->encode();
+      if (signer) {
+        signer->sign(*manifest);
+      }
+
+      passContentObjectToCallbacks(manifest);
+      TRANSPORT_LOGD("Send manifest %u", manifest->getName().getSuffix());
+      while (!content_queue_.empty()) {
+        passContentObjectToCallbacks(content_queue_.front());
+        TRANSPORT_LOGD("Send content %u",
+                       content_queue_.front()->getName().getSuffix());
+        content_queue_.pop();
+      }
+    }
+
+    io_service_.dispatch([this, buffer_size]() {
+      if (on_content_produced_) {
+        on_content_produced_(*producer_interface_,
+                             std::make_error_code(std::errc(0)), buffer_size);
+      }
+    });
+
+    TRANSPORT_LOGD("--------- END PRODUCE ------------");
+    return suffix_strategy->getTotalCount();
+  }
+
+  virtual void produce(ContentObject &content_object) {
+    io_service_.dispatch([this, &content_object]() {
+      if (on_content_object_in_output_buffer_) {
+        on_content_object_in_output_buffer_(*producer_interface_,
+                                            content_object);
+      }
+    });
+
+    output_buffer_.insert(std::static_pointer_cast<ContentObject>(
+        content_object.shared_from_this()));
+
+    io_service_.dispatch([this, &content_object]() {
+      if (on_content_object_output_) {
+        on_content_object_output_(*producer_interface_, content_object);
+      }
+    });
+
+    portal_->sendContentObject(content_object);
+  }
+
+  virtual void produce(const uint8_t *buffer, size_t buffer_size) {
+    produce(utils::MemBuf::copyBuffer(buffer, buffer_size));
+  }
+
+  virtual void produce(std::unique_ptr<utils::MemBuf> &&buffer) {
+    // This API is meant to be used just with the RTC producer.
+    // Here it cannot be used since no name for the content is specified.
+    throw errors::NotImplementedException();
+  }
+
+  virtual void asyncProduce(const Name &suffix, const uint8_t *buf,
+                            size_t buffer_size, bool is_last = true,
+                            uint32_t *start_offset = nullptr) {
+    if (!async_thread_.stopped()) {
+      async_thread_.add([this, suffix, buffer = buf, size = buffer_size,
+                         is_last, start_offset]() {
+        if (start_offset != nullptr) {
+          *start_offset = ProducerSocket::produce(suffix, buffer, size, is_last,
+                                                  *start_offset);
+        } else {
+          ProducerSocket::produce(suffix, buffer, size, is_last, 0);
+        }
+      });
+    }
+  }
+
+  void asyncProduce(const Name &suffix);
+
+  virtual void asyncProduce(Name content_name,
+                            std::unique_ptr<utils::MemBuf> &&buffer,
+                            bool is_last, uint32_t offset,
+                            uint32_t **last_segment = nullptr) {
+    if (!async_thread_.stopped()) {
+      auto a = buffer.release();
+      async_thread_.add(
+          [this, content_name, a, is_last, offset, last_segment]() {
+            auto buf = std::unique_ptr<utils::MemBuf>(a);
+            if (last_segment != NULL) {
+              **last_segment =
+                  offset + ProducerSocket::produce(content_name, std::move(buf),
+                                                   is_last, offset);
+            } else {
+              ProducerSocket::produce(content_name, std::move(buf), is_last,
+                                      offset);
+            }
+          });
+    }
+  }
+
+  virtual void asyncProduce(ContentObject &content_object) {
+    if (!async_thread_.stopped()) {
+      auto co_ptr = std::static_pointer_cast<ContentObject>(
+          content_object.shared_from_this());
+      async_thread_.add([this, content_object = std::move(co_ptr)]() {
+        ProducerSocket::produce(*content_object);
+      });
+    }
+  }
+
+  virtual void registerPrefix(const Prefix &producer_namespace) {
+    served_namespaces_.push_back(producer_namespace);
+  }
+
+  void serveForever() {
+    if (listening_thread_.joinable()) {
+      listening_thread_.join();
+    }
+  }
+
+  void stop() { portal_->stopEventsLoop(); }
+
+  asio::io_service &getIoService() override { return portal_->getIoService(); };
+
+  virtual void onInterest(Interest &interest) {
+    if (on_interest_input_) {
+      on_interest_input_(*producer_interface_, interest);
+    }
+
+    const std::shared_ptr<ContentObject> content_object =
+        output_buffer_.find(interest);
+
+    if (content_object) {
+      if (on_interest_satisfied_output_buffer_) {
+        on_interest_satisfied_output_buffer_(*producer_interface_, interest);
+      }
+
+      if (on_content_object_output_) {
+        on_content_object_output_(*producer_interface_, *content_object);
+      }
+
+      portal_->sendContentObject(*content_object);
+    } else {
+      if (on_interest_process_) {
+        on_interest_process_(*producer_interface_, interest);
+      }
+    }
+  }
+
+  virtual void onInterest(Interest::Ptr &&interest) override {
+    onInterest(*interest);
+  };
+
+  virtual int setSocketOption(int socket_option_key,
+                              uint32_t socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::DATA_PACKET_SIZE:
+        if (socket_option_value < default_values::max_content_object_size &&
+            socket_option_value > 0) {
+          data_packet_size_ = socket_option_value;
+        }
+        break;
+
+      case GeneralTransportOptions::OUTPUT_BUFFER_SIZE:
+        output_buffer_.setLimit(socket_option_value);
+        break;
+
+      case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME:
+        content_object_expiry_time_ = socket_option_value;
+        break;
+
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              std::nullptr_t socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerContentObjectCallback socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::INTEREST_INPUT:
+              if (socket_option_value == VOID_HANDLER) {
+                on_interest_input_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::INTEREST_DROP:
+              if (socket_option_value == VOID_HANDLER) {
+                on_interest_dropped_input_buffer_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::INTEREST_PASS:
+              if (socket_option_value == VOID_HANDLER) {
+                on_interest_inserted_input_buffer_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::CACHE_HIT:
+              if (socket_option_value == VOID_HANDLER) {
+                on_interest_satisfied_output_buffer_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::CACHE_MISS:
+              if (socket_option_value == VOID_HANDLER) {
+                on_interest_process_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::NEW_CONTENT_OBJECT:
+              if (socket_option_value == VOID_HANDLER) {
+                on_new_segment_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN:
+              if (socket_option_value == VOID_HANDLER) {
+                on_content_object_to_sign_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_READY:
+              if (socket_option_value == VOID_HANDLER) {
+                on_content_object_in_output_buffer_ = VOID_HANDLER;
+                break;
+              }
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT:
+              if (socket_option_value == VOID_HANDLER) {
+                on_content_object_output_ = VOID_HANDLER;
+                break;
+              }
+
+            default:
+              return SOCKET_OPTION_NOT_SET;
+          }
+
+          return SOCKET_OPTION_SET;
+        });
+  }
+
+  virtual int setSocketOption(int socket_option_key, bool socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::MAKE_MANIFEST:
+        making_manifest_ = socket_option_value;
+        break;
+
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              Name *socket_option_value) {
+    return SOCKET_OPTION_NOT_SET;
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              std::list<Prefix> socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::NETWORK_NAME:
+        served_namespaces_ = socket_option_value;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int setSocketOption(
+      int socket_option_key,
+      interface::ProducerContentObjectCallback socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerContentObjectCallback socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::NEW_CONTENT_OBJECT:
+              on_new_segment_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN:
+              on_content_object_to_sign_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_READY:
+              on_content_object_in_output_buffer_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT:
+              on_content_object_output_ = socket_option_value;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_SET;
+          }
+
+          return SOCKET_OPTION_SET;
+        });
+  }
+
+  virtual int setSocketOption(
+      int socket_option_key,
+      interface::ProducerInterestCallback socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerInterestCallback socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::INTEREST_INPUT:
+              on_interest_input_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::INTEREST_DROP:
+              on_interest_dropped_input_buffer_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::INTEREST_PASS:
+              on_interest_inserted_input_buffer_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::CACHE_HIT:
+              on_interest_satisfied_output_buffer_ = socket_option_value;
+              break;
+
+            case ProducerCallbacksOptions::CACHE_MISS:
+              on_interest_process_ = socket_option_value;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_SET;
+          }
+
+          return SOCKET_OPTION_SET;
+        });
+  }
+
+  virtual int setSocketOption(
+      int socket_option_key,
+      interface::ProducerContentCallback socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerContentCallback socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::CONTENT_PRODUCED:
+              on_content_produced_ = socket_option_value;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_SET;
+          }
+
+          return SOCKET_OPTION_SET;
+        });
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              HashAlgorithm socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::HASH_ALGORITHM:
+        hash_algorithm_ = socket_option_value;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              utils::CryptoSuite socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::CRYPTO_SUITE:
+        crypto_suite_ = socket_option_value;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int setSocketOption(
+      int socket_option_key,
+      const std::shared_ptr<utils::Signer> &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::SIGNER: {
+        utils::SpinLock::Acquire locked(signer_lock_);
+        signer_.reset();
+        signer_ = socket_option_value;
+      } break;
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_SET;
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              uint32_t &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::OUTPUT_BUFFER_SIZE:
+        socket_option_value = (uint32_t)output_buffer_.getLimit();
+        break;
+
+      case GeneralTransportOptions::DATA_PACKET_SIZE:
+        socket_option_value = (uint32_t)data_packet_size_;
+        break;
+
+      case GeneralTransportOptions::CONTENT_OBJECT_EXPIRY_TIME:
+        socket_option_value = content_object_expiry_time_;
+        break;
+
+      default:
+        return SOCKET_OPTION_NOT_SET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              bool &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::MAKE_MANIFEST:
+        socket_option_value = making_manifest_;
+        break;
+
+      default:
+        return SOCKET_OPTION_NOT_GET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              std::list<Prefix> &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::NETWORK_NAME:
+        socket_option_value = served_namespaces_;
+        break;
+
+      default:
+        return SOCKET_OPTION_NOT_GET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(
+      int socket_option_key,
+      interface::ProducerContentObjectCallback **socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerContentObjectCallback **socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::NEW_CONTENT_OBJECT:
+              *socket_option_value = &on_new_segment_;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_SIGN:
+              *socket_option_value = &on_content_object_to_sign_;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_READY:
+              *socket_option_value = &on_content_object_in_output_buffer_;
+              break;
+
+            case ProducerCallbacksOptions::CONTENT_OBJECT_OUTPUT:
+              *socket_option_value = &on_content_object_output_;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_GET;
+          }
+
+          return SOCKET_OPTION_GET;
+        });
+  }
+
+  virtual int getSocketOption(
+      int socket_option_key,
+      interface::ProducerContentCallback **socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerContentCallback **socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::CONTENT_PRODUCED:
+              *socket_option_value = &on_content_produced_;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_GET;
+          }
+
+          return SOCKET_OPTION_GET;
+        });
+  }
+
+  virtual int getSocketOption(
+      int socket_option_key,
+      interface::ProducerInterestCallback **socket_option_value) {
+    // Reschedule the function on the io_service to avoid race condition in case
+    // setSocketOption is called while the io_service is running.
+    return rescheduleOnIOService(
+        socket_option_key, socket_option_value,
+        [this](int socket_option_key,
+               ProducerInterestCallback **socket_option_value) -> int {
+          switch (socket_option_key) {
+            case ProducerCallbacksOptions::INTEREST_INPUT:
+              *socket_option_value = &on_interest_input_;
+              break;
+
+            case ProducerCallbacksOptions::INTEREST_DROP:
+              *socket_option_value = &on_interest_dropped_input_buffer_;
+              break;
+
+            case ProducerCallbacksOptions::INTEREST_PASS:
+              *socket_option_value = &on_interest_inserted_input_buffer_;
+              break;
+
+            case CACHE_HIT:
+              *socket_option_value = &on_interest_satisfied_output_buffer_;
+              break;
+
+            case CACHE_MISS:
+              *socket_option_value = &on_interest_process_;
+              break;
+
+            default:
+              return SOCKET_OPTION_NOT_GET;
+          }
+
+          return SOCKET_OPTION_GET;
+        });
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              std::shared_ptr<Portal> &socket_option_value) {
+    switch (socket_option_key) {
+      case PORTAL:
+        socket_option_value = portal_;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_GET;
+        ;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              HashAlgorithm &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::HASH_ALGORITHM:
+        socket_option_value = hash_algorithm_;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_GET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(int socket_option_key,
+                              utils::CryptoSuite &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::HASH_ALGORITHM:
+        socket_option_value = crypto_suite_;
+        break;
+      default:
+        return SOCKET_OPTION_NOT_GET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int getSocketOption(
+      int socket_option_key,
+      std::shared_ptr<utils::Signer> &socket_option_value) {
+    switch (socket_option_key) {
+      case GeneralTransportOptions::SIGNER: {
+        utils::SpinLock::Acquire locked(signer_lock_);
+        socket_option_value = signer_;
+      } break;
+      default:
+        return SOCKET_OPTION_NOT_GET;
+    }
+
+    return SOCKET_OPTION_GET;
+  }
+
+  virtual int setSocketOption(int socket_option_key,
+                              const std::string &socket_option_value) {
+    return SOCKET_OPTION_NOT_SET;
+  }
+
+  // If the thread calling lambda_func is not the same of io_service, this
+  // function reschedule the function on it
+  template <typename Lambda, typename arg2>
+  int rescheduleOnIOService(int socket_option_key, arg2 socket_option_value,
+                            Lambda lambda_func) {
+    // To enforce type check
+    std::function<int(int, arg2)> func = lambda_func;
+    int result = SOCKET_OPTION_SET;
+    if (listening_thread_.joinable() &&
+        std::this_thread::get_id() != listening_thread_.get_id()) {
+      std::mutex mtx;
+      /* Condition variable for the wait */
+      std::condition_variable cv;
+      bool done = false;
+      io_service_.dispatch([&socket_option_key, &socket_option_value, &mtx, &cv,
+                            &result, &done, &func]() {
+        std::unique_lock<std::mutex> lck(mtx);
+        done = true;
+        result = func(socket_option_key, socket_option_value);
+        cv.notify_all();
+      });
+      std::unique_lock<std::mutex> lck(mtx);
+      if (!done) {
+        cv.wait(lck);
+      }
+    } else {
+      result = func(socket_option_key, socket_option_value);
+    }
+
+    return result;
+  }
+
+  // If the thread calling lambda_func is not the same of io_service, this
+  // function reschedule the function on it
+  template <typename Lambda, typename arg2>
+  int rescheduleOnIOServiceWithReference(int socket_option_key,
+                                         arg2 &socket_option_value,
+                                         Lambda lambda_func) {
+    // To enforce type check
+    std::function<int(int, arg2 &)> func = lambda_func;
+    int result = SOCKET_OPTION_SET;
+    if (listening_thread_.joinable() &&
+        std::this_thread::get_id() != this->listening_thread_.get_id()) {
+      std::mutex mtx;
+      /* Condition variable for the wait */
+      std::condition_variable cv;
+      std::unique_lock<std::mutex> lck(mtx);
+      bool done = false;
+      io_service_.dispatch([&socket_option_key, &socket_option_value, &mtx, &cv,
+                            &result, &done, &func]() {
+        std::unique_lock<std::mutex> lck(mtx);
+        done = true;
+        result = func(socket_option_key, socket_option_value);
+
+        if (!done) {
+          cv.wait(lck);
+        }
+      });
+    } else {
+      result = func(socket_option_key, socket_option_value);
+    }
+
+    return result;
+  }
+
+  // Threads
+ protected:
+  interface::ProducerSocket *producer_interface_;
+  std::thread listening_thread_;
+  asio::io_service io_service_;
+  std::shared_ptr<Portal> portal_;
+  std::atomic<size_t> data_packet_size_;
+  std::list<Prefix>
+      served_namespaces_;  // No need to be threadsafe, this is always modified
+                           // by the application thread
+  std::atomic<uint32_t> content_object_expiry_time_;
+
+  // buffers
+  // ContentStore is thread-safe
+  utils::ContentStore output_buffer_;
+
+  utils::EventThread async_thread_;
+  int registration_status_;
+
+  std::atomic<bool> making_manifest_;
+
+  // map for storing sequence numbers for several calls of the publish
+  // function
+  std::unordered_map<Name, std::unordered_map<int, uint32_t>> seq_number_map_;
+
+  std::atomic<HashAlgorithm> hash_algorithm_;
+  std::atomic<utils::CryptoSuite> crypto_suite_;
+  utils::SpinLock signer_lock_;
+  std::shared_ptr<utils::Signer> signer_;
+  core::NextSegmentCalculationStrategy suffix_strategy_;
+
+  // While manifests are being built, contents are stored in a queue
+  std::queue<std::shared_ptr<ContentObject>> content_queue_;
+
+  // callbacks
+  ProducerInterestCallback on_interest_input_;
+  ProducerInterestCallback on_interest_dropped_input_buffer_;
+  ProducerInterestCallback on_interest_inserted_input_buffer_;
+  ProducerInterestCallback on_interest_satisfied_output_buffer_;
+  ProducerInterestCallback on_interest_process_;
+
+  ProducerContentObjectCallback on_new_segment_;
+  ProducerContentObjectCallback on_content_object_to_sign_;
+  ProducerContentObjectCallback on_content_object_in_output_buffer_;
+  ProducerContentObjectCallback on_content_object_output_;
+  ProducerContentObjectCallback on_content_object_evicted_from_output_buffer_;
+
+  ProducerContentCallback on_content_produced_;
+
+ private:
+  void listen() {
+    bool first = true;
+
+    for (core::Prefix &producer_namespace : served_namespaces_) {
+      if (first) {
+        core::BindConfig bind_config(producer_namespace, 1000);
+        portal_->bind(bind_config);
+        portal_->setProducerCallback(this);
+        first = !first;
+      } else {
+        portal_->registerRoute(producer_namespace);
+      }
+    }
+
+    portal_->runEventsLoop();
+  }
+
+  void passContentObjectToCallbacks(
+      const std::shared_ptr<ContentObject> &content_object) {
+    if (content_object) {
+      io_service_.dispatch([this, content_object]() {
+        if (on_new_segment_) {
+          on_new_segment_(*producer_interface_, *content_object);
+        }
+
+        if (on_content_object_to_sign_) {
+          on_content_object_to_sign_(*producer_interface_, *content_object);
+        }
+
+        if (on_content_object_in_output_buffer_) {
+          on_content_object_in_output_buffer_(*producer_interface_,
+                                              *content_object);
+        }
+      });
+
+      output_buffer_.insert(content_object);
+
+      io_service_.dispatch([this, content_object]() {
+        if (on_content_object_output_) {
+          on_content_object_output_(*producer_interface_, *content_object);
+        }
+      });
+
+      portal_->sendContentObject(*content_object);
+    }
+  }
+};  // namespace implementation
+
+}  // namespace implementation
+
+}  // namespace transport