diff --git a/include/f1x/aasdk/TCP/ITCPEndpoint.hpp b/include/f1x/aasdk/TCP/ITCPEndpoint.hpp index eb01c8b..f10d1c4 100644 --- a/include/f1x/aasdk/TCP/ITCPEndpoint.hpp +++ b/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 . +*/ + #pragma once #include @@ -15,7 +33,7 @@ class ITCPEndpoint { public: typedef std::shared_ptr Pointer; - typedef io::Promise Promise; + typedef io::Promise Promise; virtual ~ITCPEndpoint() = default; diff --git a/include/f1x/aasdk/TCP/ITCPWrapper.hpp b/include/f1x/aasdk/TCP/ITCPWrapper.hpp index 2abedff..8bb6532 100644 --- a/include/f1x/aasdk/TCP/ITCPWrapper.hpp +++ b/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 . +*/ + #pragma once #include diff --git a/include/f1x/aasdk/TCP/TCPEndpoint.hpp b/include/f1x/aasdk/TCP/TCPEndpoint.hpp index 199537f..64b3ef5 100644 --- a/include/f1x/aasdk/TCP/TCPEndpoint.hpp +++ b/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 . +*/ + #pragma once #include diff --git a/include/f1x/aasdk/TCP/TCPWrapper.hpp b/include/f1x/aasdk/TCP/TCPWrapper.hpp index db1e6f2..79ab483 100644 --- a/include/f1x/aasdk/TCP/TCPWrapper.hpp +++ b/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 . +*/ + #pragma once #include diff --git a/include/f1x/aasdk/Transport/TCPTransport.hpp b/include/f1x/aasdk/Transport/TCPTransport.hpp index 024b585..c4eeefd 100644 --- a/include/f1x/aasdk/Transport/TCPTransport.hpp +++ b/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 . +*/ + #pragma once #include #include +#include #include namespace f1x @@ -25,12 +44,12 @@ private: typedef std::list> SendQueue; using std::enable_shared_from_this::shared_from_this; - void doReceive(); - void receiveHandler(ReceiveQueue::iterator queueElement); - void receiveFailureHandler(const aasdk::error::Error& e, ReceiveQueue::iterator queueElement); + void distributeReceivedData(); + void receiveHandler(size_t bytesTransferred); + 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_; diff --git a/include/f1x/aasdk/Transport/Transport.hpp b/include/f1x/aasdk/Transport/Transport.hpp new file mode 100644 index 0000000..04ccc27 --- /dev/null +++ b/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 . +*/ + +#pragma once + +#include +#include +#include +#include +#include + +namespace f1x +{ +namespace aasdk +{ +namespace transport +{ + +class Transport: public ITransport, public std::enable_shared_from_this, 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> ReceiveQueue; + typedef std::list> SendQueue; + + using std::enable_shared_from_this::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_; +}; + +} +} +} diff --git a/include/f1x/aasdk/Transport/USBTransport.hpp b/include/f1x/aasdk/Transport/USBTransport.hpp index d2f3ba3..ffffb79 100644 --- a/include/f1x/aasdk/Transport/USBTransport.hpp +++ b/include/f1x/aasdk/Transport/USBTransport.hpp @@ -21,7 +21,7 @@ #include #include #include -#include +#include #include #include @@ -32,35 +32,21 @@ namespace aasdk namespace transport { -class USBTransport: public ITransport, public std::enable_shared_from_this, 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> InTransferQueue; - typedef std::list> OutTransferQueue; - - using std::enable_shared_from_this::shared_from_this; + void enqueueReceive(common::DataBuffer buffer) override; + void enqueueSend(SendQueue::iterator queueElement) override; void receiveHandler(size_t bytesTransferred); - void distributeReceivedData(); - void rejectReceivePromises(const error::Error& e); - - void doSend(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset); - void sendHandler(OutTransferQueue::iterator queueElementIter, common::Data::size_type offset, size_t bytesTransferred); + void doSend(SendQueue::iterator queueElement, common::Data::size_type offset); + void sendHandler(SendQueue::iterator queueElement, 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; diff --git a/src/TCP/TCPEndpoint.cpp b/src/TCP/TCPEndpoint.cpp index eaca218..168eb72 100644 --- a/src/TCP/TCPEndpoint.cpp +++ b/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 . +*/ + #include 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 { diff --git a/src/TCP/TCPWrapper.cpp b/src/TCP/TCPWrapper.cpp index 7b722a0..9f016d0 100644 --- a/src/TCP/TCPWrapper.cpp +++ b/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 . +*/ + #include #include @@ -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) diff --git a/src/Transport/TCPTransport.cpp b/src/Transport/TCPTransport.cpp index 72fcab4..21f0466 100644 --- a/src/Transport/TCPTransport.cpp +++ b/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 . +*/ + #include 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(); - 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()) + try { - 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); + } + } +} + } } } diff --git a/src/Transport/Transport.cpp b/src/Transport/Transport.cpp new file mode 100644 index 0000000..60c4935 --- /dev/null +++ b/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 . +*/ + +#include + +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()); + } + }); +} + +} +} +} diff --git a/src/Transport/USBTransport.cpp b/src/Transport/USBTransport.cpp index 89505d5..3eec8df 100644 --- a/src/Transport/USBTransport.cpp +++ b/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)) - , receiveStrand_(ioService) - , sendStrand_(ioService) + : Transport(ioService) + , aoapDevice_(std::move(aoapDevice)) {} -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 { - receiveQueue_.emplace_back(std::make_pair(size, std::move(promise))); + auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(receiveStrand_); + 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) - { - try - { - this->distributeReceivedData(); - } - catch(const error::Error& e) - { - this->rejectReceivePromises(e); - } - } - }); + aoapDevice_->getInEndpoint().bulkTransfer(buffer, cReceiveTimeoutMs, std::move(usbEndpointPromise)); +} + +void USBTransport::enqueueSend(SendQueue::iterator queueElement) +{ + this->doSend(queueElement, 0); } 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() -{ - 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) +void USBTransport::doSend(SendQueue::iterator queueElement, 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 { - this->sendHandler(queueElementIter, offset, bytesTransferred); + usbEndpointPromise->then([this, self = this->shared_from_this(), queueElement, offset](size_t bytesTransferred) mutable { + this->sendHandler(queueElement, offset, bytesTransferred); }, - [this, self = this->shared_from_this(), queueElementIter](const error::Error& e) mutable { - queueElementIter->second->reject(e); - sendQueue_.erase(queueElementIter); + [this, self = this->shared_from_this(), queueElement](const error::Error& e) mutable { + queueElement->second->reject(e); + 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(); - sendQueue_.erase(queueElementIter); + queueElement->second->resolve(); + sendQueue_.erase(queueElement); if(!sendQueue_.empty()) {