sum.conf

@version: 3.21
@include "scl.conf"

@define loop-socket-path "/tmp/state.sock"

options { time-reopen(5); };

python {
import time

class Wait(object):
  def parse(self, log_message):
    time.sleep(1)
    return True

};

block parser sleep()
{
   python( class(Wait) );  
};

block source state()
{
   channel {
     source { unix-stream("/tmp/state.sock" flags(no-parse)); };
     parser { syslog-parser(flags(syslog-protocol)); }; # For some reason this is needed for ewmm-parser()
     parser { ewmm-parser(); };
   }
};

log {
  source { example-msg-generator( num(1) ); };
  rewrite { set("1" value(".sum")); };

  destination { unix-stream("/tmp/state.sock" template("$(format-ewmm)") persist-name("1")); };
};


block source input()
{
   channel {
     source { stdin(flags(no-parse)); };
     rewrite { set("1" value(".input")); };
   }
};

log {
  source { input(); };
  source { state(); };

  parser { kv-parser(prefix(".")); };

  parser {
    grouping-by(
       key("")
       aggregate(
         value(".pass-throu" "1")
       )

       trigger( match("1", value(".input")) )
       timeout(9999999) #hopefully never
    );
  };

  if {
  filter { message("reset") };
  rewrite { set("0" value(".sum")); };
  }
  else {
  rewrite { set("$(+ ${.sum} $MESSAGE)" value(".sum")); };
  };

  filter { match("1" value(".pass-throu")); };
  rewrite { set("0" value(".pass-throu")); };
  rewrite { set("0" value(".input")); };
  rewrite { set("0" value("MESSAGE")); };


  destination { file("/dev/stdout" template("SUM=${.sum}\n")); };
  destination { unix-stream("/tmp/state.sock" template("$(format-ewmm)")); };
};