Boost Asio async_read and async_receive not getting called itself whenever there is a data on tcp socket But boost::asio::read got call automatically

Question

I am making a bidirectional asynchronous (non blocking) TCP client/server socket communication in Linux where a server can handle multiple clients. I am using boost ASIO TCP library framework for my requirements.

In my project there are two processes: client and server.

I want to transfer a custom structure (which has some enums, strings and other kind of data) over the TCP. For that I use Boost Serialization.

In both server and client I have implemented startReading() and startWriting() function in async mode using async_read and async_write.

To test, I

start server process, like
```
 ./test_server 9876
```
whenever a client process start from command line like
```
 ./test_client 127.0.0.1 9876
```
Instantly, it will connect we see the server notification that a client is connected.
Now client sends a hello message to the server. The server immediately replies with serialized struct.

The client is not getting the reply, whether I use async_read or async_receive.

If I am using a blocking asio::read in the client startReading(), the client gets the message.

But I want it async.

I am sharing the completed project code below, please check and give me some suggestions. I am using ubuntu 20.04 64 bit os, C++14, Boost 1.71.

Server Logs

Client Read Logs (which is working)

Client async_read logs which is not working

test_server.cpp 'its server process main function'

//test_server.cpp
#include 
#include "tcpServer.h"

int main(int argc, char *argv[])
{
  try
  {
    if (argc != 2)
    {
      std::cerr << "Missing port no: Please pass port no argument in async_tcp_server
";
      return 1;
    }
    boost::asio::io_context io_context;
    TcpServer server(io_context,std::atoi(argv[1]));
    io_context.run();
  }
  catch (std::exception &e)
  {
    std::cerr <<"Exception: " << e.what() << std::endl;
  }
  return 0;
}

This is tcpServer.h


//tcpServer.h
#ifndef TCP_SERVER_H
#define TCP_SERVER_H

#include 
#include 
#include 
#include 
#include 
#include "AppPlugin.h"


/**
*   @class  TcpConnectionHandler tcpServer.h "server/tcpServer.h"
*   @brief  
*/
class TcpConnectionHandler : public boost::enable_shared_from_this
{
private:
    tcp::socket serverSocket_;
    TcpConnectionHandler(boost::asio::io_context&);
    TestManagerComInfo messageInfo;

public:
    enum { max_length = 4096};
    char data[max_length];
    string message="ok client, I am sending the strucure!";
    
    typedef boost::shared_ptr pointer;
    // creating the pointer
    static inline pointer create(boost::asio::io_context& io_context)
    {
        return pointer(new TcpConnectionHandler(io_context));
    }
    //socket creation
    tcp::socket& socket()
    {
        return serverSocket_;
    }

    void StartReading();
    void StartWriting();
    void WriteDone(const boost::system::error_code&, size_t);
    void ReadDone(const boost::system::error_code&, size_t);
    void setMessageStruct(const TestManagerComInfo&);
    TestManagerComInfo& getMessageStruct()
    {
        return messageInfo;
    }
    void SockDiscNotifierServer();
};

/**
*   @class  TcpServer tcpServer.h "server/tcpServer.h"
*   @brief  
*/
class TcpServer 
{
private:
    tcp::acceptor acceptor_;
    boost::asio::io_context& io_context_local;
    boost::asio::io_context& getioContext()
    {
        return io_context_local;
    }
   void start_accept();

public:
    TcpServer(boost::asio::io_context&,short);  
};

#endif  // TCP_SERVER_H

This is tcpserver.cpp

//tcpServer.cpp
#include "tcpServer.h"


TcpConnectionHandler::TcpConnectionHandler(boost::asio::io_context &io_context) : serverSocket_(io_context)
{
messageInfo.requestType_ = kTerminate;
    messageInfo.name_ = "COM_IPC_GROUP_001";
    messageInfo.testLogType_ = {
        {"Test1", kTestPass},
        {"Test112", kTestFail},
        {"Test3", kTestAbort}
    };
    messageInfo.statusType_ = kCompleted;
}

void TcpConnectionHandler::setMessageStruct(const TestManagerComInfo &passedMessageStruct)
{
#if 0
    messageInfo.requestType_ = kTerminate;
    messageInfo.name_ = "COM_IPC_GROUP_001";
    messageInfo.testLogType_ = {
        {"Test1", kTestPass},
        {"Test2", kTestFail},
        {"Test3", kTestAbort}
    };
    messageInfo.statusType_ = kCompleted;
#endif

    messageInfo.requestType_ = passedMessageStruct.requestType_;
    messageInfo.name_ = passedMessageStruct.name_;
    for (const auto &entry : passedMessageStruct.testLogType_)
    {
        messageInfo.testLogType_[entry.first] = entry.second;
    }
    messageInfo.statusType_ = passedMessageStruct.statusType_;
}

void TcpConnectionHandler::StartReading()
{
    serverSocket_.async_read_some(
        boost::asio::buffer(data, max_length),
        boost::bind(&TcpConnectionHandler::ReadDone,
                    shared_from_this(),
                    boost::asio::placeholders::error,
                    boost::asio::placeholders::bytes_transferred));    
}

void TcpConnectionHandler::StartWriting()
{
    string serializedData = serializeMessageStruct(getMessageStruct());
        serverSocket_.async_write_some(
        boost::asio::buffer(serializedData, max_length),
        boost::bind(&TcpConnectionHandler::WriteDone,
                    shared_from_this(),
                    boost::asio::placeholders::error,
                    boost::asio::placeholders::bytes_transferred));
}
void TcpConnectionHandler::ReadDone(const boost::system::error_code &err, size_t bytes_transferred)
{
    if (!err)
    {
        cout << "server received data is '" << data << "'
";
        StartWriting();
    }
    else
    {
        SockDiscNotifierServer();
        cerr << "Read error: " << err.message() << endl;
    }
}

void TcpConnectionHandler::WriteDone(const boost::system::error_code &err, size_t bytes_transferred)
{
    if (!err)
    {
        cout << "Server sent message!" << endl;
        serverSocket_.is_open()?cout<<"Server socket is still connected
":cout<<"server socket got disconnected
";
    }
    else
    {
        SockDiscNotifierServer();
        cerr << "Write error: " << err.message() << endl;
    }
}

void TcpConnectionHandler::SockDiscNotifierServer()
{
    cout<<"server notifier handler for socket disconnect
";
}

/********************************************************************************************************************************************************************/

TcpServer::TcpServer(boost::asio::io_context &io_context, short port_No) : io_context_local(io_context), acceptor_(io_context,
                                                                                                                   tcp::endpoint(tcp::v4(), port_No))
{
    start_accept();
}

void TcpServer::start_accept()
{
    TcpConnectionHandler::pointer connection = TcpConnectionHandler::create(getioContext());
        acceptor_.async_accept(connection->socket(), [this,connection](boost::system::error_code ec) {
        std::cout << "async_accept -> " << ec.message() << "
";
        if (!ec) {
            connection->StartReading();
            start_accept();
        }
    });
}

test_client.cpp 'its client process main function'

//test_client.cpp
#include 

#include "tcpClient.h"

// Client side
int main(int argc, char *argv[])
{
    // we need 2 things: ip address and port number, in that order
    if (argc != 3)
    {
        cerr << "Usage: ip_address port" << endl;
        exit(0);
    }
    char *serverIp = argv[1];
    char *port = argv[2];
    boost::asio::io_context io_context;
    TcpClient tcpcli(io_context,serverIp, port);
     std::thread io_thread([&io_context]() {
        io_context.run();
    });
    tcpcli.StartWriting();
    sleep(1);
    tcpcli.StartReading();
    io_thread.join();
    return 0;
}

This is tcpClient.h

//tcpClient.h
#ifndef TCP_CLIENT_H
#define TCP_CLIENT_H

#include 
#include 
#include 
#include 
#include 

#include "AppPlugin.h"



/**
 *   @class  TcpClient tcpClient.h "client/tcpClient.h"
 *   @brief
 */
class TcpClient
{
public:
    TcpClient(boost::asio::io_context& io_context, string serverIp, string port);
    void readData(const boost::system::error_code&, size_t);
    void sendData(const boost::system::error_code&, size_t);
    enum
    {
        max_length = 4096
    };
    void SockDiscNotifierClient();
    boost::asio::ip::tcp::socket socketClient_;
    void StartReading();
    void StartWriting();

private:
    string serverIp_, port_;
    bool connectStatus;
    boost::asio::streambuf recvBuffer_;
    boost::asio::io_context& io_context_local;
    boost::asio::io_context& getioContext()
    {
        return io_context_local;
    }
    //TestManagerComInfo DeserializeMessageStruct(boost::asio::streambuf&);

};

#endif  // TCP_CLIENT_H

This is tcpClient.cpp In this code I have implemented 3 startReading() and startWriting() functions. The first is implemented by me and working second I have reffered from google and boost documentation 3rd I have taken reference from chatgpt.

//tcpClient.cpp
#include "tcpClient.h"

TcpClient::TcpClient(boost::asio::io_context& io_context, string serverIp, string port)
    : serverIp_(serverIp), port_(port), socketClient_(io_context), io_context_local(io_context)
{
    connectStatus = false;
    if(!connectStatus)
    {
        socketClient_.connect(tcp::endpoint(
        boost::asio::ip::address::from_string(serverIp_), std::stoi(port_)));
        connectStatus = true;
    }
    std::cout << "Client Started: 
"; 
}

//This StartReading is working
#if 1 //first startreading
void TcpClient::StartReading()
{
    // getting response from server
    cout<<"----->TcpClient::StartReading
";
    boost::system::error_code error;
    //boost::asio::streambuf receiveBuff;
    boost::asio::read(socketClient_, recvBuffer_, boost::asio::transfer_all(), error);
    if (error && error != boost::asio::error::eof)
    {
        SockDiscNotifierClient();
        cout << "receive failed: " << error.message() << endl;
    }
    else
    {
        DeserializeMessageStruct(recvBuffer_);        
    }
}
#endif

//This StartReading is not working
#if 0 //second startreading
void TcpClient::StartReading()
{
    cout << "----->TCPClient::StartReading
";
    boost::asio::async_read(socketClient_, recvBuffer_, boost::asio::transfer_all(),
                            [this](const boost::system::error_code &error, std::size_t bytes_transferred)
                            {
                                cout<<"in TCPClient::StartReading labda
";
                               // getioContext().run();
                                if (!error)
                                {
                                    std::string receivedData(boost::asio::buffer_cast(recvBuffer_.data()), boost::asio::buffer_size(recvBuffer_.data()));
                                    cout<<"deserializing client received
"<(socketClient_.get_executor().context());
    auto readOperation = [this,&io_context]() {
        cout << "----->TCPClient::StartReading
";
        socketClient_.async_receive(
        boost::asio::buffer(recvBuffer_.prepare(max_length)),
        [this](const boost::system::error_code &err, size_t bytes_transferred)
        {
            if (!err)
            {
                //recvBuffer_.commit(bytes_transferred); // Commit the received data
                DeserializeMessageStruct(recvBuffer_);
            }
            else
            {
                SockDiscNotifierClient();
                cerr << "Read error: " << err.message() << endl;
            }
        });
        io_context.run();
    };
    std::thread readingThread(readOperation);

    readingThread.join();
}
#endif


#if 1 //First StartWriting
void TcpClient::StartWriting()
{
     // request/message from client
    const string msg = "Hello Server Please send me the strucure!
";
    cout<<"----->TcpClient::StartWritin
";

    socketClient_.async_write_some(
        boost::asio::buffer(msg, max_length),
        boost::bind(&TcpClient::sendData, this,
                    boost::asio::placeholders::error,
                    boost::asio::placeholders::bytes_transferred));

}
#endif

#if 0 //Second StartWriting
void TcpClient::StartWriting()
{
    // Request/message from client
    const string msg = "Hello Server Please send me the structure!
";
    cout << "----->TcpClient::StartWriting
";

    // Start an asynchronous write operation
    socketClient_.async_write_some(
        boost::asio::buffer(msg, max_length),
        [this](const boost::system::error_code &error, std::size_t bytes_transferred)
        {
            if (!error)
            {
                cout << "Client Successfully Sent Message!
";
                // Start reading after writing is complete
                StartReading();
            }
            else
            {
                std::cerr << "Send error: " << error.message() << std::endl;
            }
        });
}
#endif 

#if 0 //Third StartWriting
void TcpClient::StartWriting() {
    // Request/message from client
    boost::asio::io_context& io_context = static_cast(socketClient_.get_executor().context());  
    const string msg = "Hello Server Please send me the structure!
";
    cout << "----->TcpClient::StartWriting
";
    // Create a lambda function that encapsulates the StartWriting logic
    auto writeOperation = [this, &io_context, msg]() {
        socketClient_.async_write_some(
            boost::asio::buffer(msg, max_length),
            [this](const boost::system::error_code &error, std::size_t bytes_transferred) {
                if (!error) {
                    cout << "Client Successfully Sent Message!
";
                    // Start reading after writing is complete
                     StartReading();
                } else {
                    std::cerr << "Send error: " << error.message() << std::endl;
                }
            });
            io_context.run();
    };

    // Create a thread and execute the lambda function
    std::thread writingThread(writeOperation);
   // Wait for the thread to finish
    writingThread.join();
}
#endif

void TcpClient::readData(const boost::system::error_code &err,
                         size_t bytes_transferred)
{
    if (!err)
    {
        cout << "Client Successfuly Read Message! ";
    }
    else
    {
        SockDiscNotifierClient();
        cerr << "Read error: " << err.message() << endl;
    }
}

void TcpClient::sendData(const boost::system::error_code &err,
                         size_t bytes_transferred)
{
    if (!err)
    {
        cout << "Client Successfuly Sent Message!
";
    }
    else
    {
        SockDiscNotifierClient();
        cerr << "Send error: " << err.message() << endl;
    }
}

void TcpClient::SockDiscNotifierClient()
{
    cout<<"client notifier handler for socket disconnect";
    
}

This is AppPlugin.h, which contains serialize and deserialize function also the defination of structure


//AppPlugin.h
#ifndef APPPLUGIN_H
#define APPPLUGIN_H

#include 

#include 
#include 
#include 
#include 

using boost::asio::ip::tcp;
using std::cerr;
using std::cin;
using std::cout;
using std::endl;
using std::string;

enum RequestType
{
    kSatelliteStatus = 0x00,
    kApplicationPresence = 0x01,
    kExecuteTestSuite = 0x02,
    kTestSuiteLog = 0x03,
    kTerminate = 0x04
};

enum StatusType
{
    kInit = 0x00,
    kStarted = 0x01,
    kCompleted = 0x02
};

enum TestResultType
{
    kTestNotExecute,
    kTestPass,
    kTestFail,
    kTestAbort
};

using TestLogType = std::map;

#pragma pack(push, 1)  // Pack the structure
struct TestManagerComInfo
{
    RequestType requestType_;
    std::string name_;
    TestLogType testLogType_;
    StatusType statusType_;

    template 
    void serialize(Archive& ar, const unsigned int version)
    {
        ar& requestType_;
        ar& name_;
        ar& testLogType_;
        ar& statusType_;
    }
};
#pragma pack(pop)  // Restore default packing

/**
* @fn serializeMessageStruct
* @brief Its for serialize the TestManagerComInfo struct and convert into the string
* @param messageInfoNeedtoSerialize is the argunment which is TestManagerComInfo's object
*/
inline string serializeMessageStruct(TestManagerComInfo &messageInfoNeedtoSerialize)
{
    std::ostringstream archiveStream;
    boost::archive::text_oarchive archive(archiveStream);
    archive << messageInfoNeedtoSerialize;
    return archiveStream.str();
}

inline TestManagerComInfo DeserializeMessageStruct(boost::asio::streambuf& deserializeBuff)
{
    TestManagerComInfo receivedMsgStruct;
    std::istream archiveStream(&deserializeBuff);
        boost::archive::text_iarchive archive(archiveStream);
        archive >> receivedMsgStruct;
        cout << "Received Request Type: "
             << static_cast(receivedMsgStruct.requestType_) << std::endl;
        cout << "Received Name: " << receivedMsgStruct.name_ << std::endl;
        cout << "Received Test Log:" << endl;
        for (const auto &test : receivedMsgStruct.testLogType_)
        {
            cout << "  " << test.first << ": " << test.second << endl;
        }
        std::cout << "Received Status Type: "
                  << static_cast(receivedMsgStruct.statusType_)
                  << std::endl;

    return receivedMsgStruct;
}



#endif  // APPPLUGIN_H

Boost Asio async_read and async_receive not getting called itself whenever there is a data on tcp socket But boost::asio::read got call automatically

Answers (1)

Why Is The Server Okay?

Code Review

PUTTING IT ALL TOGETHER

Related Questions