Buffering & Triggered Transducer

[haf]

Similar to the NLog wrapper, buffer and async targets, that go around other targets.

Decide whether we should have it around targets or if it should be before the targets inside the registry.

Why? Because it drastically can reduce the amount of data that needs to be shipped by the target, while not necessarily removing detail where it's needed.

Could be written as a transducer that is trigged by some condition, e.g. a fatal-level Message. When triggered, yields all previously buffered Messages to the target.

Write as a part of the filtering pipeline.

[lust4life]

maybe this can be addressed by Events Processing like this. in the next , will add a test case.

[lust4life]

      let dutiful (msg : Message, buffered : Message list) =
        let dt =DateTime.Parse("10:10").ToUniversalTime()
        let timeNow = Instant.FromDateTimeUtc(dt)
        let tenMins = Duration.FromMinutes(10L)
        let durationFromFirstBufferMsg =
          match buffered |> List.tryHead with
          | Some firstMsg -> 
            timeNow - (firstMsg.timestamp |> Instant.ofEpoch)
          | None -> Duration.MinValue
        if msg.level >= Error then Pipe.BufferAction.Delivery
        elif buffered.Length >= 4 || durationFromFirstBufferMsg > tenMins then Pipe.BufferAction.Reset
        else Pipe.BufferAction.AddToBuffer

      // context
      let processing =
        Events.stream
        |> Events.subscribers [
           Events.events
           |> Pipe.bufferConditional dutiful
           |> Pipe.map Seq.ofList
           |> Events.flattenToProcessing
           |> Pipe.map (fun msg -> 
              let sinks = if msg.level >= Error then ["email"] else ["dashboard"]
              msg |> Message.addSinks sinks)
        ]
        |> Events.toProcessing

      // given
      let! targets = [mockTarget "email"; mockTarget "dashboard";] |> Job.seqCollect
      let! (sendMsg, ctss) = run processing targets

      // when
      let makeMsg level time = 
        Message.event level (Guid.NewGuid().ToString("N"))
        |> Message.setTimestamp (Instant.FromDateTimeUtc(DateTime.Parse(time).ToUniversalTime()))
        |> sendMsg

      do! makeMsg Info "10:01"
      do! makeMsg Debug "10:02"
      do! makeMsg Verbose "10:05"
      do! makeMsg Info "10:05"
      do! makeMsg Info "10:05"

      let! msgsFromEachTarget = targets |> Seq.Con.mapJob (fun t -> t.getMsgs ())
      let (msgFromEmail, msgFromDashboard) = (msgsFromEachTarget.[0],msgsFromEachTarget.[1])
      Expect.isEmpty msgFromEmail "no error msg occur,so no msg"
      Expect.isEmpty msgFromDashboard "no error msg occur,so no msg"

      do! makeMsg Info "10:05"
      do! makeMsg Warn "10:06"
      do! makeMsg Warn "10:07"
      do! makeMsg Error "10:08"
      let! msgsFromEachTarget = targets |> Seq.Con.mapJob (fun t -> t.getMsgs ())
      let (msgFromEmail, msgFromDashboard) = (msgsFromEachTarget.[0],msgsFromEachTarget.[1])
      Expect.equal msgFromEmail.Length 1 "1 error msg send to email"
      Expect.equal msgFromDashboard.Length 3 "3 other msg send to dashboard"
   
      do! makeMsg Fatal "10:09"
      let! msgsFromEachTarget = targets |> Seq.Con.mapJob (fun t -> t.getMsgs ())
      let (msgFromEmail, msgFromDashboard) = (msgsFromEachTarget.[0],msgsFromEachTarget.[1])
      Expect.isEmpty msgFromDashboard "no bufferd msg"
      Expect.equal msgFromEmail.Length 1 "1 error msg continue send to email"
      
      do! makeMsg Info "09:50"
      do! makeMsg Info "09:55"
      do! makeMsg Info "10:05"
      do! makeMsg Info "10:06"
      do! makeMsg Fatal "10:15"
      let! msgsFromEachTarget = targets |> Seq.Con.mapJob (fun t -> t.getMsgs ())
      let (msgFromEmail, msgFromDashboard) = (msgsFromEachTarget.[0],msgsFromEachTarget.[1])
      Expect.equal msgFromEmail.Length 1 "1 Fatal msg send to email"
      Expect.equal msgFromDashboard.Length 2 "2 other msg send to dashboard, fist two are discarded beacuse they are too old"

seems can work: https://github.com/logary/logary/blob/feature/live-wpc-reconfigure/src/tests/Logary.Tests/Engine.fs#L244