muduo/examples/ace/ttcp/ttcp_asio_sync.cc at 8b7af3d18f2b161c0126bc9cd0d957fe5daa69fc · chenshuo/muduo · 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
#include <examples/ace/ttcp/common.h>

#include <muduo/base/Logging.h>
#include <boost/asio.hpp>
#include <stdio.h>

using boost::asio::ip::tcp;

void transmit(const Options& opt)
{
  try
  {
  }
  catch (std::exception& e)
  {
    LOG_ERROR << e.what();
  }
}

void receive(const Options& opt)
{
  try
  {
#if BOOST_VERSION < 107000L
    boost::asio::io_service io_service;
    tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), opt.port));
    tcp::socket socket(io_service);
#else
    boost::asio::io_context io_context;
    tcp::acceptor acceptor(io_context, tcp::endpoint(tcp::v4(), opt.port));
    tcp::socket socket(io_context);
#endif

    acceptor.accept(socket);

    struct SessionMessage sessionMessage = { 0, 0 };
    boost::system::error_code error;
    size_t nr = boost::asio::read(socket, boost::asio::buffer(&sessionMessage, sizeof sessionMessage),
#if BOOST_VERSION < 104700L
                                  boost::asio::transfer_all(),
#endif
                                  error);
    if (nr != sizeof sessionMessage)
    {
      LOG_ERROR << "read session message: " << error.message();
      exit(1);
    }

    sessionMessage.number = ntohl(sessionMessage.number);
    sessionMessage.length = ntohl(sessionMessage.length);
    printf("receive number = %d\nreceive length = %d\n",
           sessionMessage.number, sessionMessage.length);
    const int total_len = static_cast<int>(sizeof(int32_t) + sessionMessage.length);
    PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));
    std::unique_ptr<PayloadMessage, void (*)(void*)> freeIt(payload, ::free);
    assert(payload);

    for (int i = 0; i < sessionMessage.number; ++i)
    {
      payload->length = 0;
      if (boost::asio::read(socket, boost::asio::buffer(&payload->length, sizeof(payload->length)),
#if BOOST_VERSION < 104700L
                            boost::asio::transfer_all(),
#endif
                            error) != sizeof(payload->length))
      {
        LOG_ERROR << "read length: " << error.message();
        exit(1);
      }
      payload->length = ntohl(payload->length);
      assert(payload->length == sessionMessage.length);
      if (boost::asio::read(socket, boost::asio::buffer(payload->data, payload->length),
#if BOOST_VERSION < 104700L
                            boost::asio::transfer_all(),
#endif
                            error) != static_cast<size_t>(payload->length))
      {
        LOG_ERROR << "read payload data: " << error.message();
        exit(1);
      }
      int32_t ack = htonl(payload->length);
      if (boost::asio::write(socket, boost::asio::buffer(&ack, sizeof(ack))) != sizeof(ack))
      {
        LOG_ERROR << "write ack: " << error.message();
        exit(1);
      }
    }
  }
  catch (std::exception& e)
  {
    LOG_ERROR << e.what();
  }
}