tag::run_session[]
| 228 | |
| 229 | // tag::run_session[] |
| 230 | cobalt::promise<void> run_session(beast::websocket::stream<socket_type> st, |
| 231 | cobalt::channel<subscription> & subc) |
| 232 | try |
| 233 | { |
| 234 | http::request<http::empty_body> req; |
| 235 | beast::flat_buffer buf; |
| 236 | co_await http::async_read(st.next_layer(), buf, req); // <1> |
| 237 | // check the target |
| 238 | auto r = urls::parse_uri_reference(req.target()); |
| 239 | if (r.has_error() || (r->segments().size() != 2u)) // <2> |
| 240 | { |
| 241 | http::response<http::string_body> res{http::status::bad_request, 11}; |
| 242 | res.body() = r.has_error() ? r.error().message() : |
| 243 | "url needs two segments, e.g. /btc/usd"; |
| 244 | co_await http::async_write(st.next_layer(), res); |
| 245 | st.next_layer().close(); |
| 246 | co_return ; |
| 247 | } |
| 248 | |
| 249 | co_await st.async_accept(req); // <3> |
| 250 | |
| 251 | auto sym = std::string(r->segments().front()) + "-" + |
| 252 | std::string(r->segments().back()); |
| 253 | boost::algorithm::to_upper(sym); |
| 254 | // close when data gets sent |
| 255 | auto p = read_and_close(st, std::move(buf)); // <4> |
| 256 | |
| 257 | auto ptr = std::make_shared<cobalt::channel<json::object>>(1u); // <5> |
| 258 | co_await subc.write(subscription{sym, ptr}); // <6> |
| 259 | |
| 260 | while (ptr->is_open() && st.is_open()) // <7> |
| 261 | { |
| 262 | auto bb = json::serialize(co_await ptr->read()); |
| 263 | co_await st.async_write(asio::buffer(bb)); |
| 264 | } |
| 265 | |
| 266 | co_await st.async_close(beast::websocket::close_code::going_away, |
| 267 | asio::as_tuple(cobalt::use_op)); // <8> |
| 268 | st.next_layer().close(); |
| 269 | co_await p; // <9> |
| 270 | |
| 271 | } |
| 272 | catch(std::exception & e) |
| 273 | { |
| 274 | std::cerr << "Session ended with exception: " << e.what() << std::endl; |
| 275 | } |
| 276 | // end::run_session[] |
| 277 | |
| 278 | // tag::main[] |