Changeset 10923

Timestamp:

06/07/11 22:18:46 (13 years ago)

Author:

tbretz

Message:

Adapted to latest version of code; first implementration of an interface to write files.

File:

• trunk/FACT++/src/EventBuilderWrapper.h (modified) (10 diffs)

trunk/FACT++/src/EventBuilderWrapper.h

@@ -16 +16 @@
 
 #include "EventBuilder.h"
+
+extern "C" {
+    extern void StartEvtBuild();
+    extern int CloseRunFile(uint32_t runId, uint32_t closeTime);
+}
 
 class EventBuilderWrapper
@@ -41 +46 @@
     MessageImp &fMsg;
 
+    DimDescribedService fDimFiles;
+
 public:
-    EventBuilderWrapper(MessageImp &msg) : fMsg(msg)
+    EventBuilderWrapper(MessageImp &msg) : fMsg(msg),
+        fDimFiles("FAD_CONTROL/FILES", "X:1", "")
     {
         if (This)
@@ -48 +56 @@
 
         This = this;
+
+        fFitsFormat = false;
     }
     ~EventBuilderWrapper()
@@ -56 +66 @@
         fThread.join();
         //fMsg.Info("EventBuilder stopped.");
-    }
-
-    void Update(const char *msg, int severity)
+
+        for (vector<DataFileImp*>::iterator it=fFiles.begin(); it!=fFiles.end(); it++)
+            delete *it;
+    }
+
+    void Update(ostringstream &msg, int severity)
     {
         fMsg.Update(msg, severity);
@@ -79 +92 @@
     }
 
-    void Start(const vector<string> &addr)
+    void Start(const vector<tcp::endpoint> &addr)
     {
         if (IsThreadRunning())
@@ -87 +100 @@
         }
 
+        int cnt = 0;
+        for (size_t i=0; i<40; i++)
+        {
+            if (addr[i]==tcp::endpoint())
+            {
+                g_port[i].sockDef = -1;
+                continue;
+            }
+
+            // -1:  if entry shell not be used
+            //  0:  event builder will connect but ignore the events
+            //  1:  event builder will connect and build events
+            g_port[i].sockDef                  = 1;
+
+            g_port[i].sockAddr.sin_family      = AF_INET;
+            g_port[i].sockAddr.sin_addr.s_addr = htonl(addr[i].address().to_v4().to_ulong());
+            g_port[i].sockAddr.sin_port        = htons(addr[i].port());
+
+            cnt++;
+        }
+
+//        g_maxBoards = cnt;
+        g_actBoards = cnt;
+
+        g_runStat   = kModeRun;
+
         fMsg.Message("Starting EventBuilder thread");
-
-        for (size_t i=0; i<addr.size(); i++)
-        {
-            memset(g_ip[i].addr, 0, sizeof(g_ip[i].addr));
-
-            const size_t pos = addr[i].find_first_of(':');
-
-            const string a = addr[i].substr(0, pos);
-            const string p = addr[i].substr(pos+1);
-
-            strcpy(g_ip[i].addr, "127.0.0.1");
-            g_ip[i].port = atoi(p.c_str());
-        }
-
-        g_maxBoards = addr.size();
-        g_actBoards = addr.size();
-
-        g_runStat   = kModeRun;
 
         fThread = boost::thread(StartEvtBuild);
@@ -144 +165 @@
     size_t GetUsedMemory() const { return gi_usedMem; }
 
+    /*
+     struct OpenFileToDim
+     {
+        int code;
+        char fileName[FILENAME_MAX];
+     };
+
+     SignalRunOpened(runid, filename);
+     // Send num open files
+     // Send runid, (more info about the run?), filename via dim
+
+     SignalEvtWritten(runid);
+     // Send num events written of newest file
+
+     SignalRunClose(runid);
+     // Send new num open files
+     // Send empty file-name if no file is open
+
+     */
 
     // -------------- Mapped event builder callbacks ------------------
 
-    int runOpen(uint32_t runid, RUN_HEAD *h, size_t)
-    {
+    class DataFileImp
+    {
+        uint32_t fRunId;
+
+    public:
+        DataFileImp(uint32_t id) : fRunId(id) { }
+
+        virtual bool Write() = 0;
+        virtual bool Close() = 0;
+
+        uint32_t GetRunId() const { return fRunId; }
+    };
+
+
+    class DataFileRaw : public DataFileImp
+    {
+    public:
+        DataFileRaw(uint32_t id) : DataFileImp(id)  { }
+        ~DataFileRaw() { Close(); }
+
+        virtual bool Write() { return true; }
+        virtual bool Close() { return true; }
+    };
+
+    class DataFileFits : public DataFileImp
+    {
+    public:
+        DataFileFits(uint32_t id) :DataFileImp(id) { }
+        ~DataFileFits() { Close(); }
+
+        virtual bool Write() { return true; }
+        virtual bool Close() { return true; }
+    };
+
+    vector<DataFileImp*> fFiles;
+
+    bool fFitsFormat;
+
+    template<class T>
+        void Update(DimDescribedService &svc, const T &data) const
+    {
+        cout << "Update: " << svc.getName() << " (" << sizeof(T) << ")" << endl;
+        svc.setData(const_cast<T*>(&data), sizeof(T));
+        svc.updateService();
+    }
+
+    FileHandle_t runOpen(uint32_t runid, RUN_HEAD *h, size_t)
+    {
+        // Check if file already exists...
         cout << "OPEN_FILE #" << runid << endl;
         cout << " Ver= " << h->Version << endl;
@@ -157 +244 @@
         cout << " roi= " << h->Nroi << endl;
 
-        return 0;
-    }
-
-    int runWrite(int, EVENT *e, size_t)
-    {
-        cout << "WRITE_EVENT" << endl;
+        DataFileImp *file = NULL;
+        try
+        {
+            file = fFitsFormat ?
+                static_cast<DataFileImp*>(new DataFileFits(runid)) :
+                static_cast<DataFileImp*>(new DataFileRaw(runid));
+        }
+        catch (const exception &e)
+        {
+            return 0;
+        }
+
+        fFiles.push_back(file);
+
+        Update(fDimFiles, fFiles.size());
+
+//        fDimFiles.setData(fFiles.size());
+//        fDimFiles.update();
+
+        return reinterpret_cast<FileHandle_t>(file);
+    }
+
+    int runWrite(FileHandle_t handler, EVENT *e, size_t)
+    {
+        DataFileImp *file = reinterpret_cast<DataFileImp*>(handler);
+
+        cout << "WRITE_EVENT " << file->GetRunId() << endl;
 
         cout << " Evt=" << e->EventNum << endl;
@@ -170 +278 @@
         cout << " tim=" << e->PCTime << endl;
 
+        if (!file->Write())
+            return -1;
+
+        // ===> SignalEvtWritten(runid);
+        // Send num events written of newest file
+
+        /* close run runId (all all runs if runId=0) */
+        /* return: 0=close scheduled / >0 already closed / <0 does not exist */
+        //CloseRunFile(file->GetRunId(), time(NULL)+2) ;
+
         return 0;
     }
 
-    int runClose(int, RUN_TAIL *, size_t)
-    {
-        cout << "CLOSE_RUN" << endl;
-        return 0;
+    int runClose(FileHandle_t handler, RUN_TAIL *, size_t)
+    {
+        DataFileImp *file = reinterpret_cast<DataFileImp*>(handler);
+
+        cout << "CLOSE_RUN " << file->GetRunId() << endl;
+
+        fFiles.erase(find(fFiles.begin(), fFiles.end(), file));
+
+        Update(fDimFiles, fFiles.size());
+
+        //fDimFiles.setData(fFiles.size());
+        //fDimFiles.update();
+
+        const bool rc = file->Close();
+        if (!rc)
+        {
+            // Error message
+        }
+
+        delete file;
+
+        // ==> SignalRunClose(runid);
+        // Send new num open files
+        // Send empty file-name if no file is open
+
+        return rc ? 0 : -1;
     }
 };
@@ -185 +325 @@
 extern "C"
 {
-    int runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len)
+    FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len)
     {
         return EventBuilderWrapper::This->runOpen(irun, runhd, len);
     }
 
-    int runWrite(int fileId, EVENT *event, size_t len)
+    int runWrite(FileHandle_t fileId, EVENT *event, size_t len)
     {
         return EventBuilderWrapper::This->runWrite(fileId, event, len);
     }
 
-    int runClose(int fileId, RUN_TAIL *runth, size_t len)
+    int runClose(FileHandle_t fileId, RUN_TAIL *runth, size_t len)
     {
         return EventBuilderWrapper::This->runClose(fileId, runth, len);
     }
 
+    void factOut(int severity, int err, char *message)
+    {
+        //replace(message, message+strlen(message), '\n', ' ');
+
+        // FIXME: Make the output to the console stream thread-safe
+        ostringstream str;
+        str << "EventBuilder(";
+        if (err<0)
+            str << "---";
+        else
+            str << err;
+        str << "): " << message;
+        EventBuilderWrapper::This->Update(str, severity);
+    }
+
+    void factStat(int severity, int err, char* message )
+    {
+        static string last;
+        if (message==last)
+            return;
+
+        if (err!=-1)
+            factOut(severity, err, message);
+        else
+        {
+            ostringstream str("Status: ");
+            str << message;
+            EventBuilderWrapper::This->Update(str, severity);
+        }
+
+        last = message;
+    }
+
+    /*
     void message(int severity, const char *msg)
     {
         EventBuilderWrapper::This->Update(msg, severity);
-    }
+    }*/
 }