Subversion Repositories HomeAutomation

Rev

Rev 2107 | Only display areas with differences | Regard whitespace | Details | Blame | Last modification | View Log | SVN | RSS feed

Rev 2107 Rev 2216
1
/*
1
/*
2
 *
2
 *
3
 *  Copyright(C) 2010  Mattias Runge
3
 *  Copyright(C) 2010  Mattias Runge
4
 *
4
 *
5
 *  This program is free software; you can redistribute it and/or modify
5
 *  This program is free software; you can redistribute it and/or modify
6
 *  it under the terms of the GNU General Public License as published by
6
 *  it under the terms of the GNU General Public License as published by
7
 *  the Free Software Foundation; either version 2 of the License, or
7
 *  the Free Software Foundation; either version 2 of the License, or
8
 * (at your option) any later version.
8
 * (at your option) any later version.
9
 *
9
 *
10
 *  This program is distributed in the hope that it will be useful,
10
 *  This program is distributed in the hope that it will be useful,
11
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
11
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13
 *  GNU General Public License for more details.
13
 *  GNU General Public License for more details.
14
 *
14
 *
15
 *  You should have received a copy of the GNU General Public License along
15
 *  You should have received a copy of the GNU General Public License along
16
 *  with this program; if not, write to the Free Software Foundation, Inc.,
16
 *  with this program; if not, write to the Free Software Foundation, Inc.,
17
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17
 *  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18
 *
18
 *
19
 */
19
 */
20
 
20
 
21
#include <signal.h>
21
#include <signal.h>
22
#include <sys/resource.h>
22
#include <sys/resource.h>
23
 
23
 
24
#include <boost/program_options.hpp>
24
#include <boost/program_options.hpp>
25
#include <boost/thread/mutex.hpp>
25
#include <boost/thread/mutex.hpp>
26
#include <boost/thread/condition.hpp>
26
#include <boost/thread/condition.hpp>
27
#include <boost/thread/locks.hpp>
27
#include <boost/thread/locks.hpp>
28
 
28
 
29
#include "config.h"
29
#include "config.h"
30
 
30
 
31
#include "logging/Logger.h" // Deprecated
31
#include "logging/Logger.h" // Deprecated
32
#include "common/log.h"
32
#include "common/log.h"
33
#include "net/Manager.h"
33
#include "net/Manager.h"
34
#include "timer/Manager.h"
34
#include "timer/Manager.h"
35
#include "config/Manager.h"
35
#include "config/Manager.h"
36
#include "can/Protocol.h"
36
#include "can/Protocol.h"
37
#include "broker/Manager.h"
37
#include "broker/Manager.h"
38
#include "control/Manager.h"
38
#include "control/Manager.h"
39
#include "can/Network.h"
39
#include "can/Network.h"
40
#include "can/Monitor.h"
40
#include "can/Monitor.h"
41
#include "can/CanDaemon.h"
41
#include "can/CanDaemon.h"
-
 
42
#include "can/Forward.h"
42
#include "vm/Manager.h"
43
#include "vm/Manager.h"
43
#include "storage/Manager.h"
44
#include "storage/Manager.h"
44
#include "net/GetFile.h"
45
#include "net/GetFile.h"
45
#include "net/Url.h"
46
#include "net/Url.h"
46
 
47
 
47
#include "vm/plugin/System.h"
48
#include "vm/plugin/System.h"
48
#include "vm/plugin/Timer.h"
49
#include "vm/plugin/Timer.h"
49
#include "vm/plugin/Module.h"
50
#include "vm/plugin/Module.h"
50
#include "vm/plugin/Node.h"
51
#include "vm/plugin/Node.h"
51
#include "vm/plugin/Console.h"
52
#include "vm/plugin/Console.h"
52
#include "vm/plugin/Storage.h"
53
#include "vm/plugin/Storage.h"
53
#include "vm/plugin/Socket.h"
54
#include "vm/plugin/Socket.h"
54
#include "vm/plugin/JsPipe.h"
55
#include "vm/plugin/JsPipe.h"
55
 
56
 
56
#ifdef USE_PLUGIN_XORG
57
#ifdef USE_PLUGIN_XORG
57
#include "vm/plugin/Xorg.h"
58
#include "vm/plugin/Xorg.h"
58
#endif // USE_PLUGIN_XORG
59
#endif // USE_PLUGIN_XORG
59
 
60
 
60
#ifdef USE_PLUGIN_MYSQL
61
#ifdef USE_PLUGIN_MYSQL
61
#include "vm/plugin/MySql.h"
62
#include "vm/plugin/MySql.h"
62
#endif // USE_PLUGIN_MYSQL
63
#endif // USE_PLUGIN_MYSQL
63
 
64
 
64
using namespace atom;
65
using namespace atom;
65
 
66
 
66
logging::Logger LOG("Main"); // Deprecated
67
logging::Logger LOG("Main"); // Deprecated
67
static const std::string log_module_ = "main";
68
static const std::string log_module_ = "main";
68
std::vector<broker::Subscriber::Pointer> subscribers;
69
std::vector<broker::Subscriber::Pointer> subscribers;
69
boost::condition on_message_condition;
70
boost::condition on_message_condition;
70
boost::mutex guard_mutex;
71
boost::mutex guard_mutex;
71
net::GetFile::Pointer downloader;
72
net::GetFile::Pointer downloader;
72
 
73
 
73
void ContinueInitialization(std::string protocol_filename);
74
void ContinueInitialization(std::string protocol_filename);
74
void Handler(int status);
75
void Handler(int status);
75
void CleanUp();
76
void CleanUp();
76
 
77
 
77
 
78
 
78
 
79
 
79
 
80
 
80
void HandleProtocolFileDownload(bool success, net::Url url, std::string temporary_filepath)
81
void HandleProtocolFileDownload(bool success, net::Url url, std::string temporary_filepath)
81
{
82
{
82
  if (success)
83
  if (success)
83
  {
84
  {
84
    log::Info(log_module_, "Successfully downloaded protocol file!");
85
    log::Info(log_module_, "Successfully downloaded protocol file!");
85
    ContinueInitialization(temporary_filepath);
86
    ContinueInitialization(temporary_filepath);
86
  }
87
  }
87
  else
88
  else
88
  {
89
  {
89
    ContinueInitialization(url.GetRaw());
90
    ContinueInitialization(url.GetRaw());
90
  }
91
  }
91
}
92
}
92
 
93
 
93
int main(int argc, char **argv)
94
int main(int argc, char **argv)
94
{
95
{
95
  /* Set core dump parameters */
96
  /* Set core dump parameters */
96
  rlimit core_limit = { RLIM_INFINITY, RLIM_INFINITY };
97
  rlimit core_limit = { RLIM_INFINITY, RLIM_INFINITY };
97
  setrlimit( RLIMIT_CORE, &core_limit ); // enable core dumps
98
  setrlimit( RLIMIT_CORE, &core_limit ); // enable core dumps
98
 
99
 
99
 
100
 
100
  /* Print startup information */
101
  /* Print startup information */
101
  log::Info(log_module_, "Atom daemon, version %s starting...", VERSION);
102
  log::Info(log_module_, "Atom daemon, version %s starting...", VERSION);
102
  log::Info(log_module_, "Released under %s.", LICENSE);
103
  log::Info(log_module_, "Released under %s.", LICENSE);
103
  log::Info(log_module_, "Written by Mattias Runge 2010-2012.");
104
  log::Info(log_module_, "Written by Mattias Runge 2010-2012.");
104
 
105
 
105
 
106
 
106
  /* Register signal handlers */
107
  /* Register signal handlers */
107
  signal(SIGTERM, Handler);
108
  signal(SIGTERM, Handler);
108
  signal(SIGINT,  Handler);
109
  signal(SIGINT,  Handler);
109
  signal(SIGQUIT, Handler);
110
  signal(SIGQUIT, Handler);
110
  signal(SIGABRT, Handler);
111
  signal(SIGABRT, Handler);
111
  signal(SIGPIPE, Handler);
112
  signal(SIGPIPE, Handler);
112
 
113
 
113
 
114
 
114
  /* Check configuration */
115
  /* Check configuration */
115
  config::Manager::Create();
116
  config::Manager::Create();
116
 
117
 
117
  if (!config::Manager::Instance()->Set(argc, argv))
118
  if (!config::Manager::Instance()->Set(argc, argv))
118
  {
119
  {
119
    CleanUp();
120
    CleanUp();
120
    return EXIT_SUCCESS;
121
    return EXIT_SUCCESS;
121
  }
122
  }
122
 
123
 
123
 
124
 
124
  /* Set up logging */
125
  /* Set up logging */
125
  if (config::Manager::Instance()->Exist("LogFile"))
126
  if (config::Manager::Instance()->Exist("LogFile"))
126
  {
127
  {
127
    log::OpenFile(config::Manager::Instance()->GetAsString("LogFile"));
128
    log::OpenFile(config::Manager::Instance()->GetAsString("LogFile"));
128
  }
129
  }
129
 
130
 
130
  if (config::Manager::Instance()->Exist("LogLevelMask"))
131
  if (config::Manager::Instance()->Exist("LogLevelMask"))
131
  {
132
  {
132
    log::SetLevelByString(config::Manager::Instance()->GetAsString("LogLevelMask"));
133
    log::SetLevelByString(config::Manager::Instance()->GetAsString("LogLevelMask"));
133
  }
134
  }
134
 
135
 
135
 
136
 
136
  /* Enter daemon mode if requested */
137
  /* Enter daemon mode if requested */
137
  if (config::Manager::Instance()->Exist("daemon"))
138
  if (config::Manager::Instance()->Exist("daemon"))
138
  {
139
  {
139
    LOG.Info("Entering daemon mode...");
140
    LOG.Info("Entering daemon mode...");
140
 
141
 
141
    if (daemon(0, 0) == -1)
142
    if (daemon(0, 0) == -1)
142
    {
143
    {
143
      log::Error(log_module_, "Could not enter daemon mode. Exiting...");
144
      log::Error(log_module_, "Could not enter daemon mode. Exiting...");
144
      CleanUp();
145
      CleanUp();
145
      return EXIT_FAILURE;
146
      return EXIT_FAILURE;
146
    }
147
    }
147
 
148
 
148
    log::Info(log_module_, "Deamon mode entered successfully!");
149
    log::Info(log_module_, "Deamon mode entered successfully!");
149
  }
150
  }
150
 
151
 
151
 
152
 
152
  /* Create singletons */
153
  /* Create singletons */
153
  storage::Manager::Create();
154
  storage::Manager::Create();
154
  timer::Manager::Create();
155
  timer::Manager::Create();
155
  broker::Manager::Create();
156
  broker::Manager::Create();
156
  net::Manager::Create();
157
  net::Manager::Create();
157
  can::Protocol::Create();
158
  can::Protocol::Create();
158
  control::Manager::Create();
159
  control::Manager::Create();
159
  vm::Manager::Create();
160
  vm::Manager::Create();
160
 
161
 
161
 
162
 
162
 
163
 
163
  std::string protocol_filename = config::Manager::Instance()->GetAsString("ProtocolFile");
164
  std::string protocol_filename = config::Manager::Instance()->GetAsString("ProtocolFile");
164
 
165
 
165
  try
166
  try
166
  {
167
  {
167
    net::Url url(protocol_filename);
168
    net::Url url(protocol_filename);
168
   
169
 
169
    downloader = net::GetFile::Pointer(new net::GetFile());
170
    downloader = net::GetFile::Pointer(new net::GetFile());
170
   
171
 
171
    log::Info(log_module_, "Will try to download protocol file from \"%s\"...", protocol_filename.data());
172
    log::Info(log_module_, "Will try to download protocol file from \"%s\"...", protocol_filename.data());
172
   
173
 
173
    downloader->Start(url, HandleProtocolFileDownload);
174
    downloader->Start(url, HandleProtocolFileDownload);
174
  }
175
  }
175
  catch (std::exception& exception)
176
  catch (std::exception& exception)
176
  {
177
  {
177
    log::Debug(log_module_, "Could not parse ProtocolFile as a URL, \"%s\".", protocol_filename.data());
178
    log::Debug(log_module_, "Could not parse ProtocolFile as a URL, \"%s\".", protocol_filename.data());
178
    ContinueInitialization(protocol_filename);
179
    ContinueInitialization(protocol_filename);
179
  }
180
  }
180
 
181
 
181
 
182
 
182
  /* Lock main thread and wait for exit */
183
  /* Lock main thread and wait for exit */
183
  boost::mutex::scoped_lock guard(guard_mutex);
184
  boost::mutex::scoped_lock guard(guard_mutex);
184
 
185
 
185
  on_message_condition.wait(guard);
186
  on_message_condition.wait(guard);
186
 
187
 
187
 
188
 
188
  /* Cleanup */
189
  /* Cleanup */
189
  log::Info(log_module_, "Cleaning up...");
190
  log::Info(log_module_, "Cleaning up...");
190
 
191
 
191
  CleanUp();
192
  CleanUp();
192
 
193
 
193
  log::Info(log_module_, "Thank you for using Atom. Goodbye!");
194
  log::Info(log_module_, "Thank you for using Atom. Goodbye!");
194
 
195
 
195
  return EXIT_SUCCESS;
196
  return EXIT_SUCCESS;
196
}
197
}
197
 
198
 
198
void ContinueInitialization(std::string protocol_filename)
199
void ContinueInitialization(std::string protocol_filename)
199
{  
200
{
200
  /* Load protocol file */
201
  /* Load protocol file */
201
  if (!can::Protocol::Instance()->Load(protocol_filename))
202
  if (!can::Protocol::Instance()->Load(protocol_filename))
202
  {
203
  {
203
    log::Error(log_module_, "Failed to load %s!", protocol_filename.data());
204
    log::Error(log_module_, "Failed to load %s!", protocol_filename.data());
204
 
205
 
205
    on_message_condition.notify_all();
206
    on_message_condition.notify_all();
206
  }
207
  }
207
 
208
 
208
 
209
 
209
  /* Set up storage path */
210
  /* Set up storage path */
210
  storage::Manager::Instance()->SetRootPath(config::Manager::Instance()->GetAsString("StoragePath"));
211
  storage::Manager::Instance()->SetRootPath(config::Manager::Instance()->GetAsString("StoragePath"));
211
 
212
 
212
 
213
 
213
  /* Start monitor server */
214
  /* Start monitor server */
214
  if (config::Manager::Instance()->Exist("MonitorPort"))
215
  if (config::Manager::Instance()->Exist("MonitorPort"))
215
  {
216
  {
216
    subscribers.push_back(can::Monitor::Pointer(new can::Monitor(config::Manager::Instance()->GetAsInt("MonitorPort"))));
217
    subscribers.push_back(can::Monitor::Pointer(new can::Monitor(config::Manager::Instance()->GetAsInt("MonitorPort"))));
217
  }
218
  }
218
 
219
 
219
 
220
 
220
  /* Start CanDaemon server */
221
  /* Start CanDaemon server */
221
  if (config::Manager::Instance()->Exist("DaemonPort"))
222
  if (config::Manager::Instance()->Exist("DaemonPort"))
222
  {
223
  {
223
    subscribers.push_back(can::CanDaemon::Pointer(new can::CanDaemon(config::Manager::Instance()->GetAsInt("DaemonPort"))));
224
    subscribers.push_back(can::CanDaemon::Pointer(new can::CanDaemon(config::Manager::Instance()->GetAsInt("DaemonPort"))));
224
  }
225
  }
-
 
226
 
-
 
227
    /* Start CAN forward server */
-
 
228
  if (config::Manager::Instance()->Exist("CanForwardPort"))
-
 
229
  {
-
 
230
    subscribers.push_back(can::Forward::Pointer(new can::Forward(config::Manager::Instance()->GetAsInt("CanForwardPort"))));
-
 
231
  }
-
 
232
 
225
 
233
 
226
 
234
 
227
  /* Subscribe Can control manager to data */
235
  /* Subscribe Can control manager to data */
228
  subscribers.push_back(control::Manager::Instance());
236
  subscribers.push_back(control::Manager::Instance());
229
 
237
 
230
 
238
 
231
  /* Start VM plugins */
239
  /* Start VM plugins */
232
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::System(vm::Manager::Instance()->GetIoService())));
240
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::System(vm::Manager::Instance()->GetIoService())));
233
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Console(vm::Manager::Instance()->GetIoService(), config::Manager::Instance()->GetAsInt("CommandPort"))));
241
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Console(vm::Manager::Instance()->GetIoService(), config::Manager::Instance()->GetAsInt("CommandPort"))));
234
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::JsPipe(vm::Manager::Instance()->GetIoService(), config::Manager::Instance()->GetAsInt("JsPipePort"))));
242
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::JsPipe(vm::Manager::Instance()->GetIoService(), config::Manager::Instance()->GetAsInt("JsPipePort"))));
235
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Storage(vm::Manager::Instance()->GetIoService())));
243
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Storage(vm::Manager::Instance()->GetIoService())));
236
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Timer(vm::Manager::Instance()->GetIoService())));
244
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Timer(vm::Manager::Instance()->GetIoService())));
237
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Module(vm::Manager::Instance()->GetIoService())));
245
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Module(vm::Manager::Instance()->GetIoService())));
238
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Node(vm::Manager::Instance()->GetIoService())));
246
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Node(vm::Manager::Instance()->GetIoService())));
239
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Socket(vm::Manager::Instance()->GetIoService())));
247
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Socket(vm::Manager::Instance()->GetIoService())));
240
 
248
 
241
#ifdef USE_PLUGIN_XORG
249
#ifdef USE_PLUGIN_XORG
242
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Xorg(vm::Manager::Instance()->GetIoService())));
250
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::Xorg(vm::Manager::Instance()->GetIoService())));
243
#endif // USE_PLUGIN_XORG
251
#endif // USE_PLUGIN_XORG
244
 
252
 
245
#ifdef USE_PLUGIN_MYSQL
253
#ifdef USE_PLUGIN_MYSQL
246
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::MySql(vm::Manager::Instance()->GetIoService())));
254
  vm::Manager::Instance()->AddPlugin(vm::Plugin::Pointer(new vm::plugin::MySql(vm::Manager::Instance()->GetIoService())));
247
#endif // USE_PLUGIN_MYSQL
255
#endif // USE_PLUGIN_MYSQL
248
 
256
 
249
 
257
 
250
  /* Start VM */
258
  /* Start VM */
251
  vm::Manager::Instance()->Start(config::Manager::Instance()->GetAsString("ScriptPath"), config::Manager::Instance()->GetAsString("UserScriptPath"));
259
  vm::Manager::Instance()->Start(config::Manager::Instance()->GetAsString("ScriptPath"), config::Manager::Instance()->GetAsString("UserScriptPath"));
252
 
260
 
253
 
261
 
254
  /* Connect to Can network */
262
  /* Connect to Can network */
255
  if (config::Manager::Instance()->Exist("CanNet"))
263
  if (config::Manager::Instance()->Exist("CanNet"))
256
  {
264
  {
257
    common::StringList cannetworks = config::Manager::Instance()->GetAsStringVector("CanNet");
265
    common::StringList cannetworks = config::Manager::Instance()->GetAsStringVector("CanNet");
258
 
266
 
259
    for (uint n = 0; n < cannetworks.size(); n++)
267
    for (uint n = 0; n < cannetworks.size(); n++)
260
    {
268
    {
261
      subscribers.push_back(can::Network::Pointer(new can::Network(cannetworks[n])));
269
      subscribers.push_back(can::Network::Pointer(new can::Network(cannetworks[n])));
262
    }
270
    }
263
  }
271
  }
264
}
272
}
265
 
273
 
266
void CleanUp()
274
void CleanUp()
267
{
275
{
268
  downloader.reset();
276
  downloader.reset();
269
 
277
 
270
  if (net::Manager::Instance().use_count() > 0)
278
  if (net::Manager::Instance().use_count() > 0)
271
  {
279
  {
272
    net::Manager::Instance()->Stop();
280
    net::Manager::Instance()->Stop();
273
  }
281
  }
274
 
282
 
275
  subscribers.clear();
283
  subscribers.clear();
276
  config::Manager::Delete();
284
  config::Manager::Delete();
277
  timer::Manager::Delete();
285
  timer::Manager::Delete();
278
  control::Manager::Delete();
286
  control::Manager::Delete();
279
  can::Protocol::Delete();
287
  can::Protocol::Delete();
280
  broker::Manager::Delete();
288
  broker::Manager::Delete();
281
  vm::Manager::Delete();
289
  vm::Manager::Delete();
282
  net::Manager::Delete();
290
  net::Manager::Delete();
283
  storage::Manager::Delete();
291
  storage::Manager::Delete();
284
 
292
 
285
  log::CloseFile();
293
  log::CloseFile();
286
}
294
}
287
 
295
 
288
void Handler(int status)
296
void Handler(int status)
289
{
297
{
290
  std::string signal_name = "Unknown";
298
  std::string signal_name = "Unknown";
291
 
299
 
292
  switch (status)
300
  switch (status)
293
  {
301
  {
294
    case SIGTERM:
302
    case SIGTERM:
295
    {
303
    {
296
      signal_name = "Terminate";
304
      signal_name = "Terminate";
297
      break;
305
      break;
298
    }
306
    }
299
 
307
 
300
    case SIGINT:
308
    case SIGINT:
301
    {
309
    {
302
      signal_name = "Interrupt";
310
      signal_name = "Interrupt";
303
      break;
311
      break;
304
    }
312
    }
305
 
313
 
306
    case SIGQUIT:
314
    case SIGQUIT:
307
    {
315
    {
308
      signal_name = "Quit";
316
      signal_name = "Quit";
309
      break;
317
      break;
310
    }
318
    }
311
 
319
 
312
    case SIGABRT:
320
    case SIGABRT:
313
    {
321
    {
314
      signal_name = "Abort";
322
      signal_name = "Abort";
315
      break;
323
      break;
316
    }
324
    }
317
 
325
 
318
    case SIGIO:
326
    case SIGIO:
319
    {
327
    {
320
      signal_name = "I/O";
328
      signal_name = "I/O";
321
      break;
329
      break;
322
    }
330
    }
323
 
331
 
324
    case SIGPIPE:
332
    case SIGPIPE:
325
    {
333
    {
326
      signal_name = "Pipe";
334
      signal_name = "Pipe";
327
      break;
335
      break;
328
    }
336
    }
329
  }
337
  }
330
 
338
 
331
  log::Info(log_module_, "Received signal %s (%d).", signal_name.c_str(), status);
339
  log::Info(log_module_, "Received signal %s (%d).", signal_name.c_str(), status);
332
 
340
 
333
  if (status != SIGPIPE)
341
  if (status != SIGPIPE)
334
  {
342
  {
335
    on_message_condition.notify_all();
343
    on_message_condition.notify_all();
336
  }
344
  }
337
}
345
}
338
 
346