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

Last change on this file since 10796 was 10780, checked in by tbretz, 9 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.