итоговый сервер

lesson_04/client/tcp_client.cpp

@@ -0,0 +1,16 @@
+/*
+ * tcp_client.cpp
+ *
+ *  Created on: 21 сент. 2022 г.
+ *      Author: alexander
+ */
+
+#include <client/tcp_client.hpp>
+
+namespace azh
+{
+    TCPClient::TCPClient()
+    {
+
+    }
+}

lesson_04/client/tcp_client.hpp

@@ -0,0 +1,19 @@
+/*
+ * tcp_client.hpp
+ *
+ *  Created on: 21 сент. 2022 г.
+ *      Author: alexander
+ */
+
+#pragma once
+
+namespace azh
+{
+
+    class TCPClient
+    {
+    public:
+        TCPClient();
+    };
+
+}

lesson_04/connection/connection.cpp

@@ -1,50 +0,0 @@
-/*
- * connection.cpp
- *
- *  Created on: 20 сент. 2022 г.
- *      Author: alexander
- */
-
-#include <connection/connection.hpp>
-#include <iostream>
-
-namespace azh
-{
-    Connection::Connection(io::ip::tcp::socket&& socket) : _socket(std::move(socket))
-    {
-    }
-
-    tcp::socket& Connection::socket()
-    {
-        return _socket;
-    }
-
-    void Connection::start()
-    {
-        auto strongThis = shared_from_this();
-
-        boost::asio::async_write(_socket, boost::asio::buffer(_message),
-                [strongThis](const boost::system::error_code &error, size_t bytesTransferred)
-                {
-                    if (error)
-                        std::cerr << "Failed to send message!" << std::endl;
-                    else
-                        std::cout << "Sent " << bytesTransferred << " bytes" << std::endl;
-                });
-
-        boost::asio::streambuf buffer;
-
-        _socket.async_receive(buffer.prepare(512), [this](const boost::system::error_code &error, size_t bytesTransferred)
-        {
-            if (error == boost::asio::error::eof)
-                std::cout << "Client disconnected properly!" << std::endl;
-            else if (error)
-                std::cerr << "Client disconnected in bad way!" << std::endl;
-        });
-    }
-
-    Connection::pointer Connection::create(io::ip::tcp::socket&& socket)
-    {
-        return pointer(new Connection(std::move(socket)));
-    }
-}

lesson_04/connection/connection.hpp

@@ -1,33 +0,0 @@
-/*
- * connection.hpp
- *
- *  Created on: 20 сент. 2022 г.
- *      Author: alexander
- */
-
-#pragma once
-
-#include <boost/asio.hpp>
-#include <memory>
-
-namespace azh
-{
-    using boost::asio::ip::tcp;
-    namespace io = boost::asio;
-
-    class Connection : public std::enable_shared_from_this<Connection>
-    {
-    public:
-        using pointer = std::shared_ptr<Connection>;
-
-        static pointer create(io::ip::tcp::socket&& socket);
-        tcp::socket& socket();
-        void start();
-    private:
-        tcp::socket _socket;
-        std::string _message = "Hello, client!\n";
-    private:
-        explicit Connection(io::ip::tcp::socket&& socket);
-    };
-
-}

lesson_04/connection/tcp_connection.cpp

@@ -0,0 +1,103 @@
+/*
+ * tcp_connection.cpp
+ *
+ *  Created on: 20 сент. 2022 г.
+ *      Author: alexander
+ */
+
+#include <connection/tcp_connection.hpp>
+#include <iostream>
+
+namespace azh
+{
+    TCPConnection::TCPConnection(io::ip::tcp::socket&& socket) : _socket(std::move(socket))
+    {
+        boost::system::error_code ec;
+
+        std::stringstream name;
+        name << _socket.remote_endpoint();
+
+        _username = name.str();
+    }
+
+    tcp::socket& TCPConnection::socket()
+    {
+        return _socket;
+    }
+
+    void TCPConnection::start(MessageHandler&& messageHandler, ErrorHandler&& errorHandler)
+    {
+        _messageHandler = std::move(messageHandler);
+        _errorHandler = std::move(errorHandler);
+
+        asyncRead();
+    }
+
+    void TCPConnection::post(const std::string& message)
+    {
+        bool queueIdle = _outgoingMessages.empty();
+        _outgoingMessages.push(message);
+
+        if (queueIdle)
+            asyncWrite();
+    }
+
+    void TCPConnection::asyncRead()
+    {
+        io::async_read_until(_socket, _streamBuf, '\n', [self = shared_from_this()]
+            (boost::system::error_code ec, size_t bytesTransferred) {
+                self->onRead(ec, bytesTransferred);
+        });
+    }
+
+    void TCPConnection::onRead(boost::system::error_code &ec, size_t bytesTransferred)
+    {
+        if (ec)
+        {
+            _socket.close(ec);
+
+            _errorHandler();
+
+            return;
+        }
+
+        std::stringstream message;
+        message << _username << ": " << std::istream(&_streamBuf).rdbuf();
+
+        _streamBuf.consume(bytesTransferred);
+
+        _messageHandler(message.str());
+
+        asyncRead();
+    }
+
+    void TCPConnection::asyncWrite()
+    {
+        io::async_write(_socket, io::buffer(_outgoingMessages.front()), [self = shared_from_this()]
+            (boost::system::error_code ec, size_t bytesTransferred) {
+                self->onWrite(ec, bytesTransferred);
+        });
+    }
+
+    void TCPConnection::onWrite(boost::system::error_code &ec, size_t bytesTransferred)
+    {
+        if (ec)
+        {
+            _socket.close(ec);
+
+            _errorHandler();
+
+            return;
+        }
+
+        _outgoingMessages.pop();
+
+        if (!_outgoingMessages.empty())
+            asyncWrite();
+    }
+
+    TCPConnection::Pointer TCPConnection::create(io::ip::tcp::socket&& socket)
+    {
+        return Pointer(new TCPConnection(std::move(socket)));
+    }
+}

lesson_04/connection/tcp_connection.hpp

@@ -0,0 +1,57 @@
+/*
+ * tcp_connection.hpp
+ *
+ *  Created on: 20 сент. 2022 г.
+ *      Author: alexander
+ */
+
+#pragma once
+
+#include <boost/asio.hpp>
+#include <memory>
+#include <queue>
+
+namespace azh
+{
+    using boost::asio::ip::tcp;
+    namespace io = boost::asio;
+
+    using MessageHandler = std::function<void(std::string)>;
+    using ErrorHandler = std::function<void()>;
+
+    class TCPConnection : public std::enable_shared_from_this<TCPConnection>
+    {
+    public:
+        using Pointer = std::shared_ptr<TCPConnection>;
+
+        static Pointer create(io::ip::tcp::socket&& socket);
+
+        inline const std::string& getUsername() const
+        {
+            return _username;
+        }
+
+        tcp::socket& socket();
+        void start(MessageHandler&& messageHandler, ErrorHandler&& errorHandler);
+
+        void post(const std::string& message);
+    private:
+        tcp::socket _socket;
+        std::string _username;
+
+        std::queue<std::string> _outgoingMessages;
+        io::streambuf _streamBuf {65536};
+
+        MessageHandler _messageHandler;
+        ErrorHandler _errorHandler;
+    private:
+        explicit TCPConnection(io::ip::tcp::socket&& socket);
+
+        void asyncRead();
+        void onRead(boost::system::error_code &ec, size_t bytesTransferred);
+
+        void asyncWrite();
+        void onWrite(boost::system::error_code &ec, size_t bytesTransferred);
+    };
+
+}

lesson_04/main.cpp

@@ -1,5 +1,6 @@
 #include <ap/ap.hpp>
 #include <server/tcp_server.hpp>
+#include <connection/tcp_connection.hpp>
 #include <iostream>
 
 void server(int port)
@@ -7,6 +8,22 @@ void server(int port)
     using namespace azh;
 
     TCPServer server(IPV::V4, port);
+
+    server.onJoin = [](TCPConnection::Pointer server)
+    {
+        std::cout << "User has joined the server " << server->getUsername() << std::endl;
+    };
+
+    server.onLeave = [](TCPConnection::Pointer server)
+    {
+        std::cout << "User has left the server " << server->getUsername() << std::endl;
+    };
+
+    server.onClientMessage = [&server](const std::string& message)
+    {
+        server.broadcast(message);
+    };
+
     server.run();
 }
 

lesson_04/server/tcp_server.cpp

@@ -1,12 +1,12 @@
 /*
- * server.cpp
+ * tcp_server.cpp
  *
  *  Created on: 20 сент. 2022 г.
  *      Author: alexander
  */
 
+#include <connection/tcp_connection.hpp>
 #include <server/tcp_server.hpp>
-#include <connection/connection.hpp>
 #include <iostream>
 
 namespace azh
@@ -33,9 +33,12 @@ namespace azh
         return 0;
     }
 
-    void TCPServer::broadcast(const std::string& message)
+    void TCPServer::broadcast(const std::string &message)
     {
-
+        for (auto &connection : _connections)
+        {
+            connection->post(message);
+        }
     }
 
     void TCPServer::startAccept()
@@ -44,12 +47,25 @@ namespace azh
 
         _acceptor.async_accept(*_socket, [this](const boost::system::error_code &error)
         {
-            auto connection = Connection::create(std::move(*_socket));
+            auto connection = TCPConnection::create(std::move(*_socket));
+
+            if (onJoin)
+            onJoin(connection);
 
             _connections.insert(connection);
 
             if (!error)
-                connection->start();
+                connection->start(
+                    [this](const std::string& message)
+                    {
+                        if (onClientMessage) onClientMessage(message);
+                    },
+                    [&, weak = std::weak_ptr(connection)]
+                    {
+                        if (auto shared = weak.lock(); shared && _connections.erase(shared))
+                        if (onLeave) onLeave(shared);
+                    }
+                );
 
             startAccept();
         });

lesson_04/server/tcp_server.hpp

@@ -1,5 +1,5 @@
 /*
- * server.hpp
+ * tcp_server.hpp
  *
  *  Created on: 20 сент. 2022 г.
  *      Author: alexander
@@ -8,7 +8,7 @@
 #pragma once
 
 #include <boost/asio.hpp>
-#include <connection/connection.hpp>
+#include <connection/tcp_connection.hpp>
 #include <vector>
 #include <optional>
 #include <unordered_set>
@@ -24,12 +24,20 @@ namespace azh
 
     class TCPServer
     {
+        using OnJoinHandler = std::function<void(TCPConnection::Pointer)>;
+        using OnLeaveHandler = std::function<void(TCPConnection::Pointer)>;
+        using OnClientMessageHandler = std::function<void(std::string)>;
+
     public:
         TCPServer(IPV ipv, int port);
 
         int run();
 
         void broadcast(const std::string& message);
+    public:
+        OnJoinHandler onJoin;
+        OnLeaveHandler onLeave;
+        OnClientMessageHandler onClientMessage;
     private:
         IPV _ipVersion;
         int _port;
@@ -37,7 +45,7 @@ namespace azh
         io::io_context _ioContext;
         io::ip::tcp::acceptor _acceptor;
         std::optional<io::ip::tcp::socket> _socket;
-        std::unordered_set<Connection::pointer> _connections;
+        std::unordered_set<TCPConnection::Pointer> _connections;
     private:
         void startAccept();
     };