Example: Filter6

Documentation of the Filter6 tool used for trans2.test.

INTRODUCTION

The Filer6 tool is used to test the filter-feature of ccmqmsgque.

• read more at: filter
• read more at: MqContextC-ServiceApi
• read more at: MqContextC-StorageApi
• read more at: MkLogFileC

To run the filter test, a first client, one or more filters and a final server are created. All are connected to the ccmqmsgque protocol.

The trans2.test carries out common filter tests and special stress tests. A stress test is performed by exiting one or more filters or servers and observing the response and behavior when reconnecting.

The GOAL for this test is:

• test services: ServiceCreate, ServiceStorage and ServiceProxy
• test database: StorageExport, StorageImport, StorageErrCnt, StorageDecrRef and StorageDelete
• test logging: LogFileOpen and LogFileWriteC
• test proxy: ProxyForward
• test kill-and-recover feature

CODE filter

/**
 *   @file         NHI1/example/cc/Filter6.cc
 *   @brief        tag: nhi1-release-250425
 *   @copyright    (C) NHI - #1 - Project - Group
 *                 This software has NO permission to copy,
 *                 please contact AUTHOR for additional information
 */

/* 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]