mirrors/gnss-sdr
1
0
Fork 0
mirror of https://github.com/gnss-sdr/gnss-sdr synced 2024-11-19 16:24:58 +00:00
Code Issues Releases Wiki Activity

Merge branch 'next' of https://github.com/gnss-sdr/gnss-sdr into next

Browse Source
This commit is contained in:
Carles Fernandez 2018-06-11 11:46:52 +02:00
commit 02dbca4f59
2 changed files with 39 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/core/system_parameters/rtcm.cc
View File

@ -53,7 +53,7 @@ Rtcm::Rtcm(unsigned short port)
reserved_field = std::bitset<6>("000000");
rtcm_message_queue = std::make_shared<concurrent_queue<std::string> >();
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), RTCM_port);
servers.emplace_back(io_service, endpoint);
servers.emplace_back(io_context, endpoint);
server_is_running = false;
}
@ -85,13 +85,13 @@ Rtcm::~Rtcm()
// *****************************************************************************************************
void Rtcm::run_server()
{
std::cout << "Starting a TCP Server on port " << RTCM_port << std::endl;
std::cout << "Starting a TCP/IP server of RTCM messages on port " << RTCM_port << std::endl;
try
{
std::thread tq([&] { std::make_shared<Queue_Reader>(io_service, rtcm_message_queue, RTCM_port)->do_read_queue(); });
std::thread tq([&] { std::make_shared<Queue_Reader>(io_context, rtcm_message_queue, RTCM_port)->do_read_queue(); });
tq.detach();
std::thread t([&] { io_service.run(); });
std::thread t([&] { io_context.run(); });
server_is_running = true;
t.detach();
}
@ -104,13 +104,13 @@ void Rtcm::run_server()
void Rtcm::stop_service()
{
io_service.stop();
io_context.stop();
}
void Rtcm::stop_server()
{
std::cout << "Stopping TCP Server on port " << RTCM_port << std::endl;
std::cout << "Stopping TCP/IP server on port " << RTCM_port << std::endl;
rtcm_message_queue->push("Goodbye"); // this terminates tq
Rtcm::stop_service();
servers.front().close_server();

51
src/core/system_parameters/rtcm.h
View File

@ -677,10 +677,10 @@ private:
{
if (first == true)
{
std::cout << "Client from " << socket_.remote_endpoint().address() << " says ";
LOG(INFO) << "Client says:";
first = false;
}
std::cout << client_says.substr(0, 80) << std::endl;
LOG(INFO) << client_says;
client_says = client_says.substr(80, client_says.length() - 80);
}
do_read_message_header();
@ -749,21 +749,21 @@ private:
: public std::enable_shared_from_this<Tcp_Internal_Client>
{
public:
Tcp_Internal_Client(boost::asio::io_service& io_service,
Tcp_Internal_Client(boost::asio::io_service& io_context,
boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
: io_service_(io_service), socket_(io_service)
: io_context_(io_context), socket_(io_context)
{
do_connect(endpoint_iterator);
}
inline void close()
{
io_service_.post([this]() { socket_.close(); });
io_context_.post([this]() { socket_.close(); });
}
inline void write(const Rtcm_Message& msg)
{
io_service_.post(
io_context_.post(
[this, msg]() {
bool write_in_progress = !write_msgs_.empty();
write_msgs_.push_back(msg);
@ -827,7 +827,7 @@ private:
});
}
boost::asio::io_service& io_service_;
boost::asio::io_service& io_context_;
boost::asio::ip::tcp::socket socket_;
Rtcm_Message read_msg_;
std::deque<Rtcm_Message> write_msgs_;
@ -837,13 +837,13 @@ private:
class Queue_Reader
{
public:
Queue_Reader(boost::asio::io_service& io_service, std::shared_ptr<concurrent_queue<std::string> >& queue, int port) : queue_(queue)
Queue_Reader(boost::asio::io_service& io_context, std::shared_ptr<concurrent_queue<std::string> >& queue, int port) : queue_(queue)
{
boost::asio::ip::tcp::resolver resolver(io_service);
boost::asio::ip::tcp::resolver resolver(io_context);
std::string host("localhost");
std::string port_str = std::to_string(port);
auto queue_endpoint_iterator = resolver.resolve({host.c_str(), port_str.c_str()});
c = std::make_shared<Tcp_Internal_Client>(io_service, queue_endpoint_iterator);
c = std::make_shared<Tcp_Internal_Client>(io_context, queue_endpoint_iterator);
}
inline void do_read_queue()
@ -871,8 +871,8 @@ private:
class Tcp_Server
{
public:
Tcp_Server(boost::asio::io_service& io_service, const boost::asio::ip::tcp::endpoint& endpoint)
: acceptor_(io_service), socket_(io_service)
Tcp_Server(boost::asio::io_service& io_context, const boost::asio::ip::tcp::endpoint& endpoint)
: acceptor_(io_context), socket_(io_context)
{
acceptor_.open(endpoint.protocol());
acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
@ -895,21 +895,35 @@ private:
{
if (first_client)
{
std::cout << "The TCP Server is up and running. Accepting connections ..." << std::endl;
std::cout << "The TCP/IP server of RTCM messages is up and running. Accepting connections ..." << std::endl;
first_client = false;
}
else
{
std::cout << "Starting RTCM TCP server session..." << std::endl;
std::cout << "Serving client from " << socket_.remote_endpoint().address() << std::endl;
LOG(INFO) << "Serving client from " << socket_.remote_endpoint().address();
std::cout << "Starting RTCM TCP/IP server session..." << std::endl;
boost::system::error_code ec2;
boost::asio::ip::tcp::endpoint endpoint = socket_.remote_endpoint(ec2);
if (ec2)
{
// Error creating remote_endpoint
std::cout << "Error getting remote IP address, closing session." << std::endl;
LOG(INFO) << "Error getting remote IP address";
start_session = false;
}
std::make_shared<Rtcm_Session>(std::move(socket_), room_)->start();
else
{
std::string remote_addr = endpoint.address().to_string();
std::cout << "Serving client from " << remote_addr << std::endl;
LOG(INFO) << "Serving client from " << remote_addr;
}
}
if (start_session) std::make_shared<Rtcm_Session>(std::move(socket_), room_)->start();
}
else
{
std::cout << "Error when invoking a RTCM session. " << ec << std::endl;
}
start_session = true;
do_accept();
});
}
@ -918,9 +932,10 @@ private:
boost::asio::ip::tcp::socket socket_;
Rtcm_Listener_Room room_;
bool first_client = true;
bool start_session = true;
};
boost::asio::io_service io_service;
boost::asio::io_service io_context;
std::shared_ptr<concurrent_queue<std::string> > rtcm_message_queue;
std::thread t;
std::thread tq;