src: Small source reorganization
Change-Id: I442f3ad0392ef4220e6435fc48e6d69acbd02076
diff --git a/src/handles/base-handle.cpp b/src/handles/base-handle.cpp
new file mode 100644
index 0000000..76fa36c
--- /dev/null
+++ b/src/handles/base-handle.cpp
@@ -0,0 +1,32 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "base-handle.hpp"
+
+namespace repo {
+
+uint64_t
+BaseHandle::generateProcessId()
+{
+ static boost::random::mt19937_64 gen;
+ static boost::random::uniform_int_distribution<uint64_t> dist(0, 0xFFFFFFFFFFFFFFFFLL);
+ return dist(gen);
+}
+
+}
diff --git a/src/handles/base-handle.hpp b/src/handles/base-handle.hpp
new file mode 100644
index 0000000..3223940
--- /dev/null
+++ b/src/handles/base-handle.hpp
@@ -0,0 +1,120 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef REPO_HANDLES_BASE_HANDLE_HPP
+#define REPO_HANDLES_BASE_HANDLE_HPP
+
+#include "common.hpp"
+
+#include "storage/storage-handle.hpp"
+#include "repo-command-response.hpp"
+#include "repo-command-parameter.hpp"
+
+namespace repo {
+
+class BaseHandle : noncopyable
+{
+
+public:
+ class Error : std::runtime_error
+ {
+ public:
+ explicit
+ Error(const std::string& what)
+ : std::runtime_error(what)
+ {
+ }
+ };
+
+public:
+ BaseHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain, Scheduler& scheduler)
+ : m_face(face)
+ , m_storageHandle(storageHandle)
+ , m_keyChain(keyChain)
+ , m_scheduler(scheduler)
+ {
+ }
+
+ virtual void
+ listen(const Name& prefix) = 0;
+
+protected:
+
+ inline Face&
+ getFace()
+ {
+ return m_face;
+ }
+
+ inline StorageHandle&
+ getStorageHandle()
+ {
+ return m_storageHandle;
+ }
+
+ inline Scheduler&
+ getScheduler()
+ {
+ return m_scheduler;
+ }
+
+ uint64_t
+ generateProcessId();
+
+ void
+ reply(const Interest& commandInterest, const RepoCommandResponse& response);
+
+
+ /**
+ * @brief extract RepoCommandParameter from a command Interest.
+ * @param interest command Interest
+ * @param prefix Name prefix up to command-verb
+ * @param[out] parameter parsed parameter
+ * @throw RepoCommandParameter::Error parse error
+ */
+ void
+ extractParameter(const Interest& interest, const Name& prefix, RepoCommandParameter& parameter);
+
+private:
+
+ Face& m_face;
+ StorageHandle& m_storageHandle;
+ KeyChain& m_keyChain;
+ Scheduler& m_scheduler;
+};
+
+inline void
+BaseHandle::reply(const Interest& commandInterest, const RepoCommandResponse& response)
+{
+ Data rdata(commandInterest.getName());
+ rdata.setContent(response.wireEncode());
+ m_keyChain.sign(rdata);
+ m_face.put(rdata);
+}
+
+inline void
+BaseHandle::extractParameter(const Interest& interest, const Name& prefix,
+ RepoCommandParameter& parameter)
+{
+ parameter.wireDecode(interest.getName().get(prefix.size()).blockFromValue());
+}
+
+} // namespace repo
+
+#endif // REPO_HANDLES_BASE_HANDLE_HPP
diff --git a/src/handles/delete-handle.cpp b/src/handles/delete-handle.cpp
new file mode 100644
index 0000000..49e4d98
--- /dev/null
+++ b/src/handles/delete-handle.cpp
@@ -0,0 +1,197 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "delete-handle.hpp"
+
+namespace repo {
+
+DeleteHandle::DeleteHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain,
+ Scheduler& scheduler, CommandInterestValidator& validator)
+ : BaseHandle(face, storageHandle, keyChain, scheduler)
+ , m_validator(validator)
+{
+}
+
+void
+DeleteHandle::onInterest(const Name& prefix, const Interest& interest)
+{
+ //std::cout << "call DeleteHandle" << std::endl;
+ m_validator.validate(interest, bind(&DeleteHandle::onValidated, this, _1, prefix),
+ bind(&DeleteHandle::onValidationFailed, this, _1, prefix));
+}
+
+
+void
+DeleteHandle::onRegisterFailed(const Name& prefix, const std::string& reason)
+{
+ throw Error("Delete prefix registration failed");
+}
+
+
+void
+DeleteHandle::onCheckInterest(const Name& prefix, const Interest& interest)
+{
+ BOOST_ASSERT(false); // Deletion progress check, not implemented
+}
+
+
+void
+DeleteHandle::onCheckRegisterFailed(const Name& prefix, const std::string& reason)
+{
+ throw Error("Delete check prefix registration failed");
+}
+
+
+void
+DeleteHandle::onValidated(const shared_ptr<const Interest>& interest, const Name& prefix)
+{
+ RepoCommandParameter parameter;
+
+ try {
+ extractParameter(*interest, prefix, parameter);
+ }
+ catch (RepoCommandParameter::Error) {
+ negativeReply(*interest, 403);
+ return;
+ }
+
+ if (parameter.hasSelectors()) {
+
+ if (parameter.hasStartBlockId() || parameter.hasEndBlockId()) {
+ negativeReply(*interest, 402);
+ return;
+ }
+
+ //choose data with selector and delete it
+ processSelectorDeleteCommand(*interest, parameter);
+ return;
+ }
+
+ if (!parameter.hasStartBlockId() && !parameter.hasEndBlockId()) {
+ processSingleDeleteCommand(*interest, parameter);
+ return;
+ }
+
+ processSegmentDeleteCommand(*interest, parameter);
+}
+
+void
+DeleteHandle::onValidationFailed(const shared_ptr<const Interest>& interest, const Name& prefix)
+{
+ std::cout << "invalidated" << std::endl;
+ negativeReply(*interest, 401);
+}
+
+void
+DeleteHandle::listen(const Name& prefix)
+{
+ getFace().setInterestFilter(Name(prefix).append("delete"),
+ bind(&DeleteHandle::onInterest, this, _1, _2),
+ bind(&DeleteHandle::onRegisterFailed, this, _1, _2));
+}
+
+void
+DeleteHandle::positiveReply(const Interest& interest, const RepoCommandParameter& parameter,
+ uint64_t statusCode, uint64_t nDeletedDatas)
+{
+ RepoCommandResponse response;
+ if (parameter.hasProcessId()) {
+ response.setProcessId(parameter.getProcessId());
+ response.setStatusCode(statusCode);
+ response.setDeleteNum(nDeletedDatas);
+ }
+ else {
+ response.setStatusCode(403);
+ }
+ reply(interest, response);
+}
+
+void
+DeleteHandle::negativeReply(const Interest& interest, uint64_t statusCode)
+{
+ RepoCommandResponse response;
+ response.setStatusCode(statusCode);
+ reply(interest, response);
+}
+
+void
+DeleteHandle::processSingleDeleteCommand(const Interest& interest,
+ RepoCommandParameter& parameter)
+{
+ uint64_t nDeletedDatas = 0;
+ if (getStorageHandle().deleteData(parameter.getName())) {
+ nDeletedDatas++;
+ }
+ positiveReply(interest, parameter, 200, nDeletedDatas);
+}
+
+void
+DeleteHandle::processSelectorDeleteCommand(const Interest& interest,
+ RepoCommandParameter& parameter)
+{
+ uint64_t nDeletedDatas = 0;
+ Name name = parameter.getName();
+ Selectors selectors = parameter.getSelectors();
+ vector<Name> names;
+ getStorageHandle().readNameAny(name, selectors, names);
+
+ for (vector<Name>::iterator it = names.begin(); it != names.end(); ++it) {
+ if (getStorageHandle().deleteData(*it)) {
+ nDeletedDatas++;
+ }
+ }
+
+ //All data has been deleted, return 200
+ positiveReply(interest, parameter, 200, nDeletedDatas);
+}
+
+void
+DeleteHandle::processSegmentDeleteCommand(const Interest& interest,
+ RepoCommandParameter& parameter)
+{
+ if (!parameter.hasStartBlockId())
+ parameter.setStartBlockId(0);
+
+ if (parameter.hasEndBlockId()) {
+ SegmentNo startBlockId = parameter.getStartBlockId();
+ SegmentNo endBlockId = parameter.getEndBlockId();
+
+ if (startBlockId > endBlockId) {
+ negativeReply(interest, 403);
+ return;
+ }
+
+ Name prefix = parameter.getName();
+ uint64_t nDeletedDatas = 0;
+ for (SegmentNo i = startBlockId; i <= endBlockId; i++) {
+ Name name = prefix;
+ name.appendSegment(i);
+ if (getStorageHandle().deleteData(name)) {
+ nDeletedDatas++;
+ }
+ }
+ //All the data deleted, return 200
+ positiveReply(interest, parameter, 200, nDeletedDatas);
+ }
+ else {
+ BOOST_ASSERT(false); // segmented deletion without EndBlockId, not implemented
+ }
+}
+
+} //namespace repo
diff --git a/src/handles/delete-handle.hpp b/src/handles/delete-handle.hpp
new file mode 100644
index 0000000..680e8dc
--- /dev/null
+++ b/src/handles/delete-handle.hpp
@@ -0,0 +1,95 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef REPO_HANDLES_DELETE_HANDLE_HPP
+#define REPO_HANDLES_DELETE_HANDLE_HPP
+
+#include "base-handle.hpp"
+
+namespace repo {
+
+using std::vector;
+
+class DeleteHandle : public BaseHandle
+{
+
+public:
+ class Error : public BaseHandle::Error
+ {
+ public:
+ explicit
+ Error(const std::string& what)
+ : BaseHandle::Error(what)
+ {
+ }
+ };
+
+public:
+ DeleteHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain,
+ Scheduler& scheduler, CommandInterestValidator& validator);
+
+ virtual void
+ listen(const Name& prefix);
+
+private:
+ void
+ onInterest(const Name& prefix, const Interest& interest);
+
+ void
+ onRegisterFailed(const Name& prefix, const std::string& reason);
+
+ void
+ onValidated(const shared_ptr<const Interest>& interest, const Name& prefix);
+
+ void
+ onValidationFailed(const shared_ptr<const Interest>& interest, const Name& prefix);
+
+ /**
+ * @todo delete check has not been realized due to the while loop of segmented data deletion.
+ */
+ void
+ onCheckInterest(const Name& prefix, const Interest& interest);
+
+ void
+ onCheckRegisterFailed(const Name& prefix, const std::string& reason);
+
+ void
+ positiveReply(const Interest& interest, const RepoCommandParameter& parameter,
+ uint64_t statusCode, uint64_t nDeletedDatas);
+
+ void
+ negativeReply(const Interest& interest, uint64_t statusCode);
+
+ void
+ processSingleDeleteCommand(const Interest& interest, RepoCommandParameter& parameter);
+
+ void
+ processSelectorDeleteCommand(const Interest& interest, RepoCommandParameter& parameter);
+
+ void
+ processSegmentDeleteCommand(const Interest& interest, RepoCommandParameter& parameter);
+
+private:
+ CommandInterestValidator& m_validator;
+
+};
+
+} // namespace repo
+
+#endif // REPO_HANDLES_DELETE_HANDLE_HPP
diff --git a/src/handles/read-handle.cpp b/src/handles/read-handle.cpp
new file mode 100644
index 0000000..d0e80b3
--- /dev/null
+++ b/src/handles/read-handle.cpp
@@ -0,0 +1,48 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "read-handle.hpp"
+
+namespace repo {
+
+void
+ReadHandle::onInterest(const Name& prefix, const Interest& interest)
+{
+ Data data;
+ if (getStorageHandle().readData(interest, data)) {
+ getFace().put(data);
+ }
+}
+
+void
+ReadHandle::onRegisterFailed(const Name& prefix, const std::string& reason)
+{
+ std::cerr << "ERROR: Failed to register prefix in local hub's daemon" << std::endl;
+ getFace().shutdown();
+}
+
+void
+ReadHandle::listen(const Name& prefix)
+{
+ getFace().setInterestFilter(prefix,
+ bind(&ReadHandle::onInterest, this, _1, _2),
+ bind(&ReadHandle::onRegisterFailed, this, _1, _2));
+}
+
+} //namespace repo
diff --git a/src/handles/read-handle.hpp b/src/handles/read-handle.hpp
new file mode 100644
index 0000000..93468c0
--- /dev/null
+++ b/src/handles/read-handle.hpp
@@ -0,0 +1,52 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef REPO_HANDLES_READ_HANDLE_HPP
+#define REPO_HANDLES_READ_HANDLE_HPP
+
+#include "base-handle.hpp"
+
+namespace repo {
+
+class ReadHandle : public BaseHandle
+{
+
+public:
+ ReadHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain, Scheduler& scheduler)
+ : BaseHandle(face, storageHandle, keyChain, scheduler)
+ {
+ }
+
+ virtual void
+ listen(const Name& prefix);
+
+private:
+ /**
+ * @brief Read data from backend storage
+ */
+ void
+ onInterest(const Name& prefix, const Interest& interest);
+
+ void
+ onRegisterFailed(const Name& prefix, const std::string& reason);
+};
+
+} // namespace repo
+
+#endif // REPO_HANDLES_READ_HANDLE_HPP
diff --git a/src/handles/tcp-bulk-insert-handle.cpp b/src/handles/tcp-bulk-insert-handle.cpp
new file mode 100644
index 0000000..a69dfe2
--- /dev/null
+++ b/src/handles/tcp-bulk-insert-handle.cpp
@@ -0,0 +1,218 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tcp-bulk-insert-handle.hpp"
+
+namespace repo {
+
+const size_t MAX_NDN_PACKET_SIZE = 8800;
+
+namespace detail {
+
+class TcpBulkInsertClient : noncopyable
+{
+public:
+ TcpBulkInsertClient(TcpBulkInsertHandle& writer,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket)
+ : m_writer(writer)
+ , m_socket(socket)
+ , m_hasStarted(false)
+ , m_inputBufferSize(0)
+ {
+ }
+
+ static void
+ startReceive(const shared_ptr<TcpBulkInsertClient>& client)
+ {
+ BOOST_ASSERT(!client->m_hasStarted);
+
+ client->m_socket->async_receive(
+ boost::asio::buffer(client->m_inputBuffer, MAX_NDN_PACKET_SIZE), 0,
+ bind(&TcpBulkInsertClient::handleReceive, client, _1, _2, client));
+
+ client->m_hasStarted = true;
+ }
+
+private:
+ void
+ handleReceive(const boost::system::error_code& error,
+ std::size_t nBytesReceived,
+ const shared_ptr<TcpBulkInsertClient>& client);
+
+private:
+ TcpBulkInsertHandle& m_writer;
+ shared_ptr<boost::asio::ip::tcp::socket> m_socket;
+ bool m_hasStarted;
+ uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE];
+ std::size_t m_inputBufferSize;
+};
+
+} // namespace detail
+
+TcpBulkInsertHandle::TcpBulkInsertHandle(boost::asio::io_service& ioService,
+ StorageHandle& storageHandle)
+ : m_acceptor(ioService)
+ , m_storageHandle(storageHandle)
+{
+}
+
+void
+TcpBulkInsertHandle::listen(const std::string& host, const std::string& port)
+{
+ using namespace boost::asio;
+
+ ip::tcp::resolver resolver(m_acceptor.get_io_service());
+ ip::tcp::resolver::query query(host, port);
+
+ ip::tcp::resolver::iterator endpoint = resolver.resolve(query);
+ ip::tcp::resolver::iterator end;
+
+ if (endpoint == end)
+ throw Error("Cannot listen on [" + host + ":" + port + "]");
+
+ m_localEndpoint = *endpoint;
+ std::cerr << "Start listening on " << m_localEndpoint << std::endl;
+
+ m_acceptor.open(m_localEndpoint .protocol());
+ m_acceptor.set_option(ip::tcp::acceptor::reuse_address(true));
+ if (m_localEndpoint.address().is_v6())
+ {
+ m_acceptor.set_option(ip::v6_only(true));
+ }
+ m_acceptor.bind(m_localEndpoint);
+ m_acceptor.listen(255);
+
+ shared_ptr<ip::tcp::socket> clientSocket =
+ make_shared<ip::tcp::socket>(boost::ref(m_acceptor.get_io_service()));
+ m_acceptor.async_accept(*clientSocket,
+ bind(&TcpBulkInsertHandle::handleAccept, this, _1,
+ clientSocket));
+}
+
+void
+TcpBulkInsertHandle::stop()
+{
+ m_acceptor.cancel();
+ m_acceptor.close();
+}
+
+void
+TcpBulkInsertHandle::handleAccept(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket)
+{
+ using namespace boost::asio;
+
+ if (error) {
+ // if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+ // return;
+ return;
+ }
+
+ std::cerr << "New connection from " << socket->remote_endpoint() << std::endl;
+
+ shared_ptr<detail::TcpBulkInsertClient> client =
+ make_shared<detail::TcpBulkInsertClient>(boost::ref(*this), socket);
+ detail::TcpBulkInsertClient::startReceive(client);
+
+ // prepare accepting the next connection
+ shared_ptr<ip::tcp::socket> clientSocket =
+ make_shared<ip::tcp::socket>(boost::ref(m_acceptor.get_io_service()));
+ m_acceptor.async_accept(*clientSocket,
+ bind(&TcpBulkInsertHandle::handleAccept, this, _1,
+ clientSocket));
+}
+
+void
+detail::TcpBulkInsertClient::handleReceive(const boost::system::error_code& error,
+ std::size_t nBytesReceived,
+ const shared_ptr<detail::TcpBulkInsertClient>& client)
+{
+ if (error)
+ {
+ if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
+ return;
+
+ boost::system::error_code error;
+ m_socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, error);
+ m_socket->close(error);
+ return;
+ }
+
+ m_inputBufferSize += nBytesReceived;
+
+ // do magic
+
+ std::size_t offset = 0;
+
+ bool isOk = true;
+ Block element;
+ while (m_inputBufferSize - offset > 0)
+ {
+ isOk = Block::fromBuffer(m_inputBuffer + offset, m_inputBufferSize - offset, element);
+ if (!isOk)
+ break;
+
+ offset += element.size();
+ BOOST_ASSERT(offset <= m_inputBufferSize);
+
+ if (element.type() == ndn::Tlv::Data)
+ {
+ try {
+ Data data(element);
+ bool isOk = m_writer.getStorageHandle().insertData(data);
+ if (isOk)
+ std::cerr << "Successfully injected " << data.getName() << std::endl;
+ else
+ std::cerr << "FAILED to inject " << data.getName() << std::endl;
+ }
+ catch (std::runtime_error& error) {
+ /// \todo Catch specific error after determining what wireDecode() can throw
+ std::cerr << "Error decoding received Data packet" << std::endl;
+ }
+ }
+ }
+ if (!isOk && m_inputBufferSize == MAX_NDN_PACKET_SIZE && offset == 0)
+ {
+ boost::system::error_code error;
+ m_socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, error);
+ m_socket->close(error);
+ return;
+ }
+
+ if (offset > 0)
+ {
+ if (offset != m_inputBufferSize)
+ {
+ std::copy(m_inputBuffer + offset, m_inputBuffer + m_inputBufferSize,
+ m_inputBuffer);
+ m_inputBufferSize -= offset;
+ }
+ else
+ {
+ m_inputBufferSize = 0;
+ }
+ }
+
+ m_socket->async_receive(boost::asio::buffer(m_inputBuffer + m_inputBufferSize,
+ MAX_NDN_PACKET_SIZE - m_inputBufferSize), 0,
+ bind(&TcpBulkInsertClient::handleReceive, this, _1, _2, client));
+}
+
+
+} // namespace repo
diff --git a/src/handles/tcp-bulk-insert-handle.hpp b/src/handles/tcp-bulk-insert-handle.hpp
new file mode 100644
index 0000000..17233f4
--- /dev/null
+++ b/src/handles/tcp-bulk-insert-handle.hpp
@@ -0,0 +1,72 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef REPO_HANDLES_TCP_BULK_INSERT_HANDLE_HPP
+#define REPO_HANDLES_TCP_BULK_INSERT_HANDLE_HPP
+
+#include "common.hpp"
+#include "storage/storage-handle.hpp"
+
+#include <boost/asio.hpp>
+
+namespace repo {
+
+class TcpBulkInsertHandle : noncopyable
+{
+public:
+ class Error : public std::runtime_error
+ {
+ public:
+ explicit
+ Error(const std::string& what)
+ : std::runtime_error(what)
+ {
+ }
+ };
+
+public:
+ TcpBulkInsertHandle(boost::asio::io_service& ioService,
+ StorageHandle& storageHandle);
+
+ void
+ listen(const std::string& host, const std::string& port);
+
+ void
+ stop();
+
+ StorageHandle&
+ getStorageHandle()
+ {
+ return m_storageHandle;
+ }
+
+private:
+ void
+ handleAccept(const boost::system::error_code& error,
+ const shared_ptr<boost::asio::ip::tcp::socket>& socket);
+
+private:
+ boost::asio::ip::tcp::acceptor m_acceptor;
+ boost::asio::ip::tcp::endpoint m_localEndpoint;
+ StorageHandle& m_storageHandle;
+};
+
+} // namespace repo
+
+#endif // REPO_HANDLES_TCP_BULK_INSERT_HANDLE_HPP
diff --git a/src/handles/write-handle.cpp b/src/handles/write-handle.cpp
new file mode 100644
index 0000000..0ea17ae
--- /dev/null
+++ b/src/handles/write-handle.cpp
@@ -0,0 +1,546 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "write-handle.hpp"
+
+namespace repo {
+
+static const int RETRY_TIMEOUT = 3;
+static const int DEFAULT_CREDIT = 12;
+static const ndn::time::milliseconds NOEND_TIMEOUT(10000);
+static const ndn::time::milliseconds PROCESS_DELETE_TIME(10000);
+
+WriteHandle::WriteHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain,
+ Scheduler& scheduler, CommandInterestValidator& validator)
+ : BaseHandle(face, storageHandle, keyChain, scheduler)
+ , m_validator(validator)
+ , m_retryTime(RETRY_TIMEOUT)
+ , m_credit(DEFAULT_CREDIT)
+ , m_noEndTimeout(NOEND_TIMEOUT)
+{
+}
+
+void
+WriteHandle::deleteProcess(ProcessId processId)
+{
+ m_processes.erase(processId);
+}
+
+// Interest.
+void
+WriteHandle::onInterest(const Name& prefix, const Interest& interest)
+{
+ m_validator.validate(interest,
+ bind(&WriteHandle::onValidated, this, _1, prefix),
+ bind(&WriteHandle::onValidationFailed, this, _1));
+}
+
+// onRegisterFailed.
+void
+WriteHandle::onRegisterFailed(const Name& prefix, const std::string& reason)
+{
+ throw Error("Insert prefix registration failed");
+}
+
+// onRegisterFailed for insert.
+void
+WriteHandle::onCheckRegisterFailed(const Name& prefix, const std::string& reason)
+{
+ throw Error("Insert check prefix registration failed");
+}
+
+void
+WriteHandle::onValidated(const shared_ptr<const Interest>& interest, const Name& prefix)
+{
+ //m_validResult = 1;
+ RepoCommandParameter parameter;
+ try {
+ extractParameter(*interest, prefix, parameter);
+ }
+ catch (RepoCommandParameter::Error) {
+ negativeReply(*interest, 403);
+ return;
+ }
+
+ if (parameter.hasStartBlockId() || parameter.hasEndBlockId()) {
+ if (parameter.hasSelectors()) {
+ negativeReply(*interest, 402);
+ return;
+ }
+ processSegmentedInsertCommand(*interest, parameter);
+ }
+ else {
+ processSingleInsertCommand(*interest, parameter);
+ }
+
+}
+
+void
+WriteHandle::onValidationFailed(const shared_ptr<const Interest>& interest)
+{
+ std::cout << "invalidated" << std::endl;
+ negativeReply(*interest, 401);
+}
+
+void
+WriteHandle::onData(const Interest& interest, ndn::Data& data, ProcessId processId)
+{
+ //std::cout << "onData" << std::endl;
+ //std::cout << "I: " << interest.toUri() << std::endl;
+ //std::cout << "D: " << data.getName().toUri() << std::endl;
+ if (m_processes.count(processId) == 0) {
+ return;
+ }
+
+ ProcessInfo& process = m_processes[processId];
+ RepoCommandResponse& response = process.response;
+
+ if (response.getInsertNum() == 0) {
+ getStorageHandle().insertData(data);
+ response.setInsertNum(1);
+ }
+
+ deferredDeleteProcess(processId);
+}
+
+void
+WriteHandle::onSegmentData(const Interest& interest, Data& data, ProcessId processId)
+{
+ //std::cout << "I: " << interest.toUri() << std::endl;
+ //std::cout << "D: " << data.getName().toUri() << std::endl;
+ //retrieve the process from the responsemap
+
+ if (m_processes.count(processId) == 0) {
+ return;
+ }
+ RepoCommandResponse& response = m_processes[processId].response;
+
+ //refresh endBlockId
+ Name::Component finalBlockId = data.getFinalBlockId();
+
+ if (!finalBlockId.empty()) {
+ SegmentNo final = finalBlockId.toSegment();
+ if (response.hasEndBlockId()) {
+ if (final < response.getEndBlockId()) {
+ response.setEndBlockId(final);
+ }
+ }
+ else {
+ response.setEndBlockId(final);
+ }
+ }
+
+ //insert data
+ //std::cout << "start to insert" << std::endl;
+ if (getStorageHandle().insertData(data)) {
+ response.setInsertNum(response.getInsertNum() + 1);
+ }
+ //std::cout << "end of insert" << std::endl;
+
+ //it->second = response;
+
+ onSegmentDataControl(processId, interest);
+}
+
+void
+WriteHandle::onTimeout(const ndn::Interest& interest, ProcessId processId)
+{
+ std::cout << "Timeout" << std::endl;
+ m_processes.erase(processId);
+}
+
+void
+WriteHandle::onSegmentTimeout(const Interest& interest, ProcessId processId)
+{
+ std::cout << "SegTimeout" << std::endl;
+
+ onSegmentTimeoutControl(processId, interest);
+}
+
+void
+WriteHandle::listen(const Name& prefix)
+{
+ Name insertPrefix;
+ insertPrefix.append(prefix).append("insert");
+ getFace().setInterestFilter(insertPrefix,
+ bind(&WriteHandle::onInterest, this, _1, _2),
+ bind(&WriteHandle::onRegisterFailed, this, _1, _2));
+ Name insertCheckPrefix;
+ insertCheckPrefix.append(prefix).append("insert check");
+ getFace().setInterestFilter(insertCheckPrefix,
+ bind(&WriteHandle::onCheckInterest, this, _1, _2),
+ bind(&WriteHandle::onRegisterFailed, this, _1, _2));
+}
+
+void
+WriteHandle::segInit(ProcessId processId, const RepoCommandParameter& parameter)
+{
+ ProcessInfo& process = m_processes[processId];
+ process.credit = 0;
+
+ map<SegmentNo, int>& processRetry = process.retryCounts;
+
+ Name name = parameter.getName();
+ SegmentNo startBlockId = parameter.getStartBlockId();
+
+ uint64_t initialCredit = m_credit;
+
+ if (parameter.hasEndBlockId()) {
+ initialCredit =
+ std::min(initialCredit, parameter.getEndBlockId() - parameter.getStartBlockId() + 1);
+ }
+ else {
+ // set noEndTimeout timer
+ process.noEndTime = ndn::time::steady_clock::now() +
+ m_noEndTimeout;
+ }
+ process.credit = initialCredit;
+ SegmentNo segment = startBlockId;
+ for (; segment < startBlockId + initialCredit; ++segment) {
+ Name fetchName = name;
+ fetchName.appendSegment(segment);
+ Interest interest(fetchName);
+ //std::cout << "seg:" << j<<std::endl;
+ getFace().expressInterest(interest,
+ bind(&WriteHandle::onSegmentData, this, _1, _2, processId),
+ bind(&WriteHandle::onSegmentTimeout, this, _1, processId));
+ process.credit--;
+ processRetry[segment] = 0;
+ }
+
+ queue<SegmentNo>& nextSegmentQueue = process.nextSegmentQueue;
+
+ process.nextSegment = segment;
+ nextSegmentQueue.push(segment);
+}
+
+void
+WriteHandle::onSegmentDataControl(ProcessId processId, const Interest& interest)
+{
+ //std::cout << "onSegmentDataControl: " << processId << std::endl;
+
+ if (m_processes.count(processId) == 0) {
+ return;
+ }
+ ProcessInfo& process = m_processes[processId];
+ RepoCommandResponse& response = process.response;
+ int& processCredit = process.credit;
+ //onSegmentDataControl is called when a data returns.
+ //When data returns, processCredit++
+ processCredit++;
+ SegmentNo& nextSegment = process.nextSegment;
+ queue<SegmentNo>& nextSegmentQueue = process.nextSegmentQueue;
+ map<SegmentNo, int>& retryCounts = process.retryCounts;
+
+ //read whether notime timeout
+ if (!response.hasEndBlockId()) {
+
+ ndn::time::steady_clock::TimePoint& noEndTime = process.noEndTime;
+ ndn::time::steady_clock::TimePoint now = ndn::time::steady_clock::now();
+
+ if (now > noEndTime) {
+ std::cout << "noEndtimeout: " << processId << std::endl;
+ //m_processes.erase(processId);
+ //StatusCode should be refreshed as 405
+ response.setStatusCode(405);
+ //schedule a delete event
+ deferredDeleteProcess(processId);
+ return;
+ }
+ }
+
+ //read whether this process has total ends, if ends, remove control info from the maps
+ if (response.hasEndBlockId()) {
+ uint64_t nSegments =
+ response.getEndBlockId() - response.getStartBlockId() + 1;
+ if (response.getInsertNum() >= nSegments) {
+ //m_processes.erase(processId);
+ //All the data has been inserted, StatusCode is refreshed as 200
+ response.setStatusCode(200);
+ deferredDeleteProcess(processId);
+ return;
+ }
+ }
+
+ //check whether there is any credit
+ if (processCredit == 0)
+ return;
+
+
+ //check whether sent queue empty
+ if (nextSegmentQueue.empty()) {
+ //do not do anything
+ return;
+ }
+
+ //pop the queue
+ SegmentNo sendingSegment = nextSegmentQueue.front();
+ nextSegmentQueue.pop();
+
+ //check whether sendingSegment exceeds
+ if (response.hasEndBlockId() && sendingSegment > response.getEndBlockId()) {
+ //do not do anything
+ return;
+ }
+
+ //read whether this is retransmitted data;
+ SegmentNo fetchedSegment =
+ interest.getName().get(interest.getName().size() - 1).toSegment();
+
+ BOOST_ASSERT(retryCounts.count(fetchedSegment) != 0);
+
+ //find this fetched data, remove it from this map
+ //rit->second.erase(oit);
+ retryCounts.erase(fetchedSegment);
+ //express the interest of the top of the queue
+ Name fetchName(interest.getName().getPrefix(-1));
+ fetchName.appendSegment(sendingSegment);
+ Interest fetchInterest(fetchName);
+ getFace().expressInterest(fetchInterest,
+ bind(&WriteHandle::onSegmentData, this, _1, _2, processId),
+ bind(&WriteHandle::onSegmentTimeout, this, _1, processId));
+ //When an interest is expressed, processCredit--
+ processCredit--;
+ //std::cout << "sent seg: " << sendingSegment << std::endl;
+ if (retryCounts.count(sendingSegment) == 0) {
+ //not found
+ retryCounts[sendingSegment] = 0;
+ }
+ else {
+ //found
+ retryCounts[sendingSegment] = retryCounts[sendingSegment] + 1;
+ }
+ //increase the next seg and put it into the queue
+ if (!response.hasEndBlockId() || (nextSegment + 1) <= response.getEndBlockId()) {
+ nextSegment++;
+ nextSegmentQueue.push(nextSegment);
+ }
+}
+
+void
+WriteHandle::onSegmentTimeoutControl(ProcessId processId, const Interest& interest)
+{
+ if (m_processes.count(processId) == 0) {
+ return;
+ }
+ ProcessInfo& process = m_processes[processId];
+ // RepoCommandResponse& response = process.response;
+ // SegmentNo& nextSegment = process.nextSegment;
+ // queue<SegmentNo>& nextSegmentQueue = process.nextSegmentQueue;
+ map<SegmentNo, int>& retryCounts = process.retryCounts;
+
+ SegmentNo timeoutSegment = interest.getName().get(-1).toSegment();
+
+ std::cout << "timeoutSegment: " << timeoutSegment << std::endl;
+
+ BOOST_ASSERT(retryCounts.count(timeoutSegment) != 0);
+
+ //read the retry time. If retry out of time, fail the process. if not, plus
+ int& retryTime = retryCounts[timeoutSegment];
+ if (retryTime >= m_retryTime) {
+ //fail this process
+ std::cout << "Retry timeout: " << processId << std::endl;
+ m_processes.erase(processId);
+ return;
+ }
+ else {
+ //Reput it in the queue, retryTime++
+ retryTime++;
+ Interest retryInterest(interest.getName());
+ getFace().expressInterest(retryInterest,
+ bind(&WriteHandle::onSegmentData, this, _1, _2, processId),
+ bind(&WriteHandle::onSegmentTimeout, this, _1, processId));
+ }
+
+}
+
+void
+WriteHandle::onCheckInterest(const Name& prefix, const Interest& interest)
+{
+ m_validator.validate(interest,
+ bind(&WriteHandle::onCheckValidated, this, _1, prefix),
+ bind(&WriteHandle::onCheckValidationFailed, this, _1));
+
+}
+
+void
+WriteHandle::onCheckValidated(const shared_ptr<const Interest>& interest, const Name& prefix)
+{
+ RepoCommandParameter parameter;
+ try {
+ extractParameter(*interest, prefix, parameter);
+ }
+ catch (RepoCommandParameter::Error) {
+ negativeReply(*interest, 403);
+ return;
+ }
+
+ if (!parameter.hasProcessId()) {
+ negativeReply(*interest, 403);
+ return;
+ }
+ //check whether this process exists
+ ProcessId processId = parameter.getProcessId();
+ if (m_processes.count(processId) == 0) {
+ std::cout << "no such processId: " << processId << std::endl;
+ negativeReply(*interest, 404);
+ return;
+ }
+
+ ProcessInfo& process = m_processes[processId];
+
+ RepoCommandResponse& response = process.response;
+
+ //Check whether it is single data fetching
+ if (!response.hasStartBlockId() &&
+ !response.hasEndBlockId()) {
+ reply(*interest, response);
+ return;
+ }
+
+ //read if noEndtimeout
+ if (!response.hasEndBlockId()) {
+ extendNoEndTime(process);
+ reply(*interest, response);
+ return;
+ }
+ else {
+ reply(*interest, response);
+ }
+}
+
+void
+WriteHandle::onCheckValidationFailed(const shared_ptr<const Interest>& interest)
+{
+ negativeReply(*interest, 401);
+}
+
+void
+WriteHandle::deferredDeleteProcess(ProcessId processId)
+{
+ getScheduler().scheduleEvent(PROCESS_DELETE_TIME,
+ ndn::bind(&WriteHandle::deleteProcess, this, processId));
+}
+
+void
+WriteHandle::processSingleInsertCommand(const Interest& interest,
+ RepoCommandParameter& parameter)
+{
+ ProcessId processId = generateProcessId();
+
+ ProcessInfo& process = m_processes[processId];
+
+ RepoCommandResponse& response = process.response;
+ response.setStatusCode(100);
+ response.setProcessId(processId);
+ response.setInsertNum(0);
+
+ reply(interest, response);
+
+ response.setStatusCode(300);
+
+ Interest fetchInterest(parameter.getName());
+ if (parameter.hasSelectors()) {
+ fetchInterest.setSelectors(parameter.getSelectors());
+ }
+ getFace().expressInterest(fetchInterest,
+ bind(&WriteHandle::onData, this, _1, _2, processId),
+ bind(&WriteHandle::onTimeout, this, _1, processId));
+}
+
+void
+WriteHandle::processSegmentedInsertCommand(const Interest& interest,
+ RepoCommandParameter& parameter)
+{
+ if (parameter.hasEndBlockId()) {
+ //normal fetch segment
+ if (!parameter.hasStartBlockId()) {
+ parameter.setStartBlockId(0);
+ }
+
+ SegmentNo startBlockId = parameter.getStartBlockId();
+ SegmentNo endBlockId = parameter.getEndBlockId();
+ //std::cout << "startBlockId: " << startBlockId << std::endl;
+ //std::cout << "endBlockId: " << endBlockId << std::endl;
+ if (startBlockId > endBlockId) {
+ negativeReply(interest, 403);
+ return;
+ }
+
+ ProcessId processId = generateProcessId();
+ ProcessInfo& process = m_processes[processId];
+ //std::cout << "processId: " << processId << std::endl;
+ RepoCommandResponse& response = process.response;
+ response.setStatusCode(100);
+ response.setProcessId(processId);
+ response.setInsertNum(0);
+ response.setStartBlockId(startBlockId);
+ response.setEndBlockId(endBlockId);
+
+ reply(interest, response);
+
+ //300 means data fetching is in progress
+ response.setStatusCode(300);
+
+ segInit(processId, parameter);
+ }
+ else {
+ //no EndBlockId, so fetch FinalBlockId in data, if timeout, stop
+ ProcessId processId = generateProcessId();
+ ProcessInfo& process = m_processes[processId];
+ //std::cout << "processId: " << processId << std::endl;
+ RepoCommandResponse& response = process.response;
+ response.setStatusCode(100);
+ response.setProcessId(processId);
+ response.setInsertNum(0);
+ response.setStartBlockId(parameter.getStartBlockId());
+ reply(interest, response);
+
+ //300 means data fetching is in progress
+ response.setStatusCode(300);
+
+ segInit(processId, parameter);
+ }
+}
+
+void
+WriteHandle::extendNoEndTime(ProcessInfo& process)
+{
+ ndn::time::steady_clock::TimePoint& noEndTime = process.noEndTime;
+ ndn::time::steady_clock::TimePoint now = ndn::time::steady_clock::now();
+ RepoCommandResponse& response = process.response;
+ if (now > noEndTime) {
+ response.setStatusCode(405);
+ return;
+ }
+ //extends noEndTime
+ process.noEndTime =
+ ndn::time::steady_clock::now() + m_noEndTimeout;
+
+}
+
+void
+WriteHandle::negativeReply(const Interest& interest, int statusCode)
+{
+ RepoCommandResponse response;
+ response.setStatusCode(statusCode);
+ reply(interest, response);
+}
+
+} //namespace repo
diff --git a/src/handles/write-handle.hpp b/src/handles/write-handle.hpp
new file mode 100644
index 0000000..075e78c
--- /dev/null
+++ b/src/handles/write-handle.hpp
@@ -0,0 +1,225 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2014, Regents of the University of California.
+ *
+ * This file is part of NDN repo-ng (Next generation of NDN repository).
+ * See AUTHORS.md for complete list of repo-ng authors and contributors.
+ *
+ * repo-ng is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * repo-ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE. See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * repo-ng, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef REPO_HANDLES_WRITE_HANDLE_HPP
+#define REPO_HANDLES_WRITE_HANDLE_HPP
+
+#include "base-handle.hpp"
+
+#include <queue>
+
+namespace repo {
+
+using std::map;
+using std::pair;
+using std::queue;
+
+/**
+ * @brief WriteHandle provides basic credit based congestion control.
+ *
+ * First repo sends interests of credit number and then credit will be 0.
+ *
+ * If a data comes, credit++ and sends a interest then credit--.
+ *
+ * If the interest timeout, repo will retry and send interest in retrytimes.
+ *
+ * If one interest timeout beyond retrytimes, the fetching process will terminate.
+ *
+ * Another case is that if command will insert segmented data without EndBlockId.
+ *
+ * The repo will keep fetching data in noendTimeout time.
+ *
+ * If data returns with FinalBlockId, this detecting timeout process will terminate.
+ *
+ * If client sends a insert check command, the noendTimeout timer will be set to 0.
+ *
+ * If repo cannot get FinalBlockId in noendTimeout time, the fetching process will terminate.
+ */
+class WriteHandle : public BaseHandle
+{
+
+public:
+ class Error : public BaseHandle::Error
+ {
+ public:
+ explicit
+ Error(const std::string& what)
+ : BaseHandle::Error(what)
+ {
+ }
+ };
+
+
+public:
+ WriteHandle(Face& face, StorageHandle& storageHandle, KeyChain& keyChain,
+ Scheduler& scheduler, CommandInterestValidator& validator);
+
+ virtual void
+ listen(const Name& prefix);
+
+private:
+ /**
+ * @brief Information of insert process including variables for response
+ * and credit based flow control
+ */
+ struct ProcessInfo
+ {
+ //ProcessId id;
+ RepoCommandResponse response;
+ queue<SegmentNo> nextSegmentQueue; ///< queue of waiting segment
+ /// to be sent when having credits
+ SegmentNo nextSegment; ///< last segment put into the nextSegmentQueue
+ map<SegmentNo, int> retryCounts; ///< to store retrying times of timeout segment
+ int credit; ///< congestion control credits of process
+
+ /**
+ * @brief the latest time point at which EndBlockId must be determined
+ *
+ * Segmented fetch process will terminate if EndBlockId cannot be
+ * determined before this time point.
+ * It is initialized to now()+noEndTimeout when segmented fetch process begins,
+ * and reset to now()+noEndTimeout each time an insert status check command is processed.
+ */
+ ndn::time::steady_clock::TimePoint noEndTime;
+ };
+
+private: // insert command
+ /**
+ * @brief handle insert commands
+ */
+ void
+ onInterest(const Name& prefix, const Interest& interest);
+
+ void
+ onValidated(const shared_ptr<const Interest>& interest, const Name& prefix);
+
+ void
+ onValidationFailed(const shared_ptr<const Interest>& interest);
+
+ /**
+ * @brief insert command prefix register failed
+ */
+ void
+ onRegisterFailed(const Name& prefix, const std::string& reason);
+
+private: // single data fetching
+ /**
+ * @brief fetch one data
+ */
+ void
+ onData(const Interest& interest, Data& data, ProcessId processId);
+
+ /**
+ * @brief handle when fetching one data timeout
+ */
+ void
+ onTimeout(const Interest& interest, ProcessId processId);
+
+ void
+ processSingleInsertCommand(const Interest& interest, RepoCommandParameter& parameter);
+
+private: // segmented data fetching
+ /**
+ * @brief fetch segmented data
+ */
+ void
+ onSegmentData(const Interest& interest, Data& data, ProcessId processId);
+
+ /**
+ * @brief Timeout when fetching segmented data. Data can be fetched RETRY_TIMEOUT times.
+ */
+ void
+ onSegmentTimeout(const Interest& interest, ProcessId processId);
+
+ /**
+ * @brief initiate fetching segmented data
+ */
+ void
+ segInit(ProcessId processId, const RepoCommandParameter& parameter);
+
+ /**
+ * @brief control for sending interests in function onSegmentData()
+ */
+ void
+ onSegmentDataControl(ProcessId processId, const Interest& interest);
+
+ /**
+ * @brief control for sending interest in function onSegmentTimeout
+ */
+ void
+ onSegmentTimeoutControl(ProcessId processId, const Interest& interest);
+
+ void
+ processSegmentedInsertCommand(const Interest& interest, RepoCommandParameter& parameter);
+
+ /**
+ * @brief extends noEndTime of process if not noEndTimeout, set StatusCode 405
+ *
+ * called by onCheckValidated() if there is no EndBlockId. If not noEndTimeout,
+ * extends noEndTime of process. If noEndTimeout, set status code 405 as noEndTimeout.
+ */
+ void
+ extendNoEndTime(ProcessInfo& process);
+
+private: // insert state check command
+ /**
+ * @brief handle insert check command
+ */
+ void
+ onCheckInterest(const Name& prefix, const Interest& interest);
+
+ /**
+ * @brief insert check command prefix register failed
+ */
+ void
+ onCheckRegisterFailed(const Name& prefix, const std::string& reason);
+
+ void
+ onCheckValidated(const shared_ptr<const Interest>& interest, const Name& prefix);
+
+ void
+ onCheckValidationFailed(const shared_ptr<const Interest>& interest);
+
+private:
+ void
+ deleteProcess(ProcessId processId);
+
+ /**
+ * @brief schedule a event to delete the process
+ */
+ void
+ deferredDeleteProcess(ProcessId processId);
+
+ void
+ negativeReply(const Interest& interest, int statusCode);
+
+private:
+
+ CommandInterestValidator& m_validator;
+
+ map<ProcessId, ProcessInfo> m_processes;
+
+ int m_retryTime;
+ int m_credit;
+ ndn::time::milliseconds m_noEndTimeout;
+};
+
+} // namespace repo
+
+#endif // REPO_HANDLES_WRITE_HANDLE_HPP