21101426787

Committed 17 Jan 2026 09:52PM UTC coverage: 28.005% (+0.01%) from 27.995%

Build # 21101426787

Build Type

push

github

Committed by

reinder

Commit Message

[traintastic] fixed restart crash (out of memory)

Run Details

12 of 38 new or added lines in 11 files covered. (31.58%)

5 existing lines in 2 files now uncovered.

8165 of 29156 relevant lines covered (28.0%)

193.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/server/src/network/clientconnection.cpp

/**
 * This file is part of Traintastic,
 * see <https://github.com/traintastic/traintastic>.
 *
 * Copyright (C) 2019-2026 Reinder Feenstra
 *
 * This program 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 2
 * of the License, or (at your option) any later version.
 *
 * This program 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 this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */

#include "clientconnection.hpp"
#include "server.hpp"
#include "../traintastic/traintastic.hpp"
#include "../core/eventloop.hpp"
#include "session.hpp"
#include "../log/log.hpp"

ClientConnection::ClientConnection(Server& server, std::shared_ptr<boost::beast::websocket::stream<boost::beast::tcp_stream>> ws)
  : WebSocketConnection(server, std::move(ws), "client")
  , m_authenticated{false}
{
  assert(isServerThread());
  m_ws->binary(true);
}

ClientConnection::~ClientConnection()
{
  assert(isEventLoopThread());
  assert(!m_session);
}

void ClientConnection::doRead()
{
  assert(isServerThread());

  m_ws->async_read(m_readBuffer,
    [this, weak=weak_from_this()](const boost::system::error_code& ec, std::size_t /*bytesReceived*/)
    {
      if(weak.expired())
        return;

      if(!ec)
      {
        while(m_readBuffer.size() >= sizeof(Message::Header))
        {
          const Message::Header& header = *reinterpret_cast<const Message::Header*>(m_readBuffer.cdata().data());
          if(m_readBuffer.size() >= sizeof(Message::Header) + header.dataSize)
          {
            auto message = std::make_shared<Message>(header);
            if(header.dataSize != 0)
            {
              std::memcpy(message->data(), static_cast<const std::byte*>(m_readBuffer.cdata().data()) + sizeof(header), message->dataSize());
            }
            EventLoop::call(&ClientConnection::processMessage, this, message);
            m_readBuffer.consume(sizeof(header) + header.dataSize);
          }
          else
          {
            break;
          }
        }
        doRead();
      }
      else if(ec == boost::asio::error::eof || ec == boost::asio::error::connection_aborted || ec == boost::asio::error::connection_reset)
      {
        EventLoop::call(std::bind(&ClientConnection::connectionLost, this));
      }
      else if(ec != boost::asio::error::operation_aborted)
      {
        Log::log(id, LogMessage::E1007_SOCKET_READ_FAILED_X, ec);
        EventLoop::call(std::bind(&ClientConnection::disconnect, this));
      }
    });
}

void ClientConnection::doWrite()
{
  assert(isServerThread());

  m_ws->async_write(boost::asio::buffer(**m_writeQueue.front(), m_writeQueue.front()->size()),
    [this, weak=weak_from_this()](const boost::system::error_code& ec, std::size_t /*bytesTransferred*/)
    {
      if(weak.expired())
        return;

      if(!ec)
      {
        m_writeQueue.pop();
        if(!m_writeQueue.empty())
          doWrite();
      }
      else if(ec != boost::asio::error::operation_aborted)
      {
        Log::log(id, LogMessage::E1006_SOCKET_WRITE_FAILED_X, ec);
        EventLoop::call(std::bind(&ClientConnection::disconnect, this));
      }
    });
}

void ClientConnection::processMessage(const std::shared_ptr<Message> message)
{
  assert(isEventLoopThread());

  if(m_authenticated && m_session)
  {
    if(m_session->processMessage(*message))
      return;
  }
  else if(m_authenticated && !m_session)
  {
    if(message->command() == Message::Command::NewSession && message->type() == Message::Type::Request)
    {
      m_session = std::make_shared<Session>(std::dynamic_pointer_cast<ClientConnection>(shared_from_this()));
      auto response = Message::newResponse(message->command(), message->requestId());
      response->write(m_session->uuid());
      m_session->writeObject(*response, Traintastic::instance);
      sendMessage(std::move(response));
      return;
    }
  }
  else
  {
    if(message->command() == Message::Command::Login && message->type() == Message::Type::Request)
    {
      m_authenticated = true; // oke for now, login can be added later :)
      sendMessage(Message::newResponse(message->command(), message->requestId()));
      return;
    }
  }

  if(message->type() == Message::Type::Request)
  {
    //assert(false);
    sendMessage(Message::newErrorResponse(message->command(), message->requestId(), LogMessage::C1014_INVALID_COMMAND));
  }
}

void ClientConnection::sendMessage(std::unique_ptr<Message> message)
{
  assert(isEventLoopThread());

  ioContext().post(
    [this, msg=std::make_shared<std::unique_ptr<Message>>(std::move(message))]()
    {
      const bool wasEmpty = m_writeQueue.empty();
      m_writeQueue.emplace(std::move(*msg));
      if(wasEmpty)
        doWrite();
    });
}

void ClientConnection::disconnect()
{
  assert(isEventLoopThread());

  m_session.reset();

  WebSocketConnection::disconnect();
}

1	/**
2	* This file is part of Traintastic,
3	* see <https://github.com/traintastic/traintastic>.
4	*
5	* Copyright (C) 2019-2026 Reinder Feenstra
6	*
7	* This program is free software; you can redistribute it and/or
8	* modify it under the terms of the GNU General Public License
9	* as published by the Free Software Foundation; either version 2
10	* of the License, or (at your option) any later version.
11	*
12	* This program is distributed in the hope that it will be useful,
13	* but WITHOUT ANY WARRANTY; without even the implied warranty of
14	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	* GNU General Public License for more details.
16	*
17	* You should have received a copy of the GNU General Public License
18	* along with this program; if not, write to the Free Software
19	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
20	*/
21
22	#include "clientconnection.hpp"
23	#include "server.hpp"
24	#include "../traintastic/traintastic.hpp"
25	#include "../core/eventloop.hpp"
26	#include "session.hpp"
27	#include "../log/log.hpp"
28
29	ClientConnection::ClientConnection(Server& server, std::shared_ptr<boost::beast::websocket::stream<boost::beast::tcp_stream>> ws)	×
30	: WebSocketConnection(server, std::move(ws), "client")	×
31	, m_authenticated{false}	×
32	{
33	assert(isServerThread());	×
34	m_ws->binary(true);	×
35	}	×
36
37	ClientConnection::~ClientConnection()	×
38	{
39	assert(isEventLoopThread());	×
40	assert(!m_session);	×
41	}	×
42
43	void ClientConnection::doRead()	×
44	{
45	assert(isServerThread());	×
46
47	m_ws->async_read(m_readBuffer,	×
48	[this, weak=weak_from_this()](const boost::system::error_code& ec, std::size_t /bytesReceived/)	×
49	{
50	if(weak.expired())	×
51	return;	×
52
53	if(!ec)	×
54	{
55	while(m_readBuffer.size() >= sizeof(Message::Header))	×
56	{
57	const Message::Header& header = reinterpret_cast<const Message::Header>(m_readBuffer.cdata().data());	×
58	if(m_readBuffer.size() >= sizeof(Message::Header) + header.dataSize)	×
59	{
60	auto message = std::make_shared<Message>(header);	×
61	if(header.dataSize != 0)	×
62	{
63	std::memcpy(message->data(), static_cast<const std::byte*>(m_readBuffer.cdata().data()) + sizeof(header), message->dataSize());	×
64	}
65	EventLoop::call(&ClientConnection::processMessage, this, message);	×
66	m_readBuffer.consume(sizeof(header) + header.dataSize);	×
67	}	×
68	else
69	{
70	break;	×
71	}
72	}
73	doRead();	×
74	}
75	else if(ec == boost::asio::error::eof \|\| ec == boost::asio::error::connection_aborted \|\| ec == boost::asio::error::connection_reset)	×
76	{
77	EventLoop::call(std::bind(&ClientConnection::connectionLost, this));	×
78	}
NEW 79	else if(ec != boost::asio::error::operation_aborted)	×
80	{
81	Log::log(id, LogMessage::E1007_SOCKET_READ_FAILED_X, ec);	×
82	EventLoop::call(std::bind(&ClientConnection::disconnect, this));	×
83	}
84	});
85	}	×
86
87	void ClientConnection::doWrite()	×
88	{
89	assert(isServerThread());	×
90
91	m_ws->async_write(boost::asio::buffer(**m_writeQueue.front(), m_writeQueue.front()->size()),	×
92	[this, weak=weak_from_this()](const boost::system::error_code& ec, std::size_t /bytesTransferred/)	×
93	{
94	if(weak.expired())	×
95	return;	×
96
97	if(!ec)	×
98	{
99	m_writeQueue.pop();	×
100	if(!m_writeQueue.empty())	×
101	doWrite();	×
102	}
103	else if(ec != boost::asio::error::operation_aborted)	×
104	{
105	Log::log(id, LogMessage::E1006_SOCKET_WRITE_FAILED_X, ec);	×
106	EventLoop::call(std::bind(&ClientConnection::disconnect, this));	×
107	}
108	});
109	}	×
110
111	void ClientConnection::processMessage(const std::shared_ptr<Message> message)	×
112	{
113	assert(isEventLoopThread());	×
114
115	if(m_authenticated && m_session)	×
116	{
117	if(m_session->processMessage(*message))	×
118	return;	×
119	}
120	else if(m_authenticated && !m_session)	×
121	{
122	if(message->command() == Message::Command::NewSession && message->type() == Message::Type::Request)	×
123	{
124	m_session = std::make_shared<Session>(std::dynamic_pointer_cast<ClientConnection>(shared_from_this()));	×
125	auto response = Message::newResponse(message->command(), message->requestId());	×
126	response->write(m_session->uuid());	×
127	m_session->writeObject(*response, Traintastic::instance);	×
128	sendMessage(std::move(response));	×
129	return;	×
130	}	×
131	}
132	else
133	{
134	if(message->command() == Message::Command::Login && message->type() == Message::Type::Request)	×
135	{
136	m_authenticated = true; // oke for now, login can be added later :)	×
137	sendMessage(Message::newResponse(message->command(), message->requestId()));	×
138	return;	×
139	}
140	}
141
142	if(message->type() == Message::Type::Request)	×
143	{
144	//assert(false);
145	sendMessage(Message::newErrorResponse(message->command(), message->requestId(), LogMessage::C1014_INVALID_COMMAND));	×
146	}
147	}
148
149	void ClientConnection::sendMessage(std::unique_ptr<Message> message)	×
150	{
151	assert(isEventLoopThread());	×
152
153	ioContext().post(	×
154	[this, msg=std::make_shared<std::unique_ptr<Message>>(std::move(message))]()	×
155	{
156	const bool wasEmpty = m_writeQueue.empty();	×
157	m_writeQueue.emplace(std::move(*msg));	×
158	if(wasEmpty)	×
159	doWrite();	×
160	});	×
161	}	×
162
163	void ClientConnection::disconnect()	×
164	{
165	assert(isEventLoopThread());	×
166
167	m_session.reset();	×
168
169	WebSocketConnection::disconnect();	×
170	}	×

traintastic / traintastic / 21101426787

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous