source: trunk/FACT++/src/dclient5.cc@ 19285

Last change on this file since 19285 was 10780, checked in by tbretz, 14 years ago
Replaced stringstream in log-stream by ostringstream.
File size: 21.0 KB
Line 
1#include <boost/bind.hpp>
2#if BOOST_VERSION < 104400
3#if (__GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ > 4))
4#undef BOOST_HAS_RVALUE_REFS
5#endif
6#endif
7#include <boost/thread.hpp>
8#include <boost/asio/error.hpp>
9#include <boost/asio/deadline_timer.hpp>
10
11#include "Event.h"
12#include "Shell.h"
13#include "StateMachineDim.h"
14#include "Connection.h"
15#include "Configuration.h"
16#include "Timers.h"
17#include "Console.h"
18
19#include "tools.h"
20
21namespace ba = boost::asio;
22namespace bs = boost::system;
23
24using ba::deadline_timer;
25using ba::ip::tcp;
26
27using namespace std;
28
29
30// ------------------------------------------------------------------------
31
32#include "LocalControl.h"
33
34// ------------------------------------------------------------------------
35
36class ConnectionFAD : public Connection
37{
38 MessageImp &fMsg;
39
40 int state;
41
42 char fReadBuffer[1000];
43
44public:
45 void ConnectionEstablished()
46 {
47 StartAsyncRead();
48 }
49
50 void HandleReadTimeout(const bs::error_code &error)
51 {
52 return;
53 if (!is_open())
54 {
55 // For example: Here we could schedule a new accept if we
56 // would not want to allow two connections at the same time.
57 return;
58 }
59
60 // 125: Operation canceled
61
62 if (error && error!=ba::error::basic_errors::operation_aborted)
63 {
64 ostringstream str;
65
66 str << "HandleReadTimeout: " << error.message() << " (" << error << ")";// << endl;
67 if (error==ba::error::misc_errors::eof)
68 Warn(str); // Connection: EOF (closed by remote host)
69 else
70 Error(str);
71 }
72
73 // Check whether the deadline has passed. We compare the deadline
74 // against the current time since a new asynchronous operation
75 // may have moved the deadline before this actor had a chance
76 // to run.
77 if (fInTimeout.expires_at() > deadline_timer::traits_type::now())
78 return;
79
80 Error("fInTimeout has expired...");
81
82 PostClose();
83 }
84
85 void HandleReceivedData(const bs::error_code& error, size_t bytes_received, int)
86 {
87 // Do not schedule a new read if the connection failed.
88 if (bytes_received==0 || error)
89 {
90 // 107: Transport endpoint is not connected
91 // 125: Operation canceled
92 if (error && error!=ba::error::basic_errors::not_connected)
93 {
94 ostringstream str;
95 str << "Reading from " << URL() << ": " << error.message() << " (" << error << ")";// << endl;
96 Error(str);
97 }
98 PostClose(error!=ba::error::basic_errors::operation_aborted);
99 return;
100 }
101
102 string txt;
103
104 if (bytes_received==2)
105 {
106 txt = string(fReadBuffer, bytes_received);
107 //std::vector<char> buf(128);
108 //bytes_transferred = sock.receive(boost::asio::buffer(d3));
109
110 fMsg() << "Received b=" << bytes_received << ": " << (int)fReadBuffer[0] << " " << (int)txt[0] << " '" << txt << "' " << " " << error.message() << " (" << error << ")" << endl;
111
112 if (fReadBuffer[0]=='T')
113 {
114 // AsyncRead + Deadline
115 // Do all manipulation to the buffer BEFORE this call!
116 AsyncRead(ba::buffer(fReadBuffer+2, 21)/*,
117 &Connection::HandleReceivedData*/);
118 AsyncWait(fInTimeout, 5000, &Connection::HandleReadTimeout);
119 }
120 else
121 {
122 // AsyncRead + Deadline
123 // Do all manipulation to the buffer BEFORE this call!
124 AsyncRead(ba::buffer(fReadBuffer+2, 35)/*,
125 &Connection::HandleReceivedData*/);
126 AsyncWait(fInTimeout, 5000, &Connection::HandleReadTimeout);
127 }
128 }
129 else
130 {
131 txt = string(fReadBuffer, bytes_received+2);
132 const int s = atoi(fReadBuffer+35);
133 if (s==9)
134 Info("Requested time received: "+txt);
135 else
136 state = s;
137
138 Out() << "Received b=" << bytes_received << ": " << (int)fReadBuffer[0] << " " << (int)txt[0] << " '" << txt << "' " << " " << error.message() << " (" << error << ")" << endl;
139 memset(fReadBuffer, 0, 100);
140
141 // Do all manipulation to the buffer BEFORE this call!
142 AsyncRead(ba::buffer(fReadBuffer, 2)/*,
143 &Connection::HandleReceivedData*/);
144
145
146 }
147 }
148
149 int GetState() const { return state; }
150
151 void StartAsyncRead()
152 {
153 // Start also a dealine_time for a proper timeout
154 // Therefore we must know how often we expect messages
155 // FIXME: Add deadline_counter
156
157 memset(fReadBuffer, 0, 100);
158
159 // AsyncRead + Deadline
160 AsyncRead(ba::buffer(fReadBuffer, 2)/*,
161 &Connection::HandleReceivedData*/);
162 AsyncWait(fInTimeout, 5000, &Connection::HandleReadTimeout);
163 }
164
165 /*
166 ConnectionFAD(ba::io_service& io_service, const string &addr, int port) :
167 Connection(io_service, addr, port), state(0) { }
168 ConnectionFAD(ba::io_service& io_service, const string &addr, const string &port) :
169 Connection(io_service, addr, port), state(0) { }
170 */
171
172 ConnectionFAD(ba::io_service& ioservice, MessageImp &imp) :
173 Connection(ioservice, imp()), fMsg(imp), state(0)
174 {
175 }
176};
177
178template <class T>
179class StateMachineFAD : public T, public ba::io_service
180{
181public:
182 enum states_t
183 {
184 kSM_Disconnected = 1,
185 kSM_Connecting,
186 kSM_Connected,
187 kSM_Running,
188 kSM_SomeRunning,
189 kSM_Starting,
190 kSM_Stopping,
191 kSM_Reconnect,
192 kSM_SetUrl,
193 };
194
195 ConnectionFAD c1;
196 ConnectionFAD c2;
197 ConnectionFAD c3;
198 ConnectionFAD c4;
199 ConnectionFAD c5;
200 ConnectionFAD c6;
201 ConnectionFAD c7;
202 ConnectionFAD c8;
203 ConnectionFAD c9;
204
205 /*
206 int Write(const Time &time, const char *txt, int qos)
207 {
208 return T::Write(time, txt, qos);
209 }
210 */
211 Timers fTimers;
212
213 StateMachineFAD(const string &name="", ostream &out=cout) :
214 T(out, name),
215 c1(*this, *this), c2(*this, *this), c3(*this, *this), c4(*this, *this),
216 c5(*this, *this), c6(*this, *this), c7(*this, *this), c8(*this, *this),
217 c9(*this, *this), fTimers(out)
218 {
219// c1.SetEndpoint();
220 c2.SetEndpoint("localhost", 4001);
221 c3.SetEndpoint("ftmboard1.ethz.ch", 5000);
222 c4.SetEndpoint("localhost", 4003);
223 c5.SetEndpoint("localhost", 4004);
224 c6.SetEndpoint("localhost", 4005);
225 c7.SetEndpoint("localhost", 4006);
226 c8.SetEndpoint("localhost", 4007);
227 c9.SetEndpoint("localhost", 4008);
228
229 c1.SetLogStream(this);
230 c2.SetLogStream(this);
231 c3.SetLogStream(this);
232 c4.SetLogStream(this);
233 c5.SetLogStream(this);
234 c6.SetLogStream(this);
235 c7.SetLogStream(this);
236 c8.SetLogStream(this);
237 c9.SetLogStream(this);
238
239 c1.StartConnect(); // This sets the connection to "open"
240 c2.StartConnect(); // This sets the connection to "open"
241 c3.StartConnect(); // This sets the connection to "open"
242 //c4.StartConnect(); // This sets the connection to "open"
243 //c5.StartConnect(); // This sets the connection to "open"
244 //c6.StartConnect(); // This sets the connection to "open"
245 //c7.StartConnect(); // This sets the connection to "open"
246 //c8.StartConnect(); // This sets the connection to "open"
247 //c9.StartConnect(); // This sets the connection to "open"
248
249 AddStateName(kSM_Disconnected, "Disconnected");
250 AddStateName(kSM_Connecting, "Connecting"); // Some connected
251 AddStateName(kSM_Connected, "Connected");
252 AddStateName(kSM_Running, "Running");
253 AddStateName(kSM_SomeRunning, "SomeRunning");
254 AddStateName(kSM_Starting, "Starting");
255 AddStateName(kSM_Stopping, "Stopping");
256
257 AddEvent(kSM_Running, "START", kSM_Connected).
258 AssignFunction(boost::bind(&StateMachineFAD::Start, this, _1, 5));
259 AddEvent(kSM_Connected, "STOP", kSM_Running);
260
261 AddEvent("TIME", kSM_Running);
262 AddEvent("LED", kSM_Connected);
263
264 T::AddEvent("TESTI", "I");
265 T::AddEvent("TESTI2", "I:2");
266 T::AddEvent("TESTIF", "I:2;F:2");
267 T::AddEvent("TESTIC", "I:2;C");
268
269 T::AddEvent("CMD", "C").
270 AssignFunction(boost::bind(&StateMachineFAD::Command, this, _1));
271
272 AddEvent(kSM_Reconnect, "RECONNECT");
273
274 AddEvent(kSM_SetUrl, "SETURL", "C");
275 }
276
277 int Command(const EventImp &evt)
278 {
279 string cmd = evt.GetText();
280
281 size_t p0 = cmd.find_first_of(' ');
282 if (p0==string::npos)
283 p0 = cmd.length();
284
285 T::Out() << "\nCommand: '" << cmd.substr(0, p0) << "'" << cmd.substr(p0)<< "'" << endl;
286 /*
287 const Converter c(T::Out(), "B:5;I:2;F;W;O;C", "yes no false 0 1 31 42 11.12 \"test hallo\" ");
288
289 T::Out() << c.GetRc() << endl;
290 T::Out() << c.N() << endl;
291 T::Out() << c.Get<bool>(0) << endl;
292 T::Out() << c.Get<bool>(1) << endl;
293 T::Out() << c.Get<bool>(2) << endl;
294 T::Out() << c.Get<bool>(3) << endl;
295 T::Out() << c.Get<bool>(4) << endl;
296 T::Out() << c.Get<int>(5) << endl;
297 T::Out() << c.Get<int>(6) << endl;
298 T::Out() << c.Get<float>(7) << endl;
299 T::Out() << c.Get<int>(7) << endl;
300 T::Out() << c.Get<string>(8) << endl;
301 T::Out() << c.Get<string>(9) << endl;
302 T::Out() << c.Get<string>(10) << endl;
303 */
304 return T::GetCurrentState();
305 }
306 int Start(const EventImp &evt, int i)
307 {
308 switch (evt.GetTargetState())
309 {
310 case kSM_Running: // We are coming from kRunning
311 case kSM_Starting: // We are coming from kConnected
312 T::Out() << "Received Start(" << i << ")" << endl;
313 c1.PostMessage("START", 10);
314 c2.PostMessage("START", 10);
315 // We could introduce a "waiting for execution" state
316 return T::GetCurrentState();
317 }
318 return T::kSM_FatalError;
319 }
320
321 void Close()
322 {
323 c1.PostClose();
324 c2.PostClose();
325 c3.PostClose();
326 c4.PostClose();
327 c5.PostClose();
328 c6.PostClose();
329 c7.PostClose();
330 c8.PostClose();
331 c9.PostClose();
332 }
333
334
335 int Execute()
336 {
337 // Dispatch at most one handler from the queue. In contrary
338 // to run_run(), it doesn't wait until a handler is available
339 // which can be dispatched, so poll_one() might return with 0
340 // handlers dispatched. The handlers are always dispatched
341 // synchronously.
342
343 fTimers.SetT();
344 const int n = poll_one();
345 fTimers.Proc(n==0 && T::IsQueueEmpty());
346
347// return c3.IsConnected() ? kSM_Connected : kSM_Disconnected;
348
349
350 // None is connected
351 if (!c1.IsConnected() && !c2.IsConnected())
352 return kSM_Disconnected;
353
354 // Some are connected
355 if (c1.IsConnected()!=c2.IsConnected())
356 return kSM_Connecting;
357
358 if (c1.GetState()==0 && c2.GetState()==0 && T::GetCurrentState()!=kSM_Starting)
359 return kSM_Connected;
360
361 if (c1.GetState()==1 && c2.GetState()==1 && T::GetCurrentState()!=kSM_Stopping)
362 return kSM_Running;
363
364 return kSM_SomeRunning;//GetCurrentState();
365 }
366
367 int Transition(const Event &evt)
368 {
369 ConnectionFAD *con1 = &c1;
370 ConnectionFAD *con2 = &c2;
371
372 switch (evt.GetTargetState())
373 {
374 case kSM_SetUrl:
375 T::Out() << evt.GetText() << endl;
376 c1.SetEndpoint(evt.GetText());
377 return T::GetCurrentState();
378 case kSM_Reconnect:
379 // Close all connections
380 c1.PostClose(false);
381 c2.PostClose(false);
382 c3.PostClose(false);
383
384 // Now wait until all connection have been closed and
385 // all pending handlers have been processed
386 poll();
387
388 // Now we can reopen the connection
389 c1.PostClose(true);
390 c2.PostClose(true);
391 c3.PostClose(true);
392
393
394 //c4.PostClose(true);
395 //c5.PostClose(true);
396 //c6.PostClose(true);
397 //c7.PostClose(true);
398 //c8.PostClose(true);
399 //c9.PostClose(true);
400 return T::GetCurrentState();
401 case kSM_Running: // We are coming from kRunning
402 case kSM_Starting: // We are coming from kConnected
403 T::Out() << "Received START" << endl;
404 con1->PostMessage("START", 10);
405 con2->PostMessage("START", 10);
406 // We could introduce a "waiting for execution" state
407 return T::GetCurrentState();
408 return kSM_Starting; //GetCurrentState();
409
410 case kSM_Connected: // We are coming from kConnected
411 case kSM_Stopping: // We are coming from kRunning
412 T::Out() << "Received STOP" << endl;
413 con1->PostMessage("STOP", 10);
414 con2->PostMessage("STOP", 10);
415 // We could introduce a "waiting for execution" state
416 return T::GetCurrentState();
417 return kSM_Stopping;//GetCurrentState();
418 }
419
420 return T::kSM_FatalError; //evt.GetTargetState();
421 }
422 int Configure(const Event &evt)
423 {
424 if (evt.GetName()=="TIME")
425 {
426 c1.PostMessage("TIME", 10);
427 c2.PostMessage("TIME", 10);
428 }
429
430 vector<char> v(2);
431 v[0] = 0xc0;
432 v[1] = 0x00;
433
434 if (evt.GetName()=="LED")
435 c3.PostMessage(v);
436
437 return T::GetCurrentState();
438 }
439};
440
441// ------------------------------------------------------------------------
442
443template<class S>
444int RunDim(Configuration &conf)
445{
446 /*
447 initscr(); // Start curses mode
448 cbreak(); // Line buffering disabled, Pass on
449 intrflush(stdscr, FALSE);
450 start_color(); // Initialize ncurses colors
451 use_default_colors(); // Assign terminal default colors to -1
452 for (int i=1; i<8; i++)
453 init_pair(i, i, -1); // -1: def background
454 scrollok(stdscr, true);
455 */
456
457 WindowLog wout;
458
459 //log.SetWindow(stdscr);
460 if (conf.Has("log"))
461 if (!wout.OpenLogFile(conf.Get<string>("log")))
462 wout << kRed << "ERROR - Couldn't open log-file " << conf.Get<string>("log") << ": " << strerror(errno) << endl;
463
464 // Start io_service.Run to use the StateMachineImp::Run() loop
465 // Start io_service.run to only use the commandHandler command detaching
466 StateMachineFAD<S> io_service("DATA_LOGGER", wout);
467 io_service.Run();
468
469 return 0;
470}
471
472template<class T, class S>
473int RunShell(Configuration &conf)
474{
475 static T shell(conf.GetName().c_str(), conf.Get<int>("console")!=1);
476
477 WindowLog &win = shell.GetStreamIn();
478 WindowLog &wout = shell.GetStreamOut();
479
480 if (conf.Has("log"))
481 if (!wout.OpenLogFile(conf.Get<string>("log")))
482 win << kRed << "ERROR - Couldn't open log-file " << conf.Get<string>("log") << ": " << strerror(errno) << endl;
483
484 StateMachineFAD<S> io_service("DATA_LOGGER", wout);
485 shell.SetReceiver(io_service);
486
487 boost::thread t(boost::bind(&StateMachineFAD<S>::Run, &io_service));
488
489 //io_service.SetReady();
490
491 shell.Run(); // Run the shell
492 io_service.Stop(); // Signal Loop-thread to stop
493 // io_service.Close(); // Obsolete, done by the destructor
494 // wout << "join: " << t.timed_join(boost::posix_time::milliseconds(0)) << endl;
495
496 // Wait until the StateMachine has finished its thread
497 // before returning and destroying the dim objects which might
498 // still be in use.
499 t.join();
500
501 return 0;
502}
503
504/*
505 Extract usage clause(s) [if any] for SYNOPSIS.
506 Translators: "Usage" and "or" here are patterns (regular expressions) which
507 are used to match the usage synopsis in program output. An example from cp
508 (GNU coreutils) which contains both strings:
509 Usage: cp [OPTION]... [-T] SOURCE DEST
510 or: cp [OPTION]... SOURCE... DIRECTORY
511 or: cp [OPTION]... -t DIRECTORY SOURCE...
512 */
513void PrintUsage()
514{
515 cout << "\n"
516 "The console connects to all available Dim Servers and allows to "
517 "easily access all of their commands.\n"
518 "\n"
519 "Usage: test3 [-c type] [OPTIONS]\n"
520 " or: test3 [OPTIONS]\n"
521 "\n"
522 "Options:\n"
523 "The following describes the available commandline options. "
524 "For further details on how command line option are parsed "
525 "and in which order which configuration sources are accessed "
526 "please refer to the class reference of the Configuration class.";
527 cout << endl;
528
529}
530
531void PrintHelp()
532{
533 cout << "\n"
534 "The default is that the program is started without user interaction. "
535 "All actions are supposed to arrive as DimCommands. Using the -c "
536 "option, a local shell can be initialized. With h or help a short "
537 "help message about the usuage can be brought to the screen."
538 << endl;
539
540 /*
541 cout << "bla bla bla" << endl << endl;
542 cout << endl;
543 cout << "Environment:" << endl;
544 cout << "environment" << endl;
545 cout << endl;
546 cout << "Examples:" << endl;
547 cout << "test exam" << endl;
548 cout << endl;
549 cout << "Files:" << endl;
550 cout << "files" << endl;
551 cout << endl;
552 */
553}
554
555/*
556 The first line of the --version information is assumed to be in one
557 of the following formats:
558
559 <version>
560 <program> <version>
561 {GNU,Free} <program> <version>
562 <program> ({GNU,Free} <package>) <version>
563 <program> - {GNU,Free} <package> <version>
564
565 and separated from any copyright/author details by a blank line.
566
567 Handle multi-line bug reporting sections of the form:
568
569 Report <program> bugs to <addr>
570 GNU <package> home page: <url>
571 ...
572*/
573void PrintVersion(const char *name)
574{
575 cout <<
576 name << " - "PACKAGE_STRING"\n"
577 "\n"
578 "Written by Thomas Bretz et al.\n"
579 "\n"
580 "Report bugs to <"PACKAGE_BUGREPORT">\n"
581 "Home page: "PACKAGE_URL"\n"
582 "\n"
583 "Copyright (C) 2011 by the FACT Collaboration.\n"
584 "This is free software; see the source for copying conditions.\n"
585 << endl;
586}
587
588
589void SetupConfiguration(Configuration &conf)
590{
591 const string n = conf.GetName()+".log";
592
593 po::options_description config("Program options");
594 config.add_options()
595 ("dns", var<string>("localhost"), "Dim nameserver host name (Overwites DIM_DNS_NODE environment variable)")
596 ("log,l", var<string>(n), "Write log-file")
597 ("no-dim,d", po_switch(), "Disable dim services")
598 ("console,c", var<int>(), "Use console (0=shell, 1=simple buffered, X=simple unbuffered)")
599 ;
600
601 conf.AddEnv("dns", "DIM_DNS_NODE");
602
603 conf.AddOptions(config);
604}
605
606int main(int argc, const char* argv[])
607{
608 Configuration conf(argv[0]);
609 conf.SetPrintUsage(PrintUsage);
610 SetupConfiguration(conf);
611
612 po::variables_map vm;
613 try
614 {
615 vm = conf.Parse(argc, argv);
616 }
617 catch (std::exception &e)
618 {
619#if BOOST_VERSION > 104000
620 po::multiple_occurrences *MO = dynamic_cast<po::multiple_occurrences*>(&e);
621 if (MO)
622 cout << "Error: " << e.what() << " of '" << MO->get_option_name() << "' option." << endl;
623 else
624#endif
625 cout << "Error: " << e.what() << endl;
626 cout << endl;
627
628 return -1;
629 }
630
631 if (conf.HasPrint())
632 return -1;
633
634 if (conf.HasVersion())
635 {
636 PrintVersion(argv[0]);
637 return -1;
638 }
639
640 if (conf.HasHelp())
641 {
642 PrintHelp();
643 return -1;
644 }
645
646 // To allow overwriting of DIM_DNS_NODE set 0 to 1
647 setenv("DIM_DNS_NODE", conf.Get<string>("dns").c_str(), 1);
648
649 try
650 {
651 // No console access at all
652 if (!conf.Has("console"))
653 {
654 if (conf.Get<bool>("no-dim"))
655 return RunDim<StateMachine>(conf);
656 else
657 return RunDim<StateMachineDim>(conf);
658 }
659 // Cosole access w/ and w/o Dim
660 if (conf.Get<bool>("no-dim"))
661 {
662 if (conf.Get<int>("console")==0)
663 return RunShell<LocalShell, StateMachine>(conf);
664 else
665 return RunShell<LocalConsole, StateMachine>(conf);
666 }
667 else
668 {
669 if (conf.Get<int>("console")==0)
670 return RunShell<LocalShell, StateMachineDim>(conf);
671 else
672 return RunShell<LocalConsole, StateMachineDim>(conf);
673 }
674 }
675 catch (std::exception& e)
676 {
677 std::cerr << "Exception: " << e.what() << "\n";
678 }
679
680 return 0;
681}
682
683/*
684class FADctrlDim : public StateMachineFAD<StateMachineDim>
685{
686public:
687FADctrlDim(const std::string &name="DATA_LOGGER", std::ostream &out=std::cout)
688: StateMachineFAD<StateMachineDim>(out, name) { }
689};
690
691 class FADctrlLocalShell : public StateMachineFAD<StateMachine>
692{
693public:
694 ostream &win;
695
696 FADctrlLocalShell(std::ostream &out, std::ostream &out2)
697 : StateMachineFAD<StateMachine>(out), win(out2) { }
698
699 FADctrlLocalShell(std::ostream &out=std::cout)
700 : StateMachineFAD<StateMachine>(out), win(out) { }
701
702};
703*/
Note: See TracBrowser for help on using the repository browser.