xref: /openbmc/bmcweb/http/server_sent_event.hpp (revision 41fe81c2) 
1 #pragma once
2 #include "boost_formatters.hpp"
3 #include "http_body.hpp"
4 #include "http_request.hpp"
5 #include "http_response.hpp"
6 
7 #include <boost/asio/buffer.hpp>
8 #include <boost/asio/steady_timer.hpp>
9 #include <boost/beast/core/multi_buffer.hpp>
10 #include <boost/beast/websocket.hpp>
11 
12 #include <array>
13 #include <cstddef>
14 #include <functional>
15 #include <optional>
16 
17 namespace crow
18 {
19 
20 namespace sse_socket
21 {
22 struct Connection : public std::enable_shared_from_this<Connection>
23 {
24   public:
25     Connection() = default;
26 
27     Connection(const Connection&) = delete;
28     Connection(Connection&&) = delete;
29     Connection& operator=(const Connection&) = delete;
30     Connection& operator=(const Connection&&) = delete;
31     virtual ~Connection() = default;
32 
33     virtual boost::asio::io_context& getIoContext() = 0;
34     virtual void close(std::string_view msg = "quit") = 0;
35     virtual void sendEvent(std::string_view id, std::string_view msg) = 0;
36 };
37 
38 template <typename Adaptor>
39 class ConnectionImpl : public Connection
40 {
41   public:
ConnectionImpl(Adaptor && adaptorIn,std::function<void (Connection &,const Request &)> openHandlerIn,std::function<void (Connection &)> closeHandlerIn)42     ConnectionImpl(
43         Adaptor&& adaptorIn,
44         std::function<void(Connection&, const Request&)> openHandlerIn,
45         std::function<void(Connection&)> closeHandlerIn) :
46         adaptor(std::move(adaptorIn)),
47         timer(static_cast<boost::asio::io_context&>(
48             adaptor.get_executor().context())),
49         openHandler(std::move(openHandlerIn)),
50         closeHandler(std::move(closeHandlerIn))
51 
52     {
53         BMCWEB_LOG_DEBUG("SseConnectionImpl: SSE constructor {}", logPtr(this));
54     }
55 
56     ConnectionImpl(const ConnectionImpl&) = delete;
57     ConnectionImpl(const ConnectionImpl&&) = delete;
58     ConnectionImpl& operator=(const ConnectionImpl&) = delete;
59     ConnectionImpl& operator=(const ConnectionImpl&&) = delete;
60 
~ConnectionImpl()61     ~ConnectionImpl() override
62     {
63         BMCWEB_LOG_DEBUG("SSE ConnectionImpl: SSE destructor {}", logPtr(this));
64     }
65 
getIoContext()66     boost::asio::io_context& getIoContext() override
67     {
68         return static_cast<boost::asio::io_context&>(
69             adaptor.get_executor().context());
70     }
71 
start(const Request & req)72     void start(const Request& req)
73     {
74         if (!openHandler)
75         {
76             BMCWEB_LOG_CRITICAL("No open handler???");
77             return;
78         }
79         openHandler(*this, req);
80         sendSSEHeader();
81     }
82 
close(const std::string_view msg)83     void close(const std::string_view msg) override
84     {
85         BMCWEB_LOG_DEBUG("Closing connection with reason {}", msg);
86         // send notification to handler for cleanup
87         if (closeHandler)
88         {
89             closeHandler(*this);
90         }
91         BMCWEB_LOG_DEBUG("Closing SSE connection {} - {}", logPtr(this), msg);
92         boost::beast::get_lowest_layer(adaptor).close();
93     }
94 
sendSSEHeader()95     void sendSSEHeader()
96     {
97         BMCWEB_LOG_DEBUG("Starting SSE connection");
98 
99         res.set(boost::beast::http::field::content_type, "text/event-stream");
100         boost::beast::http::response_serializer<BodyType>& serial =
101             serializer.emplace(res);
102 
103         boost::beast::http::async_write_header(
104             adaptor, serial,
105             std::bind_front(&ConnectionImpl::sendSSEHeaderCallback, this,
106                             shared_from_this()));
107     }
108 
sendSSEHeaderCallback(const std::shared_ptr<Connection> &,const boost::system::error_code & ec,size_t)109     void sendSSEHeaderCallback(const std::shared_ptr<Connection>& /*self*/,
110                                const boost::system::error_code& ec,
111                                size_t /*bytesSent*/)
112     {
113         serializer.reset();
114         if (ec)
115         {
116             BMCWEB_LOG_ERROR("Error sending header{}", ec);
117             close("async_write_header failed");
118             return;
119         }
120         BMCWEB_LOG_DEBUG("SSE header sent - Connection established");
121 
122         // SSE stream header sent, So let us setup monitor.
123         // Any read data on this stream will be error in case of SSE.
124         adaptor.async_read_some(boost::asio::buffer(buffer),
125                                 std::bind_front(&ConnectionImpl::afterReadError,
126                                                 this, shared_from_this()));
127     }
128 
afterReadError(const std::shared_ptr<Connection> &,const boost::system::error_code & ec,size_t bytesRead)129     void afterReadError(const std::shared_ptr<Connection>& /*self*/,
130                         const boost::system::error_code& ec, size_t bytesRead)
131     {
132         BMCWEB_LOG_DEBUG("Read {}", bytesRead);
133         if (ec == boost::asio::error::operation_aborted)
134         {
135             return;
136         }
137         if (ec)
138         {
139             BMCWEB_LOG_ERROR("Read error: {}", ec);
140         }
141 
142         close("Close SSE connection");
143     }
144 
doWrite()145     void doWrite()
146     {
147         if (doingWrite)
148         {
149             return;
150         }
151         if (inputBuffer.size() == 0)
152         {
153             BMCWEB_LOG_DEBUG("inputBuffer is empty... Bailing out");
154             return;
155         }
156         startTimeout();
157         doingWrite = true;
158 
159         adaptor.async_write_some(
160             inputBuffer.data(),
161             std::bind_front(&ConnectionImpl::doWriteCallback, this,
162                             shared_from_this()));
163     }
164 
doWriteCallback(const std::shared_ptr<Connection> &,const boost::beast::error_code & ec,size_t bytesTransferred)165     void doWriteCallback(const std::shared_ptr<Connection>& /*self*/,
166                          const boost::beast::error_code& ec,
167                          size_t bytesTransferred)
168     {
169         timer.cancel();
170         doingWrite = false;
171         inputBuffer.consume(bytesTransferred);
172 
173         if (ec == boost::asio::error::eof)
174         {
175             BMCWEB_LOG_ERROR("async_write_some() SSE stream closed");
176             close("SSE stream closed");
177             return;
178         }
179 
180         if (ec)
181         {
182             BMCWEB_LOG_ERROR("async_write_some() failed: {}", ec.message());
183             close("async_write_some failed");
184             return;
185         }
186         BMCWEB_LOG_DEBUG("async_write_some() bytes transferred: {}",
187                          bytesTransferred);
188 
189         doWrite();
190     }
191 
sendEvent(std::string_view id,std::string_view msg)192     void sendEvent(std::string_view id, std::string_view msg) override
193     {
194         if (msg.empty())
195         {
196             BMCWEB_LOG_DEBUG("Empty data, bailing out.");
197             return;
198         }
199 
200         dataFormat(id, msg);
201 
202         doWrite();
203     }
204 
dataFormat(std::string_view id,std::string_view msg)205     void dataFormat(std::string_view id, std::string_view msg)
206     {
207         constexpr size_t bufferLimit = 10485760U; // 10MB
208         if (id.size() + msg.size() + inputBuffer.size() >= bufferLimit)
209         {
210             BMCWEB_LOG_ERROR("SSE Buffer overflow while waiting for client");
211             close("Buffer overflow");
212             return;
213         }
214         std::string rawData;
215         if (!id.empty())
216         {
217             rawData += "id: ";
218             rawData.append(id);
219             rawData += "\n";
220         }
221 
222         rawData += "data: ";
223         for (char character : msg)
224         {
225             rawData += character;
226             if (character == '\n')
227             {
228                 rawData += "data: ";
229             }
230         }
231         rawData += "\n\n";
232 
233         size_t copied = boost::asio::buffer_copy(
234             inputBuffer.prepare(rawData.size()), boost::asio::buffer(rawData));
235         inputBuffer.commit(copied);
236     }
237 
startTimeout()238     void startTimeout()
239     {
240         std::weak_ptr<Connection> weakSelf = weak_from_this();
241         timer.expires_after(std::chrono::seconds(30));
242         timer.async_wait(std::bind_front(&ConnectionImpl::onTimeoutCallback,
243                                          this, weak_from_this()));
244     }
245 
onTimeoutCallback(const std::weak_ptr<Connection> & weakSelf,const boost::system::error_code & ec)246     void onTimeoutCallback(const std::weak_ptr<Connection>& weakSelf,
247                            const boost::system::error_code& ec)
248     {
249         std::shared_ptr<Connection> self = weakSelf.lock();
250         if (!self)
251         {
252             BMCWEB_LOG_CRITICAL("{} Failed to capture connection",
253                                 logPtr(self.get()));
254             return;
255         }
256 
257         if (ec == boost::asio::error::operation_aborted)
258         {
259             BMCWEB_LOG_DEBUG("Timer operation aborted");
260             // Canceled wait means the path succeeded.
261             return;
262         }
263         if (ec)
264         {
265             BMCWEB_LOG_CRITICAL("{} timer failed {}", logPtr(self.get()), ec);
266         }
267 
268         BMCWEB_LOG_WARNING("{} Connection timed out, closing",
269                            logPtr(self.get()));
270 
271         self->close("closing connection");
272     }
273 
274   private:
275     std::array<char, 1> buffer{};
276     boost::beast::multi_buffer inputBuffer;
277 
278     Adaptor adaptor;
279 
280     using BodyType = bmcweb::HttpBody;
281     boost::beast::http::response<BodyType> res;
282     std::optional<boost::beast::http::response_serializer<BodyType>> serializer;
283     boost::asio::steady_timer timer;
284     bool doingWrite = false;
285 
286     std::function<void(Connection&, const Request&)> openHandler;
287     std::function<void(Connection&)> closeHandler;
288 };
289 } // namespace sse_socket
290 } // namespace crow
291