ClientServer/TcpAcceptor.cpp at master · IEntin/ClientServer · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
/*
 *  Copyright (C) 2021 Ilya Entin
 */

#include <boost/stacktrace.hpp>

#include "TcpAcceptor.h"

#include "Connection.h"
#include "IOUtility.h"
#include "Options.h"
#include "Server.h"
#include "Tcp.h"

namespace tcp {

TcpAcceptor::TcpAcceptor(ServerWeakPtr server) :
  _server(server),
  _acceptor(_ioContext) {}

bool TcpAcceptor::start() {
  boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::tcp::v4(), Options::_tcpPort);
  boost::system::error_code ec;
  _acceptor.open(boost::asio::ip::tcp::v4(), ec);
  if (!ec)
    _acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true), ec);
  if (!ec)
    _acceptor.bind(endpoint, ec);
  if (!ec)
    _acceptor.listen(boost::asio::socket_base::max_listen_connections, ec);
  if (!ec) {
    boost::asio::post(_ioContext, [this] { accept(); });
  }
  if (ec) {
    LogError << ec.what() << " tcpPort=" << Options::_tcpPort << '\n';
    return false;
  }
  return true;
}

void TcpAcceptor::stop() {
  _stopped = true;
  boost::asio::post(_ioContext, [this] () {
    if (auto self = shared_from_this(); self) {
      _ioContext.stop();
    }
  });
}

void TcpAcceptor::run() {
  try {
    _ioContext.run();
  }
  catch (const std::exception& e) {
    LogError << boost::stacktrace::stacktrace() << '\n';
    LogError << e.what() << '\n';
  }
}

std::tuple<HEADERTYPE,
	   std::string,
	   std::string,
	   std::string,
	   std::string>
TcpAcceptor::connectionType(boost::asio::ip::tcp::socket& socket) {
  std::string primarySignatureWithKey;
  std::string primaryPubKeyAes;
  std::string secondarySignatureWithKey;
  std::string secondaryPubKeyAes;

  std::array<std::reference_wrapper<std::string>, 4> array{ std::ref(primarySignatureWithKey),
							    std::ref(primaryPubKeyAes),
							    std::ref(secondarySignatureWithKey),
							    std::ref(secondaryPubKeyAes) };
  if (!Tcp::readMessage(socket, _header, array))
    throw std::runtime_error(ioutility::createErrorString());
  assert(!isCompressed(_header) && "Expected uncompressed");
  return { extractHeaderType(_header), primarySignatureWithKey, primaryPubKeyAes,
	   secondarySignatureWithKey, secondaryPubKeyAes };
}

void TcpAcceptor::replyHeartbeat(boost::asio::ip::tcp::socket& socket) {
  Tcp::sendMessage(socket, _header);
  Logger logger(LOG_LEVEL::INFO, std::clog, false);
  logger << '*';
}

void TcpAcceptor::accept() {
  auto connection = std::make_shared<Connection>();
  _acceptor.async_accept(connection->_socket,
    [connection, this](const boost::system::error_code& ec) {
      if (_stopped)
	return;
     if (auto self = weak_from_this().lock(); !self)
	return;
      if (!ec) {
	auto [type, primarySignatureWithKey, primaryPubKeyAes,
	      secondarySignatureWithKey, secondaryPubKeyAes] = connectionType(connection->_socket);
	switch (type) {
	case HEADERTYPE::AUTHENTICATE:
	  if (auto server = _server.lock())
	    server->createTcpSession(connection,
				     primarySignatureWithKey,
				     primaryPubKeyAes,
				     secondarySignatureWithKey,
				     secondaryPubKeyAes);
	  break;
	case HEADERTYPE::HEARTBEAT:
	  replyHeartbeat(connection->_socket);
	  break;
	default:
	  break;
	}
	accept();
      }
      else {
	(ec == boost::asio::error::operation_aborted ? Debug : LogError) <<
	  ec.what() << '\n';
      }
    });
}

} // end of namespace tcp