mirror of https://github.com/opencardev/aasdk
Use Transport base class in TCPTransport
parent
1c95f086b7
commit
175dfa6b60
|
@ -18,10 +18,8 @@
|
||||||
|
|
||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
#include <list>
|
|
||||||
#include <f1x/aasdk/TCP/ITCPEndpoint.hpp>
|
#include <f1x/aasdk/TCP/ITCPEndpoint.hpp>
|
||||||
#include <f1x/aasdk/Transport/DataSink.hpp>
|
#include <f1x/aasdk/Transport/Transport.hpp>
|
||||||
#include <f1x/aasdk/Transport/ITransport.hpp>
|
|
||||||
|
|
||||||
namespace f1x
|
namespace f1x
|
||||||
{
|
{
|
||||||
|
@ -30,37 +28,19 @@ namespace aasdk
|
||||||
namespace transport
|
namespace transport
|
||||||
{
|
{
|
||||||
|
|
||||||
class TCPTransport: public ITransport, public std::enable_shared_from_this<TCPTransport>, boost::noncopyable
|
class TCPTransport: public Transport
|
||||||
{
|
{
|
||||||
public:
|
public:
|
||||||
TCPTransport(boost::asio::io_service& ioService, tcp::ITCPEndpoint::Pointer tcpEndpoint);
|
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;
|
void stop() override;
|
||||||
|
|
||||||
private:
|
private:
|
||||||
typedef std::list<std::pair<size_t, ReceivePromise::Pointer>> ReceiveQueue;
|
void enqueueReceive(common::DataBuffer buffer) override;
|
||||||
typedef std::list<std::pair<common::Data, SendPromise::Pointer>> SendQueue;
|
void enqueueSend(SendQueue::iterator queueElement) override;
|
||||||
|
void sendHandler(SendQueue::iterator queueElement, const error::Error& e);
|
||||||
|
|
||||||
using std::enable_shared_from_this<TCPTransport>::shared_from_this;
|
|
||||||
void distributeReceivedData();
|
|
||||||
void receiveHandler(size_t bytesTransferred);
|
|
||||||
void receiveFailureHandler(const aasdk::error::Error& e);
|
|
||||||
|
|
||||||
void doSend();
|
|
||||||
void sendHandler(size_t bytesTransferred, SendQueue::iterator queueElement);
|
|
||||||
void sendFailureHandler(const aasdk::error::Error& e, SendQueue::iterator queueElement);
|
|
||||||
|
|
||||||
boost::asio::io_service& ioService_;
|
|
||||||
tcp::ITCPEndpoint::Pointer tcpEndpoint_;
|
tcp::ITCPEndpoint::Pointer tcpEndpoint_;
|
||||||
|
|
||||||
boost::asio::io_service::strand receiveStrand_;
|
|
||||||
ReceiveQueue receiveQueue_;
|
|
||||||
DataSink tcpReceivedDataSink_;
|
|
||||||
|
|
||||||
boost::asio::io_service::strand sendStrand_;
|
|
||||||
SendQueue sendQueue_;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -18,11 +18,8 @@
|
||||||
|
|
||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
#include <list>
|
|
||||||
#include <queue>
|
|
||||||
#include <boost/asio.hpp>
|
#include <boost/asio.hpp>
|
||||||
#include <f1x/aasdk/Transport/Transport.hpp>
|
#include <f1x/aasdk/Transport/Transport.hpp>
|
||||||
#include <f1x/aasdk/Transport/DataSink.hpp>
|
|
||||||
#include <f1x/aasdk/USB/IAOAPDevice.hpp>
|
#include <f1x/aasdk/USB/IAOAPDevice.hpp>
|
||||||
|
|
||||||
namespace f1x
|
namespace f1x
|
||||||
|
@ -42,7 +39,6 @@ public:
|
||||||
private:
|
private:
|
||||||
void enqueueReceive(common::DataBuffer buffer) override;
|
void enqueueReceive(common::DataBuffer buffer) override;
|
||||||
void enqueueSend(SendQueue::iterator queueElement) override;
|
void enqueueSend(SendQueue::iterator queueElement) override;
|
||||||
void receiveHandler(size_t bytesTransferred);
|
|
||||||
void doSend(SendQueue::iterator queueElement, common::Data::size_type offset);
|
void doSend(SendQueue::iterator queueElement, common::Data::size_type offset);
|
||||||
void sendHandler(SendQueue::iterator queueElement, common::Data::size_type offset, size_t bytesTransferred);
|
void sendHandler(SendQueue::iterator queueElement, common::Data::size_type offset, size_t bytesTransferred);
|
||||||
|
|
||||||
|
|
|
@ -26,36 +26,37 @@ namespace transport
|
||||||
{
|
{
|
||||||
|
|
||||||
TCPTransport::TCPTransport(boost::asio::io_service& ioService, tcp::ITCPEndpoint::Pointer tcpEndpoint)
|
TCPTransport::TCPTransport(boost::asio::io_service& ioService, tcp::ITCPEndpoint::Pointer tcpEndpoint)
|
||||||
: ioService_(ioService)
|
: Transport(ioService)
|
||||||
, tcpEndpoint_(std::move(tcpEndpoint))
|
, tcpEndpoint_(std::move(tcpEndpoint))
|
||||||
, receiveStrand_(ioService_)
|
|
||||||
, sendStrand_(ioService_)
|
|
||||||
{
|
{
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void TCPTransport::receive(size_t size, ReceivePromise::Pointer promise)
|
void TCPTransport::enqueueReceive(common::DataBuffer buffer)
|
||||||
{
|
{
|
||||||
receiveStrand_.dispatch([this, self = this->shared_from_this(), size, promise = std::move(promise)]() mutable {
|
auto receivePromise = tcp::ITCPEndpoint::Promise::defer(sendStrand_);
|
||||||
receiveQueue_.emplace_back(std::make_pair(size, std::move(promise)));
|
receivePromise->then([this, self = this->shared_from_this()](auto bytesTransferred) {
|
||||||
|
this->receiveHandler(bytesTransferred);
|
||||||
if(receiveQueue_.size() == 1)
|
},
|
||||||
{
|
[this, self = this->shared_from_this()](auto e) {
|
||||||
this->distributeReceivedData();
|
this->rejectReceivePromises(e);
|
||||||
}
|
|
||||||
});
|
});
|
||||||
|
|
||||||
|
tcpEndpoint_->receive(buffer, std::move(receivePromise));
|
||||||
}
|
}
|
||||||
|
|
||||||
void TCPTransport::send(common::Data data, SendPromise::Pointer promise)
|
void TCPTransport::enqueueSend(SendQueue::iterator queueElement)
|
||||||
{
|
{
|
||||||
sendStrand_.dispatch([this, self = this->shared_from_this(), data = std::move(data), promise = std::move(promise)]() mutable {
|
auto sendPromise = tcp::ITCPEndpoint::Promise::defer(sendStrand_);
|
||||||
sendQueue_.emplace_back(std::make_pair(std::move(data), std::move(promise)));
|
|
||||||
|
|
||||||
if(sendQueue_.size() == 1)
|
sendPromise->then([this, self = this->shared_from_this(), queueElement](auto) {
|
||||||
{
|
this->sendHandler(queueElement, error::Error());
|
||||||
this->doSend();
|
},
|
||||||
}
|
[this, self = this->shared_from_this(), queueElement](auto e) {
|
||||||
|
this->sendHandler(queueElement, e);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
tcpEndpoint_->send(common::DataConstBuffer(queueElement->first), std::move(sendPromise));
|
||||||
}
|
}
|
||||||
|
|
||||||
void TCPTransport::stop()
|
void TCPTransport::stop()
|
||||||
|
@ -63,76 +64,22 @@ void TCPTransport::stop()
|
||||||
tcpEndpoint_->stop();
|
tcpEndpoint_->stop();
|
||||||
}
|
}
|
||||||
|
|
||||||
void TCPTransport::receiveHandler(size_t bytesTransferred)
|
void TCPTransport::sendHandler(SendQueue::iterator queueElement, const error::Error& e)
|
||||||
{
|
{
|
||||||
try
|
if(!e)
|
||||||
{
|
|
||||||
tcpReceivedDataSink_.commit(bytesTransferred);
|
|
||||||
this->distributeReceivedData();
|
|
||||||
}
|
|
||||||
catch(const error::Error& e)
|
|
||||||
{
|
|
||||||
//this->rejectReceivePromises(e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void TCPTransport::receiveFailureHandler(const aasdk::error::Error& e)
|
|
||||||
{
|
|
||||||
}
|
|
||||||
|
|
||||||
void TCPTransport::doSend()
|
|
||||||
{
|
|
||||||
auto queueElement = sendQueue_.begin();
|
|
||||||
|
|
||||||
auto sendPromise = tcp::ITCPEndpoint::Promise::defer(sendStrand_);
|
|
||||||
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(size_t, SendQueue::iterator queueElement)
|
|
||||||
{
|
{
|
||||||
queueElement->second->resolve();
|
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();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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
|
else
|
||||||
{
|
{
|
||||||
auto data(tcpReceivedDataSink_.consume(queueElement->first));
|
queueElement->second->reject(e);
|
||||||
queueElement->second->resolve(std::move(data));
|
|
||||||
queueElement = receiveQueue_.erase(queueElement);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
sendQueue_.erase(queueElement);
|
||||||
|
|
||||||
|
if(!sendQueue_.empty())
|
||||||
|
{
|
||||||
|
this->enqueueSend(sendQueue_.begin());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -49,6 +49,19 @@ void Transport::receive(size_t size, ReceivePromise::Pointer promise)
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void Transport::receiveHandler(size_t bytesTransferred)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
receivedDataSink_.commit(bytesTransferred);
|
||||||
|
this->distributeReceivedData();
|
||||||
|
}
|
||||||
|
catch(const error::Error& e)
|
||||||
|
{
|
||||||
|
this->rejectReceivePromises(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void Transport::distributeReceivedData()
|
void Transport::distributeReceivedData()
|
||||||
{
|
{
|
||||||
for(auto queueElement = receiveQueue_.begin(); queueElement != receiveQueue_.end();)
|
for(auto queueElement = receiveQueue_.begin(); queueElement != receiveQueue_.end();)
|
||||||
|
|
|
@ -48,19 +48,6 @@ void USBTransport::enqueueSend(SendQueue::iterator queueElement)
|
||||||
this->doSend(queueElement, 0);
|
this->doSend(queueElement, 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
void USBTransport::receiveHandler(size_t bytesTransferred)
|
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
|
||||||
receivedDataSink_.commit(bytesTransferred);
|
|
||||||
this->distributeReceivedData();
|
|
||||||
}
|
|
||||||
catch(const error::Error& e)
|
|
||||||
{
|
|
||||||
this->rejectReceivePromises(e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void USBTransport::doSend(SendQueue::iterator queueElement, common::Data::size_type offset)
|
void USBTransport::doSend(SendQueue::iterator queueElement, common::Data::size_type offset)
|
||||||
{
|
{
|
||||||
auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(sendStrand_);
|
auto usbEndpointPromise = usb::IUSBEndpoint::Promise::defer(sendStrand_);
|
||||||
|
|
Loading…
Reference in New Issue