Implementation of TCPTransport

include/f1x/aasdk/TCP/ITCPEndpoint.hpp

@@ -15,7 +15,7 @@ class ITCPEndpoint
 {
 public:
     typedef std::shared_ptr<ITCPEndpoint> Pointer;
-    typedef io::Promise<size_t> Promise;
+    typedef io::Promise<void> Promise;
 
     virtual ~ITCPEndpoint() = default;
 

include/f1x/aasdk/Transport/TCPTransport.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+#include <list>
+#include <f1x/aasdk/TCP/ITCPEndpoint.hpp>
+#include <f1x/aasdk/Transport/ITransport.hpp>
+
+namespace f1x
+{
+namespace aasdk
+{
+namespace transport
+{
+
+class TCPTransport: public ITransport, public std::enable_shared_from_this<TCPTransport>, boost::noncopyable
+{
+public:
+    TCPTransport(boost::asio::io_service& ioService, tcp::ITCPEndpoint::Pointer tcpEndpoint);
+
+    void receive(size_t size, ReceivePromise::Pointer promise) override;
+    void send(common::Data data, SendPromise::Pointer promise) override;
+    void stop() override;
+
+private:
+    typedef std::list<std::pair<size_t, ReceivePromise::Pointer>> ReceiveQueue;
+    typedef std::list<std::pair<common::Data, SendPromise::Pointer>> SendQueue;
+
+    using std::enable_shared_from_this<TCPTransport>::shared_from_this;
+    void doReceive();
+    void receiveHandler(ReceiveQueue::iterator queueElement);
+    void receiveFailureHandler(const aasdk::error::Error& e, ReceiveQueue::iterator queueElement);
+
+    void doSend();
+    void sendHandler(SendQueue::iterator queueElement);
+    void sendFailureHandler(const aasdk::error::Error& e, SendQueue::iterator queueElement);
+
+    boost::asio::io_service& ioService_;
+    tcp::ITCPEndpoint::Pointer tcpEndpoint_;
+
+    boost::asio::io_service::strand receiveStrand_;
+    ReceiveQueue receiveQueue_;
+    common::Data receiveData_;
+
+    boost::asio::io_service::strand sendStrand_;
+    SendQueue sendQueue_;
+};
+
+}
+}
+}

src/TCP/TCPEndpoint.cpp

@@ -41,11 +41,11 @@ void TCPEndpoint::stop()
     socket_.close(ec);
 }
 
-void TCPEndpoint::asyncOperationHandler(const boost::system::error_code& ec, size_t bytesTransferred, Promise::Pointer promise)
+void TCPEndpoint::asyncOperationHandler(const boost::system::error_code& ec, size_t, Promise::Pointer promise)
 {
     if(!ec)
     {
-        promise->resolve(bytesTransferred);
+        promise->resolve();
     }
     else
     {

src/Transport/TCPTransport.cpp

@@ -0,0 +1,115 @@
+#include <f1x/aasdk/Transport/TCPTransport.hpp>
+
+namespace f1x
+{
+namespace aasdk
+{
+namespace transport
+{
+
+TCPTransport::TCPTransport(boost::asio::io_service& ioService, tcp::ITCPEndpoint::Pointer tcpEndpoint)
+    : ioService_(ioService)
+    , tcpEndpoint_(std::move(tcpEndpoint))
+    , receiveStrand_(ioService_)
+    , sendStrand_(ioService_)
+{
+
+}
+
+void TCPTransport::receive(size_t size, ReceivePromise::Pointer promise)
+{
+    receiveStrand_.dispatch([this, self = this->shared_from_this(), size, promise = std::move(promise)]() mutable {
+        receiveQueue_.emplace_back(std::make_pair(size, std::move(promise)));
+
+        if(receiveQueue_.size() == 1)
+        {
+            this->doReceive();
+        }
+    });
+}
+
+void TCPTransport::send(common::Data data, SendPromise::Pointer promise)
+{
+    sendStrand_.dispatch([this, self = this->shared_from_this(), data = std::move(data), promise = std::move(promise)]() mutable {
+        sendQueue_.emplace_back(std::make_pair(std::move(data), std::move(promise)));
+
+        if(sendQueue_.size() == 1)
+        {
+            this->doSend();
+        }
+    });
+}
+
+void TCPTransport::stop()
+{
+    tcpEndpoint_->stop();
+}
+
+void TCPTransport::doReceive()
+{
+    auto queueElement = receiveQueue_.begin();
+    receiveData_.resize(queueElement->first);
+
+    auto receivePromise = tcp::ITCPEndpoint::Promise::defer(receiveStrand_);
+    receivePromise->then(std::bind(&TCPTransport::receiveHandler, this->shared_from_this(), queueElement),
+                         std::bind(&TCPTransport::receiveFailureHandler, this->shared_from_this(), std::placeholders::_1, queueElement));
+    tcpEndpoint_->receive(common::DataBuffer(receiveData_), std::move(receivePromise));
+}
+
+void TCPTransport::receiveHandler(ReceiveQueue::iterator queueElement)
+{
+    queueElement->second->resolve(std::move(receiveData_));
+    receiveQueue_.erase(queueElement);
+
+    if(!receiveQueue_.empty())
+    {
+        this->doReceive();
+    }
+}
+
+void TCPTransport::receiveFailureHandler(const aasdk::error::Error& e, ReceiveQueue::iterator queueElement)
+{
+    queueElement->second->reject(e);
+    receiveQueue_.erase(queueElement);
+
+    if(!receiveQueue_.empty())
+    {
+        this->doReceive();
+    }
+}
+
+void TCPTransport::doSend()
+{
+    auto queueElement = sendQueue_.begin();
+
+    auto sendPromise = tcp::ITCPEndpoint::Promise::defer(sendStrand_);
+    sendPromise->then(std::bind(&TCPTransport::sendHandler, this->shared_from_this(), queueElement),
+                      std::bind(&TCPTransport::sendFailureHandler, this->shared_from_this(), std::placeholders::_1, queueElement));
+    tcpEndpoint_->send(common::DataConstBuffer(queueElement->first), std::move(sendPromise));
+}
+
+void TCPTransport::sendHandler(SendQueue::iterator queueElement)
+{
+    queueElement->second->resolve();
+    sendQueue_.erase(queueElement);
+
+    if(!sendQueue_.empty())
+    {
+        this->doSend();
+    }
+}
+
+void TCPTransport::sendFailureHandler(const aasdk::error::Error& e, SendQueue::iterator queueElement)
+{
+    queueElement->second->reject(e);
+    sendQueue_.erase(queueElement);
+
+    if(!sendQueue_.empty())
+    {
+        this->doSend();
+    }
+}
+
+}
+}
+}