openMSX
CliConnection.cc
Go to the documentation of this file.
1 // TODO:
2 // - To avoid any possible conflicts, anything called from "run" should be
3 // locked.
4 // - Maybe document for each method whether it is called from the listener
5 // thread or from the main thread?
6 // - Unsubscribe at CliComm after stream is closed.
7 
8 #include "CliConnection.hh"
9 #include "EventDistributor.hh"
10 #include "Event.hh"
11 #include "CommandController.hh"
12 #include "CommandException.hh"
13 #include "TclObject.hh"
14 #include "TemporaryString.hh"
15 #include "XMLEscape.hh"
16 #include "cstdiop.hh"
17 #include "ranges.hh"
18 #include "unistdp.hh"
19 #include <cassert>
20 #include <iostream>
21 
22 #ifdef _WIN32
23 #include "SocketStreamWrapper.hh"
24 #include "SspiNegotiateServer.hh"
25 #endif
26 
27 namespace openmsx {
28 
29 // class CliConnection
30 
32  EventDistributor& eventDistributor_)
33  : parser([this](const std::string& cmd) { execute(cmd); })
34  , commandController(commandController_)
35  , eventDistributor(eventDistributor_)
36 {
37  ranges::fill(updateEnabled, false);
38 
39  eventDistributor.registerEventListener(EventType::CLICOMMAND, *this);
40 }
41 
43 {
44  eventDistributor.unregisterEventListener(EventType::CLICOMMAND, *this);
45 }
46 
47 void CliConnection::log(CliComm::LogLevel level, std::string_view message) noexcept
48 {
49  auto levelStr = CliComm::getLevelStrings();
50  output(tmpStrCat("<log level=\"", levelStr[level], "\">",
51  XMLEscape(message), "</log>\n"));
52 }
53 
54 void CliConnection::update(CliComm::UpdateType type, std::string_view machine,
55  std::string_view name, std::string_view value) noexcept
56 {
57  if (!getUpdateEnable(type)) return;
58 
59  auto updateStr = CliComm::getUpdateStrings();
60  auto tmp = strCat("<update type=\"", updateStr[type], '\"');
61  if (!machine.empty()) {
62  strAppend(tmp, " machine=\"", machine, '\"');
63  }
64  if (!name.empty()) {
65  strAppend(tmp, " name=\"", XMLEscape(name), '\"');
66  }
67  strAppend(tmp, '>', XMLEscape(value), "</update>\n");
68 
69  output(tmp);
70 }
71 
73 {
74  output("<openmsx-output>\n");
75 }
76 
78 {
79  thread = std::thread([this]() { run(); });
80 }
81 
83 {
84  output("</openmsx-output>\n");
85  close();
86 
87  poller.abort();
88  // Thread might not be running if start() was never called.
89  if (thread.joinable()) {
90  thread.join();
91  }
92 }
93 
94 void CliConnection::execute(const std::string& command)
95 {
96  eventDistributor.distributeEvent(
97  Event::create<CliCommandEvent>(command, this));
98 }
99 
100 static TemporaryString reply(std::string_view message, bool status)
101 {
102  return tmpStrCat("<reply result=\"", (status ? "ok" : "nok"), "\">",
103  XMLEscape(message), "</reply>\n");
104 }
105 
106 int CliConnection::signalEvent(const Event& event) noexcept
107 {
108  assert(getType(event) == EventType::CLICOMMAND);
109  const auto& commandEvent = get<CliCommandEvent>(event);
110  if (commandEvent.getId() == this) {
111  try {
112  auto result = commandController.executeCommand(
113  commandEvent.getCommand(), this).getString();
114  output(reply(result, true));
115  } catch (CommandException& e) {
116  std::string result = std::move(e).getMessage() + '\n';
117  output(reply(result, false));
118  }
119  }
120  return 0;
121 }
122 
123 
124 // class StdioConnection
125 
126 constexpr int BUF_SIZE = 4096;
128  EventDistributor& eventDistributor_)
129  : CliConnection(commandController_, eventDistributor_)
130 {
131  startOutput();
132 }
133 
135 {
136  end();
137 }
138 
139 void StdioConnection::run()
140 {
141  // runs in helper thread
142  while (true) {
143 #ifdef _WIN32
144  if (poller.aborted()) break;
145 #else
146  if (poller.poll(STDIN_FILENO)) break;
147 #endif
148  char buf[BUF_SIZE];
149  int n = read(STDIN_FILENO, buf, sizeof(buf));
150  if (n > 0) {
151  parser.parse(buf, n);
152  } else if (n < 0) {
153  break;
154  }
155  }
156 }
157 
158 void StdioConnection::output(std::string_view message)
159 {
160  std::cout << message << std::flush;
161 }
162 
163 void StdioConnection::close()
164 {
165  // don't close stdin/out/err
166 }
167 
168 
169 #ifdef _WIN32
170 // class PipeConnection
171 
172 // INVALID_HANDLE_VALUE is #defined as (HANDLE)(-1)
173 // but that gives a old-style-cast warning
174 static const HANDLE OPENMSX_INVALID_HANDLE_VALUE = reinterpret_cast<HANDLE>(-1);
175 
176 PipeConnection::PipeConnection(CommandController& commandController_,
177  EventDistributor& eventDistributor_,
178  std::string_view name)
179  : CliConnection(commandController_, eventDistributor_)
180 {
181  auto pipeName = strCat("\\\\.\\pipe\\", name);
182  pipeHandle = CreateFileA(pipeName.c_str(), GENERIC_READ, 0, nullptr,
183  OPEN_EXISTING, FILE_FLAG_OVERLAPPED, nullptr);
184  if (pipeHandle == OPENMSX_INVALID_HANDLE_VALUE) {
185  throw FatalError("Error reopening pipefile '", pipeName, "': error ",
186  unsigned(GetLastError()));
187  }
188 
189  shutdownEvent = CreateEventW(nullptr, FALSE, FALSE, nullptr);
190  if (!shutdownEvent) {
191  throw FatalError("Error creating shutdown event: ", GetLastError());
192  }
193 
194  startOutput();
195 }
196 
197 PipeConnection::~PipeConnection()
198 {
199  end();
200 
201  assert(pipeHandle == OPENMSX_INVALID_HANDLE_VALUE);
202  CloseHandle(shutdownEvent);
203 }
204 
205 static void InitOverlapped(LPOVERLAPPED overlapped)
206 {
207  ZeroMemory(overlapped, sizeof(*overlapped));
208  overlapped->hEvent = CreateEventW(nullptr, FALSE, FALSE, nullptr);
209  if (!overlapped->hEvent) {
210  throw FatalError("Error creating overlapped event: ", GetLastError());
211  }
212 }
213 
214 static void ClearOverlapped(LPOVERLAPPED overlapped)
215 {
216  if (overlapped->hEvent) {
217  CloseHandle(overlapped->hEvent);
218  overlapped->hEvent = nullptr;
219  }
220 }
221 
222 void PipeConnection::run()
223 {
224  // runs in helper thread
225  OVERLAPPED overlapped;
226  InitOverlapped(&overlapped);
227  HANDLE waitHandles[2] = { shutdownEvent, overlapped.hEvent };
228 
229  while (pipeHandle != OPENMSX_INVALID_HANDLE_VALUE) {
230  char buf[BUF_SIZE];
231  if (!ReadFile(pipeHandle, buf, BUF_SIZE, nullptr, &overlapped) &&
232  GetLastError() != ERROR_IO_PENDING) {
233  break; // Pipe broke
234  }
235  DWORD wait = WaitForMultipleObjects(2, waitHandles, FALSE, INFINITE);
236  if (wait == WAIT_OBJECT_0 + 1) {
237  DWORD bytesRead;
238  if (!GetOverlappedResult(pipeHandle, &overlapped, &bytesRead, TRUE)) {
239  break; // Pipe broke
240  }
241  parser.parse(buf, bytesRead);
242  } else if (wait == WAIT_OBJECT_0) {
243  break; // Shutdown
244  } else {
245  throw FatalError(
246  "WaitForMultipleObjects returned unexpectedly: ", wait);
247  }
248  }
249 
250  ClearOverlapped(&overlapped);
251 
252  // We own the pipe handle, so close it here
253  CloseHandle(pipeHandle);
254  pipeHandle = OPENMSX_INVALID_HANDLE_VALUE;
255 }
256 
257 void PipeConnection::output(std::string_view message)
258 {
259  if (pipeHandle != OPENMSX_INVALID_HANDLE_VALUE) {
260  std::cout << message << std::flush;
261  }
262 }
263 
264 void PipeConnection::close()
265 {
266  SetEvent(shutdownEvent);
267 }
268 #endif // _WIN32
269 
270 
271 // class SocketConnection
272 
274  EventDistributor& eventDistributor_,
275  SOCKET sd_)
276  : CliConnection(commandController_, eventDistributor_)
277  , sd(sd_), established(false)
278 {
279 }
280 
282 {
283  end();
284 }
285 
286 void SocketConnection::run()
287 {
288  // runs in helper thread
289 #ifdef _WIN32
290  bool ok;
291  {
292  std::lock_guard<std::mutex> lock(sdMutex);
293  // Authenticate and authorize the caller
294  SocketStreamWrapper stream(sd);
295  SspiNegotiateServer server(stream);
296  ok = server.Authenticate() && server.Authorize();
297  }
298  if (!ok) {
299  closeSocket();
300  return;
301  }
302 #endif
303  // Start output element
304  established = true; // TODO needs locking?
305  startOutput();
306 
307  // TODO is locking correct?
308  // No need to lock in this thread because we don't write to 'sd'
309  // and 'sd' only gets written to in this thread.
310  while (true) {
311  if (sd == OPENMSX_INVALID_SOCKET) return;
312 #ifndef _WIN32
313  if (poller.poll(sd)) {
314  break;
315  }
316 #endif
317  char buf[BUF_SIZE];
318  int n = sock_recv(sd, buf, BUF_SIZE);
319  if (n > 0) {
320  parser.parse(buf, n);
321  } else if (n < 0) {
322  break;
323  }
324  }
325  closeSocket();
326 }
327 
328 void SocketConnection::output(std::string_view message)
329 {
330  if (!established) { // TODO needs locking?
331  // Connection isn't authorized yet (and opening tag is not
332  // yet send). Ignore log and update messages for now.
333  return;
334  }
335  const char* data = message.data();
336  unsigned pos = 0;
337  size_t bytesLeft = message.size();
338  while (bytesLeft) {
339  int bytesSend;
340  {
341  std::lock_guard<std::mutex> lock(sdMutex);
342  if (sd == OPENMSX_INVALID_SOCKET) return;
343  bytesSend = sock_send(sd, &data[pos], bytesLeft);
344  }
345  if (bytesSend > 0) {
346  bytesLeft -= bytesSend;
347  pos += bytesSend;
348  } else {
349  // Note: On Windows we rely on closing the socket to
350  // wake up the worker thread, on other platforms
351  // we rely on Poller.
352  closeSocket();
353  poller.abort();
354  break;
355  }
356  }
357 }
358 
359 void SocketConnection::closeSocket()
360 {
361  std::lock_guard<std::mutex> lock(sdMutex);
362  if (sd != OPENMSX_INVALID_SOCKET) {
363  SOCKET _sd = sd;
365  sock_close(_sd);
366  }
367 }
368 
369 void SocketConnection::close()
370 {
371  closeSocket();
372 }
373 
374 } // namespace openmsx
void XMLEscape(std::string_view s, Output output)
Definition: XMLEscape.hh:21
void parse(const char *buf, size_t n)
TemporaryString.
static span< const char *const > getLevelStrings()
Definition: CliComm.hh:82
static span< const char *const > getUpdateStrings()
Definition: CliComm.hh:88
void end()
End this connection by sending the closing tag and then closing the stream.
virtual void close()=0
Close the connection.
void start()
Starts the helper thread.
AdhocCliCommParser parser
void startOutput()
Send opening XML tag, should be called exactly once by a subclass shortly after opening a connection.
virtual void output(std::string_view message)=0
CliConnection(CommandController &commandController, EventDistributor &eventDistributor)
void unregisterEventListener(EventType type, EventListener &listener)
Unregisters a previously registered event listener.
void distributeEvent(Event &&event)
Schedule the given event for delivery.
bool poll(int fd)
Waits for an event to occur on the given file descriptor.
Definition: Poller.cc:43
bool aborted()
Returns true iff abort() was called.
Definition: Poller.hh:28
void abort()
Aborts a poll in progress and any future poll attempts.
Definition: Poller.cc:31
SocketConnection(CommandController &commandController, EventDistributor &eventDistributor, SOCKET sd)
void output(std::string_view message) override
void output(std::string_view message) override
StdioConnection(CommandController &commandController, EventDistributor &eventDistributor)
This file implemented 3 utility functions:
Definition: Autofire.cc:9
int sock_recv(SOCKET sd, char *buf, size_t count)
Definition: Socket.cc:62
constexpr int OPENMSX_INVALID_SOCKET
Definition: Socket.hh:20
void sock_close(SOCKET sd)
Definition: Socket.cc:52
int sock_send(SOCKET sd, const char *buf, size_t count)
Definition: Socket.cc:85
EventType getType(const Event &event)
Definition: Event.hh:645
constexpr int BUF_SIZE
int SOCKET
Definition: Socket.hh:22
void fill(ForwardRange &&range, const T &value)
Definition: ranges.hh:227
TemporaryString tmpStrCat(Ts &&... ts)
Definition: strCat.hh:659
std::string strCat(Ts &&...ts)
Definition: strCat.hh:591
void strAppend(std::string &result, Ts &&...ts)
Definition: strCat.hh:669
constexpr auto end(const zstring_view &x)
Definition: zstring_view.hh:84