Akka.net 等待多个 FileSystemWatcher 可观察事件的 F# 有状态参与者

Question

我是 F# 和 Akka.Net 的新手，并试图通过它们实现以下目标：

我想创建一个接收文件位置的 actor (Tail)，然后使用 FileSystemWatcher 和一些 Observables 监听该位置的事件，将它们作为消息转发给其他 actor 进行处理。

我遇到的问题是用于侦听事件的代码一次只拾取一个事件而忽略所有其他事件。例如如果我将 20 个文件复制到正在观看的目录中，它似乎只发送其中 1 个的事件。

这是我的演员代码：

module Tail

open Akka
open Akka.FSharp
open Akka.Actor
open System
open Model
open ObserveFiles
open ConsoleWriteActor

let handleTailMessages tm =
    match tm with
        | StartTail (f,r) ->
            observeFile f consoleWriteActor |!> consoleWriteActor

    |> ignore

let spawnTail =
    fun (a : Actor<IMessage> )  -> 
    let rec l (count : int) = actor{

        let! m = a.Receive()
        handleTailMessages m
        return! l (count + 1)
    } 
    l(0)

下面是监听事件的代码：

module ObserveFiles
open System
open System.IO
open System.Threading
open Model
open Utils
open Akka
open Akka.FSharp
open Akka.Actor



let rec observeFile (absolutePath : string) (a : IActorRef )  = async{

    let fsw = new FileSystemWatcher(
                        Path = Path.GetDirectoryName(absolutePath), 
                        Filter = "*.*",
                        EnableRaisingEvents = true, 
                        NotifyFilter = (NotifyFilters.FileName ||| NotifyFilters.LastWrite ||| NotifyFilters.LastAccess ||| NotifyFilters.CreationTime ||| NotifyFilters.DirectoryName)
                        )

    let prepareMessage  (args: EventArgs) =
        let text = 
            match box args with
            | :? FileSystemEventArgs as fsa ->
                match fsa.ChangeType with
                | WatcherChangeTypes.Changed -> "Changed " + fsa.Name
                | WatcherChangeTypes.Created ->  "Created " + fsa.Name
                | WatcherChangeTypes.Deleted -> "Deleted " + fsa.Name
                | WatcherChangeTypes.Renamed -> "Renamed " + fsa.Name
                | _ -> "Some other change " + fsa.ChangeType.ToString()
            | :? ErrorEventArgs as ea -> "Error: " + ea.GetException().Message
            | o -> "some other unexpected event occurd" + o.GetType().ToString()
        WriteMessage text 


    let sendMessage x = async{  async.Return(prepareMessage x) |!> a
                                return! observeFile absolutePath a }

    let! occurance  = 
        [
        fsw.Changed |> Observable.map(fun x -> sendMessage (x :> EventArgs));
        fsw.Created |> Observable.map(fun x -> sendMessage (x :> EventArgs));
        fsw.Deleted |> Observable.map(fun x -> sendMessage (x :> EventArgs));
        fsw.Renamed |> Observable.map(fun x -> sendMessage (x :> EventArgs));
        fsw.Error |> Observable.map(fun x -> sendMessage (x :> EventArgs));
        ] 
        |> List.reduce Observable.merge
        |> Async.AwaitObservable

    return! occurance
}

为了达到这一点花了很多技巧，任何关于我如何改变它的建议，以便它在演员运行时拾取并处理所有事件将不胜感激.

Answer 1

在设计这样的任务时，我们可以将其拆分为以下组件：

创建负责接收所有消息的管理器 - 它的主要作用是响应传入的目录侦听请求。一旦请求进来，它就会创建一个子actor，负责在这个特定目录下监听。
子演员负责管理FileSystemWatcher特定路径。它应该订阅传入事件并将它们作为消息重定向到负责接收更改事件的参与者。它还应该在关闭时释放一次性资源。
Actor 负责接收更改事件 - 在我们的例子中是通过在控制台上显示它们。

示例代码：

open Akka.FSharp
open System
open System.IO

let system = System.create "observer-system" <| Configuration.defaultConfig()

let observer filePath consoleWriter (mailbox: Actor<_>) =    
    let fsw = new FileSystemWatcher(
                        Path = filePath, 
                        Filter = "*.*",
                        EnableRaisingEvents = true, 
                        NotifyFilter = (NotifyFilters.FileName ||| NotifyFilters.LastWrite ||| NotifyFilters.LastAccess ||| NotifyFilters.CreationTime ||| NotifyFilters.DirectoryName)
                        )
    // subscribe to all incoming events - send them to consoleWriter
    let subscription = 
        [fsw.Changed |> Observable.map(fun x -> x.Name + " " + x.ChangeType.ToString());
         fsw.Created |> Observable.map(fun x -> x.Name + " " + x.ChangeType.ToString());
         fsw.Deleted |> Observable.map(fun x -> x.Name + " " + x.ChangeType.ToString());
         fsw.Renamed |> Observable.map(fun x -> x.Name + " " + x.ChangeType.ToString());]
             |> List.reduce Observable.merge
             |> Observable.subscribe(fun x -> consoleWriter <! x)

    // don't forget to free resources at the end
    mailbox.Defer <| fun () -> 
        subscription.Dispose()
        fsw.Dispose()

    let rec loop () = actor {
        let! msg = mailbox.Receive()
        return! loop()
    }
    loop ()

// create actor responsible for printing messages
let writer = spawn system "console-writer" <| actorOf (printfn "%A")

// create manager responsible for serving listeners for provided paths
let manager = spawn system "manager" <| actorOf2 (fun mailbox filePath ->
    spawn mailbox ("observer-" + Uri.EscapeDataString(filePath)) (observer filePath writer) |> ignore)

manager <! "testDir"

Akka.net 等待多个 FileSystemWatcher 可观察事件的 F# 有状态参与者

Akka.net F# stateful actor that awaits multipe FileSystemWatcher Observable events

f#

asynchronous

filesystemwatcher

observable

akka.net