summaryrefslogtreecommitdiffstatshomepage
path: root/3rdparty/asio/src/examples/cpp11/operations/composed_7.cpp
blob: 8eb83dc767965a45fe7d90982c39259a34fe9eb1 (plain) (blame)
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
//
// composed_7.cpp
// ~~~~~~~~~~~~~~
//
// Copyright (c) 2003-2024 Christopher M. Kohlhoff (chris at kohlhoff dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//

#include <asio/compose.hpp>
#include <asio/deferred.hpp>
#include <asio/io_context.hpp>
#include <asio/ip/tcp.hpp>
#include <asio/steady_timer.hpp>
#include <asio/use_future.hpp>
#include <asio/write.hpp>
#include <functional>
#include <iostream>
#include <memory>
#include <sstream>
#include <string>
#include <type_traits>
#include <utility>

using asio::ip::tcp;

// NOTE: This example requires the new asio::async_compose function. For
// an example that works with the Networking TS style of completion tokens,
// please see an older version of asio.

//------------------------------------------------------------------------------

// This composed operation shows composition of multiple underlying operations.
// It automatically serialises a message, using its I/O streams insertion
// operator, before sending it N times on the socket. To do this, it must
// allocate a buffer for the encoded message and ensure this buffer's validity
// until all underlying async_write operation complete. A one second delay is
// inserted prior to each write operation, using a steady_timer.

// In this example, the composed operation's logic is implemented as a state
// machine within a hand-crafted function object.
struct async_write_messages_implementation
{
  // The implementation holds a reference to the socket as it is used for
  // multiple async_write operations.
  tcp::socket& socket_;

  // The allocated buffer for the encoded message. The std::unique_ptr smart
  // pointer is move-only, and as a consequence our implementation is also
  // move-only.
  std::unique_ptr<std::string> encoded_message_;

  // The repeat count remaining.
  std::size_t repeat_count_;

  // A steady timer used for introducing a delay.
  std::unique_ptr<asio::steady_timer> delay_timer_;

  // To manage the cycle between the multiple underlying asychronous
  // operations, our implementation is a state machine.
  enum { starting, waiting, writing } state_;

  // The first argument to our function object's call operator is a reference
  // to the enclosing intermediate completion handler. This intermediate
  // completion handler is provided for us by the asio::async_compose
  // function, and takes care of all the details required to implement a
  // conforming asynchronous operation. When calling an underlying asynchronous
  // operation, we pass it this enclosing intermediate completion handler
  // as the completion token.
  //
  // All arguments after the first must be defaulted to allow the state machine
  // to be started, as well as to allow the completion handler to match the
  // completion signature of both the async_write and steady_timer::async_wait
  // operations.
  template <typename Self>
  void operator()(Self& self,
      const std::error_code& error = std::error_code(),
      std::size_t = 0)
  {
    if (!error)
    {
      switch (state_)
      {
      case starting:
      case writing:
        if (repeat_count_ > 0)
        {
          --repeat_count_;
          state_ = waiting;
          delay_timer_->expires_after(std::chrono::seconds(1));
          delay_timer_->async_wait(std::move(self));
          return; // Composed operation not yet complete.
        }
        break; // Composed operation complete, continue below.
      case waiting:
        state_ = writing;
        asio::async_write(socket_,
            asio::buffer(*encoded_message_), std::move(self));
        return; // Composed operation not yet complete.
      }
    }

    // This point is reached only on completion of the entire composed
    // operation.

    // Deallocate the encoded message and delay timer before calling the
    // user-supplied completion handler.
    encoded_message_.reset();
    delay_timer_.reset();

    // Call the user-supplied handler with the result of the operation.
    self.complete(error);
  }
};

template <typename T, typename CompletionToken>
auto async_write_messages(tcp::socket& socket,
    const T& message, std::size_t repeat_count,
    CompletionToken&& token)
  // The return type of the initiating function is deduced from the combination
  // of:
  //
  // - the CompletionToken type,
  // - the completion handler signature, and
  // - the asynchronous operation's initiation function object.
  //
  // When the completion token is a simple callback, the return type is always
  // void. In this example, when the completion token is asio::yield_context
  // (used for stackful coroutines) the return type would also be void, as
  // there is no non-error argument to the completion handler. When the
  // completion token is asio::use_future it would be std::future<void>. When
  // the completion token is asio::deferred, the return type differs for each
  // asynchronous operation.
  //
  // In C++11 we deduce the type from the call to asio::async_compose.
  -> decltype(
      asio::async_compose<
        CompletionToken, void(std::error_code)>(
          std::declval<async_write_messages_implementation>(),
          token, socket))
{
  // Encode the message and copy it into an allocated buffer. The buffer will
  // be maintained for the lifetime of the composed asynchronous operation.
  std::ostringstream os;
  os << message;
  std::unique_ptr<std::string> encoded_message(new std::string(os.str()));

  // Create a steady_timer to be used for the delay between messages.
  std::unique_ptr<asio::steady_timer> delay_timer(
      new asio::steady_timer(socket.get_executor()));

  // The asio::async_compose function takes:
  //
  // - our asynchronous operation implementation,
  // - the completion token,
  // - the completion handler signature, and
  // - any I/O objects (or executors) used by the operation
  //
  // It then wraps our implementation in an intermediate completion handler
  // that meets the requirements of a conforming asynchronous operation. This
  // includes tracking outstanding work against the I/O executors associated
  // with the operation (in this example, this is the socket's executor).
  return asio::async_compose<
    CompletionToken, void(std::error_code)>(
      async_write_messages_implementation{
        socket, std::move(encoded_message),
        repeat_count, std::move(delay_timer),
        async_write_messages_implementation::starting},
      token, socket);
}

//------------------------------------------------------------------------------

void test_callback()
{
  asio::io_context io_context;

  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
  tcp::socket socket = acceptor.accept();

  // Test our asynchronous operation using a lambda as a callback.
  async_write_messages(socket, "Testing callback\r\n", 5,
      [](const std::error_code& error)
      {
        if (!error)
        {
          std::cout << "Messages sent\n";
        }
        else
        {
          std::cout << "Error: " << error.message() << "\n";
        }
      });

  io_context.run();
}

//------------------------------------------------------------------------------

void test_deferred()
{
  asio::io_context io_context;

  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
  tcp::socket socket = acceptor.accept();

  // Test our asynchronous operation using the deferred completion token. This
  // token causes the operation's initiating function to package up the
  // operation and its arguments to return a function object, which may then be
  // used to launch the asynchronous operation.
  auto op = async_write_messages(socket,
      "Testing deferred\r\n", 5, asio::deferred);

  // Launch the operation using a lambda as a callback.
  std::move(op)(
      [](const std::error_code& error)
      {
        if (!error)
        {
          std::cout << "Messages sent\n";
        }
        else
        {
          std::cout << "Error: " << error.message() << "\n";
        }
      });

  io_context.run();
}

//------------------------------------------------------------------------------

void test_future()
{
  asio::io_context io_context;

  tcp::acceptor acceptor(io_context, {tcp::v4(), 55555});
  tcp::socket socket = acceptor.accept();

  // Test our asynchronous operation using the use_future completion token.
  // This token causes the operation's initiating function to return a future,
  // which may be used to synchronously wait for the result of the operation.
  std::future<void> f = async_write_messages(
      socket, "Testing future\r\n", 5, asio::use_future);

  io_context.run();

  try
  {
    // Get the result of the operation.
    f.get();
    std::cout << "Messages sent\n";
  }
  catch (const std::exception& e)
  {
    std::cout << "Error: " << e.what() << "\n";
  }
}

//------------------------------------------------------------------------------

int main()
{
  test_callback();
  test_deferred();
  test_future();
}