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
|
//
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail 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)
//
// Official repository: https://github.com/vinniefalco/CppCon2018
//
#include "websocket_session.hpp"
#include <iostream>
websocket_session::
websocket_session(
tcp::socket&& socket,
boost::shared_ptr<shared_state> const& state)
: ws_(std::move(socket))
, state_(state)
{
}
websocket_session::
~websocket_session()
{
// Remove this session from the list of active sessions
state_->leave(this);
}
void
websocket_session::
fail(beast::error_code ec, char const* what)
{
// Don't report these
if( ec == net::error::operation_aborted ||
ec == websocket::error::closed)
return;
std::cerr << what << ": " << ec.message() << "\n";
}
void
websocket_session::
on_accept(beast::error_code ec)
{
// Handle the error, if any
if(ec)
return fail(ec, "accept");
// Add this session to the list of active sessions
state_->join(this);
// Read a message
ws_.async_read(
buffer_,
beast::bind_front_handler(
&websocket_session::on_read,
shared_from_this()));
}
void
websocket_session::
on_read(beast::error_code ec, std::size_t)
{
// Handle the error, if any
if(ec)
return fail(ec, "read");
// Send to all connections
state_->send(beast::buffers_to_string(buffer_.data()));
// Clear the buffer
buffer_.consume(buffer_.size());
// Read another message
ws_.async_read(
buffer_,
beast::bind_front_handler(
&websocket_session::on_read,
shared_from_this()));
}
void
websocket_session::
send(boost::shared_ptr<std::string const> const& ss)
{
// Post our work to the strand, this ensures
// that the members of `this` will not be
// accessed concurrently.
net::post(
ws_.get_executor(),
beast::bind_front_handler(
&websocket_session::on_send,
shared_from_this(),
ss));
}
void
websocket_session::
on_send(boost::shared_ptr<std::string const> const& ss)
{
// Always add to queue
queue_.push_back(ss);
// Are we already writing?
if(queue_.size() > 1)
return;
// We are not currently writing, so send this immediately
ws_.async_write(
net::buffer(*queue_.front()),
beast::bind_front_handler(
&websocket_session::on_write,
shared_from_this()));
}
void
websocket_session::
on_write(beast::error_code ec, std::size_t)
{
// Handle the error, if any
if(ec)
return fail(ec, "write");
// Remove the string from the queue
queue_.erase(queue_.begin());
// Send the next message if any
if(! queue_.empty())
ws_.async_write(
net::buffer(*queue_.front()),
beast::bind_front_handler(
&websocket_session::on_write,
shared_from_this()));
}
|
