Create base class for transports

include/f1x/aasdk/TCP/ITCPEndpoint.hpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #pragma once
 
 #include <memory>
@@ -15,7 +33,7 @@ class ITCPEndpoint
 {
 public:
     typedef std::shared_ptr<ITCPEndpoint> Pointer;
-    typedef io::Promise<void> Promise;
+    typedef io::Promise<size_t> Promise;
 
     virtual ~ITCPEndpoint() = default;
 

include/f1x/aasdk/TCP/ITCPWrapper.hpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #pragma once
 
 #include <functional>

include/f1x/aasdk/TCP/TCPEndpoint.hpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #pragma once
 
 #include <boost/asio/ip/tcp.hpp>

include/f1x/aasdk/TCP/TCPWrapper.hpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #pragma once
 
 #include <f1x/aasdk/TCP/ITCPWrapper.hpp>

include/f1x/aasdk/Transport/TCPTransport.hpp

@@ -1,7 +1,26 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #pragma once
 
 #include <list>
 #include <f1x/aasdk/TCP/ITCPEndpoint.hpp>
+#include <f1x/aasdk/Transport/DataSink.hpp>
 #include <f1x/aasdk/Transport/ITransport.hpp>
 
 namespace f1x
@@ -25,12 +44,12 @@ private:
     typedef std::list<std::pair<common::Data, SendPromise::Pointer>> SendQueue;
 
     using std::enable_shared_from_this<TCPTransport>::shared_from_this;
-    void doReceive();
+    void distributeReceivedData();
-    void receiveHandler(ReceiveQueue::iterator queueElement);
+    void receiveHandler(size_t bytesTransferred);
-    void receiveFailureHandler(const aasdk::error::Error& e, ReceiveQueue::iterator queueElement);
+    void receiveFailureHandler(const aasdk::error::Error& e);
 
     void doSend();
-    void sendHandler(SendQueue::iterator queueElement);
+    void sendHandler(size_t bytesTransferred, SendQueue::iterator queueElement);
     void sendFailureHandler(const aasdk::error::Error& e, SendQueue::iterator queueElement);
 
     boost::asio::io_service& ioService_;
@@ -38,7 +57,7 @@ private:
 
     boost::asio::io_service::strand receiveStrand_;
     ReceiveQueue receiveQueue_;
-    common::Data receiveData_;
+    DataSink tcpReceivedDataSink_;
 
     boost::asio::io_service::strand sendStrand_;
     SendQueue sendQueue_;

include/f1x/aasdk/Transport/Transport.hpp

@@ -0,0 +1,65 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#pragma once
+
+#include <list>
+#include <queue>
+#include <boost/asio.hpp>
+#include <f1x/aasdk/Transport/ITransport.hpp>
+#include <f1x/aasdk/Transport/DataSink.hpp>
+
+namespace f1x
+{
+namespace aasdk
+{
+namespace transport
+{
+
+class Transport: public ITransport, public std::enable_shared_from_this<Transport>, boost::noncopyable
+{
+public:
+    Transport(boost::asio::io_service& ioService);
+
+    void receive(size_t size, ReceivePromise::Pointer promise) override;
+    void send(common::Data data, SendPromise::Pointer promise) override;
+
+protected:
+    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<Transport>::shared_from_this;
+    void receiveHandler(size_t bytesTransferred);
+    void distributeReceivedData();
+    void rejectReceivePromises(const error::Error& e);
+
+    virtual void enqueueReceive(common::DataBuffer buffer) = 0;
+    virtual void enqueueSend(SendQueue::iterator queueElement) = 0;
+
+    DataSink receivedDataSink_;
+
+    boost::asio::io_service::strand receiveStrand_;
+    ReceiveQueue receiveQueue_;
+
+    boost::asio::io_service::strand sendStrand_;
+    SendQueue sendQueue_;
+};
+
+}
+}
+}

include/f1x/aasdk/Transport/USBTransport.hpp

@@ -21,7 +21,7 @@
 #include <list>
 #include <queue>
 #include <boost/asio.hpp>
-#include <f1x/aasdk/Transport/ITransport.hpp>
+#include <f1x/aasdk/Transport/Transport.hpp>
 #include <f1x/aasdk/Transport/DataSink.hpp>
 #include <f1x/aasdk/USB/IAOAPDevice.hpp>
 
@@ -32,35 +32,21 @@ namespace aasdk
 namespace transport
 {
 
-class USBTransport: public ITransport, public std::enable_shared_from_this<USBTransport>, boost::noncopyable
+class USBTransport: public Transport
 {
 public:
     USBTransport(boost::asio::io_service& ioService, usb::IAOAPDevice::Pointer aoapDevice);
 
-    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>> InTransferQueue;
+    void enqueueReceive(common::DataBuffer buffer) override;
-    typedef std::list<std::pair<common::Data, SendPromise::Pointer>> OutTransferQueue;
+    void enqueueSend(SendQueue::iterator queueElement) override;
-
-    using std::enable_shared_from_this<USBTransport>::shared_from_this;
     void receiveHandler(size_t bytesTransferred);
-    void distributeReceivedData();
+    void doSend(SendQueue::iterator queueElement, common::Data::size_type offset);
-    void rejectReceivePromises(const error::Error& e);
+    void sendHandler(SendQueue::iterator queueElement, common::Data::size_type offset, size_t bytesTransferred);
-
-    void doSend(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset);
-    void sendHandler(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset, size_t bytesTransferred);
 
     usb::IAOAPDevice::Pointer aoapDevice_;
-    DataSink usbReceivedDataSink_;
-
-    boost::asio::io_service::strand receiveStrand_;
-    InTransferQueue receiveQueue_;
-
-    boost::asio::io_service::strand sendStrand_;
-    OutTransferQueue sendQueue_;
 
     static constexpr uint32_t cSendTimeoutMs = 10000;
     static constexpr uint32_t cReceiveTimeoutMs = 0;

src/TCP/TCPEndpoint.cpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #include <f1x/aasdk/TCP/TCPEndpoint.hpp>
 
 namespace f1x
@@ -39,11 +57,11 @@ void TCPEndpoint::stop()
     tcpWrapper_.close(socket_);
 }
 
-void TCPEndpoint::asyncOperationHandler(const boost::system::error_code& ec, size_t, Promise::Pointer promise)
+void TCPEndpoint::asyncOperationHandler(const boost::system::error_code& ec, size_t bytesTransferred, Promise::Pointer promise)
 {
     if(!ec)
     {
-        promise->resolve();
+        promise->resolve(bytesTransferred);
     }
     else
     {

src/TCP/TCPWrapper.cpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #include <boost/asio.hpp>
 #include <f1x/aasdk/TCP/TCPWrapper.hpp>
 
@@ -15,7 +33,7 @@ void TCPWrapper::asyncWrite(boost::asio::ip::tcp::socket& socket, common::DataCo
 
 void TCPWrapper::asyncRead(boost::asio::ip::tcp::socket& socket, common::DataBuffer buffer, Handler handler)
 {
-    boost::asio::async_read(socket, boost::asio::buffer(buffer.data, buffer.size), std::move(handler));
+    socket.async_receive(boost::asio::buffer(buffer.data, buffer.size), std::move(handler));
 }
 
 void TCPWrapper::close(boost::asio::ip::tcp::socket& socket)

src/Transport/TCPTransport.cpp

@@ -1,3 +1,21 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
 #include <f1x/aasdk/Transport/TCPTransport.hpp>
 
 namespace f1x
@@ -23,7 +41,7 @@ void TCPTransport::receive(size_t size, ReceivePromise::Pointer promise)
 
         if(receiveQueue_.size() == 1)
         {
-            this->doReceive();
+            this->distributeReceivedData();
         }
     });
 }
@@ -45,37 +63,21 @@ void TCPTransport::stop()
     tcpEndpoint_->stop();
 }
 
-void TCPTransport::doReceive()
+void TCPTransport::receiveHandler(size_t bytesTransferred)
 {
-    auto queueElement = receiveQueue_.begin();
+    try
-    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();
+        tcpReceivedDataSink_.commit(bytesTransferred);
+        this->distributeReceivedData();
+    }
+    catch(const error::Error& e)
+    {
+        //this->rejectReceivePromises(e);
     }
 }
 
-void TCPTransport::receiveFailureHandler(const aasdk::error::Error& e, ReceiveQueue::iterator queueElement)
+void TCPTransport::receiveFailureHandler(const aasdk::error::Error& e)
 {
-    queueElement->second->reject(e);
-    receiveQueue_.erase(queueElement);
-
-    if(!receiveQueue_.empty())
-    {
-        this->doReceive();
-    }
 }
 
 void TCPTransport::doSend()
@@ -83,12 +85,13 @@ 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),
+    sendPromise->then(std::bind(&TCPTransport::sendHandler, this->shared_from_this(), std::placeholders::_1, 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)
+void TCPTransport::sendHandler(size_t, SendQueue::iterator queueElement)
 {
     queueElement->second->resolve();
     sendQueue_.erase(queueElement);
@@ -110,6 +113,29 @@ void TCPTransport::sendFailureHandler(const aasdk::error::Error& e, SendQueue::i
     }
 }
 
+void TCPTransport::distributeReceivedData()
+{
+    for(auto queueElement = receiveQueue_.begin(); queueElement != receiveQueue_.end();)
+    {
+        if(tcpReceivedDataSink_.getAvailableSize() < queueElement->first)
+        {
+            auto buffer = tcpReceivedDataSink_.fill();
+
+            auto receivePromise = tcp::ITCPEndpoint::Promise::defer(receiveStrand_);
+            receivePromise->then(std::bind(&TCPTransport::receiveHandler, this->shared_from_this(), std::placeholders::_1),
+                                 std::bind(&TCPTransport::receiveFailureHandler, this->shared_from_this(), std::placeholders::_1));
+            tcpEndpoint_->receive(buffer, std::move(receivePromise));
+            break;
+        }
+        else
+        {
+            auto data(tcpReceivedDataSink_.consume(queueElement->first));
+            queueElement->second->resolve(std::move(data));
+            queueElement = receiveQueue_.erase(queueElement);
+        }
+    }
+}
+
 }
 }
 }

src/Transport/Transport.cpp

@@ -0,0 +1,96 @@
+/*
+*  This file is part of aasdk library project.
+*  Copyright (C) 2018 f1x.studio (Michal Szwaj)
+*
+*  aasdk 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.
+
+*  aasdk 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 aasdk. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <f1x/aasdk/Transport/Transport.hpp>
+
+namespace f1x
+{
+namespace aasdk
+{
+namespace transport
+{
+
+Transport::Transport(boost::asio::io_service& ioService)
+    : receiveStrand_(ioService)
+    , sendStrand_(ioService)
+{}
+
+void Transport::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)
+        {
+            try
+            {
+                this->distributeReceivedData();
+            }
+            catch(const error::Error& e)
+            {
+                this->rejectReceivePromises(e);
+            }
+        }
+    });
+}
+
+void Transport::distributeReceivedData()
+{
+    for(auto queueElement = receiveQueue_.begin(); queueElement != receiveQueue_.end();)
+    {
+        if(receivedDataSink_.getAvailableSize() < queueElement->first)
+        {
+            auto buffer = receivedDataSink_.fill();
+            this->enqueueReceive(std::move(buffer));
+
+            break;
+        }
+        else
+        {
+            auto data(receivedDataSink_.consume(queueElement->first));
+            queueElement->second->resolve(std::move(data));
+            queueElement = receiveQueue_.erase(queueElement);
+        }
+    }
+}
+
+void Transport::rejectReceivePromises(const error::Error& e)
+{
+    for(auto& queueElement : receiveQueue_)
+    {
+        queueElement.second->reject(e);
+    }
+
+    receiveQueue_.clear();
+}
+
+void Transport::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->enqueueSend(sendQueue_.begin());
+        }
+    });
+}
+
+}
+}
+}

src/Transport/USBTransport.cpp

@@ -26,35 +26,33 @@ namespace transport
 {
 
 USBTransport::USBTransport(boost::asio::io_service& ioService, usb::IAOAPDevice::Pointer aoapDevice)
-    : aoapDevice_(std::move(aoapDevice))
+    : Transport(ioService)
-    , receiveStrand_(ioService)
+    , aoapDevice_(std::move(aoapDevice))
-    , sendStrand_(ioService)
 {}
 
-void USBTransport::receive(size_t size, ReceivePromise::Pointer promise)
+void USBTransport::enqueueReceive(common::DataBuffer buffer)
 {
-    receiveStrand_.dispatch([this, self = this->shared_from_this(), size, promise = std::move(promise)]() mutable {
+    auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(receiveStrand_);
-        receiveQueue_.emplace_back(std::make_pair(size, std::move(promise)));
+    usbEndpointPromise->then([this, self = this->shared_from_this()](auto bytesTransferred) {
+            this->receiveHandler(bytesTransferred);
+        },
+        [this, self = this->shared_from_this()](auto e) {
+            this->rejectReceivePromises(e);
+        });
 
-        if(receiveQueue_.size() == 1)
+    aoapDevice_->getInEndpoint().bulkTransfer(buffer, cReceiveTimeoutMs, std::move(usbEndpointPromise));
-        {
+}
-            try
+
-            {
+void USBTransport::enqueueSend(SendQueue::iterator queueElement)
-                this->distributeReceivedData();
+{
-            }
+    this->doSend(queueElement, 0);
-            catch(const error::Error& e)
-            {
-                this->rejectReceivePromises(e);
-            }
-        }
-    });
 }
 
 void USBTransport::receiveHandler(size_t bytesTransferred)
 {
     try
     {
-        usbReceivedDataSink_.commit(bytesTransferred);
+        receivedDataSink_.commit(bytesTransferred);
         this->distributeReceivedData();
     }
     catch(const error::Error& e)
@@ -63,79 +61,35 @@ void USBTransport::receiveHandler(size_t bytesTransferred)
     }
 }
 
-void USBTransport::distributeReceivedData()
+void USBTransport::doSend(SendQueue::iterator queueElement, common::Data::size_type offset)
-{
-    for(auto queueElement = receiveQueue_.begin(); queueElement != receiveQueue_.end();)
-    {
-        if(usbReceivedDataSink_.getAvailableSize() < queueElement->first)
-        {
-            auto buffer = usbReceivedDataSink_.fill();
-
-            auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(receiveStrand_);
-            usbEndpointPromise->then(std::bind(&USBTransport::receiveHandler, this->shared_from_this(), std::placeholders::_1),
-                                    std::bind(&USBTransport::rejectReceivePromises, this->shared_from_this(), std::placeholders::_1));
-            aoapDevice_->getInEndpoint().bulkTransfer(buffer, cReceiveTimeoutMs, std::move(usbEndpointPromise));
-            break;
-        }
-        else
-        {
-            auto data(usbReceivedDataSink_.consume(queueElement->first));
-            queueElement->second->resolve(std::move(data));
-            queueElement = receiveQueue_.erase(queueElement);
-        }
-    }
-}
-
-void USBTransport::rejectReceivePromises(const error::Error& e)
-{
-    for(auto& queueElement : receiveQueue_)
-    {
-        queueElement.second->reject(e);
-    }
-
-    receiveQueue_.clear();
-}
-
-void USBTransport::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(sendQueue_.begin(), 0);
-        }
-    });
-}
-
-void USBTransport::doSend(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset)
 {
     auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(sendStrand_);
-    usbEndpointPromise->then([this, self = this->shared_from_this(), queueElementIter, offset](size_t bytesTransferred) mutable {
+    usbEndpointPromise->then([this, self = this->shared_from_this(), queueElement, offset](size_t bytesTransferred) mutable {
-            this->sendHandler(queueElementIter, offset, bytesTransferred);
+            this->sendHandler(queueElement, offset, bytesTransferred);
         },
-        [this, self = this->shared_from_this(), queueElementIter](const error::Error& e) mutable {
+        [this, self = this->shared_from_this(), queueElement](const error::Error& e) mutable {
-            queueElementIter->second->reject(e);
+            queueElement->second->reject(e);
-            sendQueue_.erase(queueElementIter);
+            sendQueue_.erase(queueElement);
 
             if(!sendQueue_.empty())
             {
                 this->doSend(sendQueue_.begin(), 0);
             }
         });
-    aoapDevice_->getOutEndpoint().bulkTransfer(common::DataBuffer(queueElementIter->first, offset), cSendTimeoutMs, std::move(usbEndpointPromise));
+
+    aoapDevice_->getOutEndpoint().bulkTransfer(common::DataBuffer(queueElement->first, offset), cSendTimeoutMs, std::move(usbEndpointPromise));
 }
 
-void USBTransport::sendHandler(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset, size_t bytesTransferred)
+void USBTransport::sendHandler(SendQueue::iterator queueElement, common::Data::size_type offset, size_t bytesTransferred)
 {
-    if(offset + bytesTransferred < queueElementIter->first.size())
+    if(offset + bytesTransferred < queueElement->first.size())
     {
-        this->doSend(queueElementIter, offset + bytesTransferred);
+        this->doSend(queueElement, offset + bytesTransferred);
     }
     else
     {
-        queueElementIter->second->resolve();
+        queueElement->second->resolve();
-        sendQueue_.erase(queueElementIter);
+        sendQueue_.erase(queueElement);
 
         if(!sendQueue_.empty())
         {