Filter6.cc

Go to the documentation of this file.

 
/* LABEL-START */
 
#include  "debug_mq.h"
#include  "LibMqMsgque_cc.hh"
 
/* LABEL-END */
 
#define META_CONTEXT_S &context
 
#include <stdexcept>
#include <stdlib.h>
#include <stdio.h>
#include <regex>
#include <fstream>
 
using namespace std;
using namespace ccmqmsgque;
 
class Filter6 : public MqContextC, public MqServerSetupIF, public MqServerCleanupIF, public MqEventIF {
  friend class MqFactoryCT<Filter6>; 
 
  private:
    static MK_I32 retryCnt;
 
  private:
    Filter6(MK_TYP const typ, MqContextC* tmpl=NULL) : MqContextC(typ, tmpl) {
      ConfigSetIgnoreExit(true);
      rgx = "^(?:Filter6-1|Filter6|fs1.*)$";
    }
 
  private:
    std::ofstream outfile;
    regex rgx;
 
    void ErrorWrite () {
      auto err = ErrorFORMAT();
      if (outfile.is_open()) {
        outfile << "ERROR: " << err->GetText() << endl;
        outfile.flush();
      } else {
        err->Println();
      }
      err->Reset();
    }
 
    void LOGF () {
      // get the "link-target"
      MqContextC *ftr = SlaveGetFilter();
      // check "Ident" from the "link-target"
      if (std::regex_match(ftr->ConfigGetName(),rgx)) {
        // if "Ident" is not "transFilter" use the data as file-name to open a file 
        outfile.open(ReadSTR(), std::ios_base::app);
      } else {
        // if "Ident" is "transFilter" send the data to the "link-target"
        ProxyForward(ftr);
      }
      SendRETURN();
    }
 
    static void WRIT (MqContextC *ctx) {
      // get the "master"
      Filter6 *master = dynamic_cast<Filter6*>(ctx->SlaveGetMaster());
      if (std::regex_match(master->ConfigGetName(),master->rgx)) {
        master->outfile << ctx->ReadSTR() << endl;
        master->outfile.flush();
      } else {
        ctx->ProxyForward(master);
      }
      ctx->SendRETURN();
    }
 
    void EXIT () {
      Exit();
    }
 
    void SOEX () {
      ErrorFORMAT()->SetEXIT();
    }
 
    void FilterIn () {
      StorageExport();
      SendRETURN();
    }
 
    void Event () {
      // check if data is available
      if (StorageCount() == 0LL) {
        // no data available, set error-code to CONTINUE
        ErrorFORMAT()->SetCONTINUE();
      } else {
        MQ_LTR Id = 0LL;
        // an item is available, try to send the data
        try {
          // get the filter-context
          MqContextC *ftr = SlaveGetFilter();
          // read package from storage
          Id = StorageImport();
          // forward the entire BDY data to the ftr-target
          try {
            ProxyForward(ftr);
          } catch (const MkExceptionC& e) {
            if (StorageErrCnt(Id) < Filter6::retryCnt) {
              StorageDecrRef(Id);
              return;
            } else {
              throw;
            }
          } catch (...) {
            throw;
          }
        } catch (const exception& e) {
          ErrorCatch (e);
          // on error write the error-text and "forget" the data
          ErrorWrite();
        }
        // on "success" or on "error" delete item from storage
        if (Id != 0LL) StorageDelete(&Id);
      }
    }
 
    void ServerCleanup() {
      outfile.close();
    }
 
    // [filter_service_example]
    void ServerSetup() {
      MqContextC *ftr = SlaveGetFilter();
      // SERVER: listen on every token (+ALL)
      ServiceCreate       ("LOGF", MqServiceICB(&Filter6::LOGF));
      ServiceCreate       ("EXIT", MqServiceICB(&Filter6::EXIT));
      ServiceCreate       ("SOEX", MqServiceICB(&Filter6::SOEX));
      ServiceCreate       ("+ALL", MqServiceICB(&Filter6::FilterIn));
      ServiceStorage      ("PRNT");
      ServiceStorage      ("PRN2");
      ftr->ServiceCreate  ("WRIT", MqTokenCCB(&Filter6::WRIT));
      ServiceCreate       ("WRIT", MqTokenCCB(&Filter6::WRIT));
      ftr->ServiceProxy   ("WRT2", MQ_SLAVE_MASTER);
 
      if (isParent()) {
        LinkCheckOptionI32  ("--retryCnt", &Filter6::retryCnt);
      }
    }
    // [filter_service_example]
 
  public:
 
    static void __attribute__ ((noreturn))
    Help (MK_STRN base)
    {
      fputs("\n", stderr);
      fprintf(stderr, "usage: %s [OPTION]... [ARGUMENT]...\n", base);
      fputs("\n", stderr);
      fputs("  This tool is database (MqStorage...) test tool of NHI1.\n", stderr);
      fputs("\n", stderr);
      fprintf(stderr, "  %s [ARGUMENT]... syntax:\n", base);
      fprintf(stderr, "    aclient... %c %s ... %c aserver\n", MK_ALFA, base, MK_ALFA);
      fputs("\n", stderr);
      fputs(MqHelp (NULL), stderr);
      fputs("\n", stderr);
      fprintf(stderr, "  %s [OPTION]:\n", base);
      fputs(          "    -h, --help       print this help\n", stderr);
      fputs("\n", stderr);
      fputs(          "    --retryCnt       how often an item should be send on error (default: 3)\n", stderr);
      fputs("\n", stderr);
 
      exit(EXIT_SUCCESS);
    }
};
 
int Filter6::retryCnt = 3;
 
/*****************************************************************************/
/*                                                                           */
/*                               M A I N                                     */
/*                                                                           */
/*****************************************************************************/
 
// [error_example]
int MK_CDECL main (int argc, MK_STRN argv[])
{
  MqMsgque::CcMqSetup();
  // define factory
  auto Filter6F = MqFactoryCT<Filter6>::Add("Filter6");
  // modify default type
  Filter6F->Type()->fHelp = Filter6::Help;
  // create object from factory
  Filter6 *filter = Filter6F->New();
  try {
    filter->LinkCreate (MkBufferListC {argc, argv});
    filter->ProcessEvent (MQ_WAIT_FOREVER);
  } catch (const exception& e) {
    filter->ErrorCatch(e);
  }
  return filter->Exit();
}
// [error_example]