telemetry_server.cpp
6.28 KB
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
#include <websocketpp/config/asio_no_tls.hpp>
#include <websocketpp/server.hpp>
#include <fstream>
#include <iostream>
#include <set>
#include <streambuf>
#include <string>
/**
* The telemetry server accepts connections and sends a message every second to
* each client containing an integer count. This example can be used as the
* basis for programs that expose a stream of telemetry data for logging,
* dashboards, etc.
*
* This example uses the timer based concurrency method and is self contained
* and singled threaded. Refer to telemetry client for an example of a similar
* telemetry setup using threads rather than timers.
*
* This example also includes an example simple HTTP server that serves a web
* dashboard displaying the count. This simple design is suitable for use
* delivering a small number of files to a small number of clients. It is ideal
* for cases like embedded dashboards that don't want the complexity of an extra
* HTTP server to serve static files.
*
* This design *will* fall over under high traffic or DoS conditions. In such
* cases you are much better off proxying to a real HTTP server for the http
* requests.
*/
class telemetry_server {
public:
typedef websocketpp::connection_hdl connection_hdl;
typedef websocketpp::server<websocketpp::config::asio> server;
telemetry_server() : m_count(0) {
// set up access channels to only log interesting things
m_endpoint.clear_access_channels(websocketpp::log::alevel::all);
m_endpoint.set_access_channels(websocketpp::log::alevel::access_core);
m_endpoint.set_access_channels(websocketpp::log::alevel::app);
// Initialize the Asio transport policy
m_endpoint.init_asio();
// Bind the handlers we are using
using websocketpp::lib::placeholders::_1;
using websocketpp::lib::bind;
m_endpoint.set_open_handler(bind(&telemetry_server::on_open,this,_1));
m_endpoint.set_close_handler(bind(&telemetry_server::on_close,this,_1));
m_endpoint.set_http_handler(bind(&telemetry_server::on_http,this,_1));
}
void run(std::string docroot, uint16_t port) {
std::stringstream ss;
ss << "Running telemetry server on port "<< port <<" using docroot=" << docroot;
m_endpoint.get_alog().write(websocketpp::log::alevel::app,ss.str());
m_docroot = docroot;
// listen on specified port
m_endpoint.listen(port);
// Start the server accept loop
m_endpoint.start_accept();
// Set the initial timer to start telemetry
set_timer();
// Start the ASIO io_service run loop
try {
m_endpoint.run();
} catch (websocketpp::exception const & e) {
std::cout << e.what() << std::endl;
}
}
void set_timer() {
m_timer = m_endpoint.set_timer(
1000,
websocketpp::lib::bind(
&telemetry_server::on_timer,
this,
websocketpp::lib::placeholders::_1
)
);
}
void on_timer(websocketpp::lib::error_code const & ec) {
if (ec) {
// there was an error, stop telemetry
m_endpoint.get_alog().write(websocketpp::log::alevel::app,
"Timer Error: "+ec.message());
return;
}
std::stringstream val;
val << "count is " << m_count++;
// Broadcast count to all connections
con_list::iterator it;
for (it = m_connections.begin(); it != m_connections.end(); ++it) {
m_endpoint.send(*it,val.str(),websocketpp::frame::opcode::text);
}
// set timer for next telemetry check
set_timer();
}
void on_http(connection_hdl hdl) {
// Upgrade our connection handle to a full connection_ptr
server::connection_ptr con = m_endpoint.get_con_from_hdl(hdl);
std::ifstream file;
std::string filename = con->get_resource();
std::string response;
m_endpoint.get_alog().write(websocketpp::log::alevel::app,
"http request1: "+filename);
if (filename == "/") {
filename = m_docroot+"index.html";
} else {
filename = m_docroot+filename.substr(1);
}
m_endpoint.get_alog().write(websocketpp::log::alevel::app,
"http request2: "+filename);
file.open(filename.c_str(), std::ios::in);
if (!file) {
// 404 error
std::stringstream ss;
ss << "<!doctype html><html><head>"
<< "<title>Error 404 (Resource not found)</title><body>"
<< "<h1>Error 404</h1>"
<< "<p>The requested URL " << filename << " was not found on this server.</p>"
<< "</body></head></html>";
con->set_body(ss.str());
con->set_status(websocketpp::http::status_code::not_found);
return;
}
file.seekg(0, std::ios::end);
response.reserve(file.tellg());
file.seekg(0, std::ios::beg);
response.assign((std::istreambuf_iterator<char>(file)),
std::istreambuf_iterator<char>());
con->set_body(response);
con->set_status(websocketpp::http::status_code::ok);
}
void on_open(connection_hdl hdl) {
m_connections.insert(hdl);
}
void on_close(connection_hdl hdl) {
m_connections.erase(hdl);
}
private:
typedef std::set<connection_hdl,std::owner_less<connection_hdl>> con_list;
server m_endpoint;
con_list m_connections;
server::timer_ptr m_timer;
std::string m_docroot;
// Telemetry data
uint64_t m_count;
};
int main(int argc, char* argv[]) {
telemetry_server s;
std::string docroot;
uint16_t port = 9002;
if (argc == 1) {
std::cout << "Usage: telemetry_server [documentroot] [port]" << std::endl;
return 1;
}
if (argc >= 2) {
docroot = std::string(argv[1]);
}
if (argc >= 3) {
int i = atoi(argv[2]);
if (i <= 0 || i > 65535) {
std::cout << "invalid port" << std::endl;
return 1;
}
port = uint16_t(i);
}
s.run(docroot, port);
return 0;
}