Example: Filter6

Documentation of the Filter6 tool used for trans2.test.

INTRODUCTION

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

• 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 pymqmsgque 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

#+
#:   -*- coding: utf8 -*-
#:
#:   @file         NHI1/example/py/Filter6.py
#:   @brief        Filter6.py - 23 Jul 2024 - aotto1968
#:   @copyright    (C) NHI - #1 - Project - Group
#:                 This software has NO permissions to copy,
#:                 please contact AUTHOR for additional information
#:   @version      e6ce5cfbf6c7dfb4f0b3eb45bb50508830143224
#:   @date         Tue Jul 23 22:39:27 2024 +0200
#:   @author       aotto1968 <aotto1968@t-online.de>
#:

import sys
import os
import re
from pymqmsgque import *

class Filter6(MqContextC):

  def __init__(ctx, tmpl=None):
    ctx.ConfigSetIgnoreExit(True)
    ctx.ConfigSetServerSetup(Filter6.FilterSetup)
    ctx.ConfigSetServerCleanup(Filter6.FilterCleanup)
    ctx.ConfigSetEvent(ctx.FilterEvent)
    ctx.itms = []
    ctx.FH = None
    super().__init__()

  ## [filter_service_example]
  def FilterSetup(ctx):
    ftr = ctx.SlaveGetFilter()
    ctx.ServiceCreate("LOGF", Filter6.LOGF)
    ctx.ServiceCreate("EXIT", Filter6.EXIT)
    ctx.ServiceCreate("SOEX", Filter6.SOEX)
    ctx.ServiceStorage("PRNT")
    ctx.ServiceStorage("PRN2")
    ctx.ServiceCreate("+ALL", Filter6.FilterIn)
    ftr.ServiceCreate("WRIT", Filter6.WRIT)
    ctx.ServiceCreate("WRIT", Filter6.WRIT)
    ftr.ServiceProxy("WRT2",  MqSlaveE.MASTER)
  ## [filter_service_example]

  def FilterCleanup(ctx):
    ctx.FH = None

  def errorWrite(ctx):
    err = ctx.ErrorFORMAT()
    ctx.FH.WriteC("ERROR: " + err.GetText())
    err.Reset()

  def LOGF(ctx):
    ftr = ctx.SlaveGetFilter()
    if re.match(r'^(?:Filter6-1|Filter6|fs1.*)$',ftr.ConfigGetName()):
      ctx.FH = pymkkernel.MkLogFileC.Open(ctx,ctx.ReadSTR())
    else:
      ctx.ProxyForward(ftr)
    ctx.SendRETURN()

  def WRIT(ftr):
    master = ftr.SlaveGetMaster()
    if re.match(r'^(?:Filter6-1|Filter6|fs1.*)$',master.ConfigGetName()):
      master.FH.WriteC(ftr.ReadSTR())
    else:
      ftr.ProxyForward(master)
    ftr.SendRETURN()

  def EXIT(ctx):
    ctx.Exit()

  def SOEX(ctx):
    ctx.ErrorFORMAT().SetEXIT()

  def FilterIn(ctx):
    ctx.StorageExport()
    ctx.SendRETURN()

  def FilterEvent(ctx):
    if ctx.StorageCount() == 0:
      ctx.ErrorFORMAT().SetCONTINUE()
    else:
      id = 0
      try:
        ftr = ctx.SlaveGetFilter()
        id = ctx.StorageImport()
        try:
          ctx.ProxyForward(ftr)
        except pymkkernel.MkExceptionC:
          ex = ctx.StorageErrCnt(id)
          if ex <= 3:
            ctx.StorageDecrRef(id)
            return
          else:
            raise
        except:
          raise
      except Exception as ex:
        ctx.ErrorCatch(ex);
        ctx.errorWrite()
      ctx.StorageDelete(id)

# [error_example]
if __name__ == "__main__":
  srv = MqFactoryC.Add(Filter6).New();
  try:
    srv.LinkCreate(sys.argv)
    srv.ProcessEvent(MqWaitOnEventE.FOREVER)
  except Exception as ex:
    srv.ErrorCatch(ex);
  finally:
    srv.Exit()
# [error_example]