You can support this project by donating at:

Or see the list of project sponsors.

pipingeventsthreads

Documentation	🤨
Docstrings	😀
Tests	😀
Examples	🥺
RepositoryActivity	🥺
CI	🥺

This library in some sense similar to the cl-events, reviewed yesterday. It allows defining pipelines to process messages.

Each message can be processed sequentially or in parallel. Each node would be an instance of the segment class. There are two kinds of nodes - intermediate and final.

Intermediate nodes can filter messages or route them into other pipelines.

Final nodes are called faucets. They process the message and stop processing.

For example here is how we can build a log message processing using piping. We want to print all ERROR messages to *error-output* and to write all messages to the log file.

To create this pipeline, we need following segments. Here is "Pipeline" is a chain of segments to pass the message through:

Pipeline:
  Print["full.log"]
  Pipeline: Filter[if ERROR] -> Print[*error-output*]

Here is how we can configure this pipeline in Lisp code:

POFTHEDAY> (defparameter *pipe*
             (make-instance 'piping:pipeline))

POFTHEDAY> (defparameter *log-printer*
             (piping:add-segment
               *pipe*
               (make-instance 'piping:printer
                              :stream (open "full.log"
                                            :direction :output
                                            :if-exists :append
                                            :if-does-not-exist :create))))

;; This adds a sub-pipe where we'll filter message
;; and print if it stats with "ERROR":
POFTHEDAY> (piping:add-segment *pipe* :pipe)

POFTHEDAY> (piping:add-segment *pipe*
            (make-instance 'piping:predicate-filter
                           :predicate (lambda (message)
                                        (str:starts-with-p "ERROR: " message)))
            '(1))

POFTHEDAY> (piping:add-segment *pipe*
            (make-instance 'piping:printer
                           :stream *error-output*)
            '(1))

;; Now we'll pass two messages through this pipeline:
POFTHEDAY> (piping:pass *pipe*
             "INFO: Hello world!")

;; This one will be printed to *error-output*:
POFTHEDAY> (piping:pass *pipe*
             "ERROR: Something bad happened!")

"ERROR: Something bad happened!" 

;; But in file both messages are present:
POFTHEDAY> (force-output (piping:print-stream *log-printer*))

POFTHEDAY> (princ (alexandria:read-file-into-string "full.log"))

"INFO: Hello world!" 
"ERROR: Something bad happened!"

Working on this example, I found two things:

• there is no component to fanout messages into the nested segments or sub pipes.
• using indices to point to a place where a segment should be added is very inconvenient.

@Shinmera uses piping in his logging library verbose. I skimmed through its sources and didn't find if he has some solution of fanout absence problem.

Definitely, this library can be made a more convenient if somebody is interested to use it for other purposes.