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

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