Werte von Ereignissen sammeln

Question

Ich habe ein Ereignis. Das Ereignis wird von Zeit zu Zeit ausgelöst und ruft einen Event Handler auf, der eine Action<int> ist.

Jetzt möchte ich diese Ganzzahlen "sammeln" das Ereignis liefert mich und speichern Sie sie in Listen. Ich möchte auch den Zeitpunkt angeben, an dem eine Liste vervollständigt und eine neue Liste gestartet wird.

Die naive Lösung, an die ich dachte, ist eine Eigenschaft List<int> ValList. Der Ereignishandler fügt bei jedem Aufruf einen Wert hinzu. Die Consumer-Seite nimmt die Liste, wenn es will und wird von Zeit zu Zeit sagen ValList = new List<int>(); Um Probleme mit der Thread-Synchronisation zu vermeiden, brauche ich auch eine Sperre.

Ich finde diese Lösung so schrecklich hässlich und fragte mich Alternativen. Mit der Zeit wurde ich mehr und mehr ein funktionaler Programmierer und ich benutze es sehr oft. Aber wenn es um solche Probleme geht, denke ich immer noch an prozedural. Ich möchte wirklich eine veränderbare Liste vermeiden (ich verwende immer noch System.Collections.Immutable).

Gibt es eine schöne funktionelle Lösung ohne Mutables und Nebenwirkungen?

Answer 1

Sie sollten Reative Extensions dafür verwenden. Es behandelt Datenströme (Ereignisse) und kann die Notwendigkeit von Sperren beseitigen.

Zuerst werde ich einige Aktionsklassen für Add, Complete und RequestView Aktionen definieren. Das ist wie eine diskriminiert Vereinigung in F # zum Beispiel verhalten:

public class EventAction 
{ 
    public static EventAction Add(int value) => new AddAction(value); 
    public static readonly RequestViewAction RequestView = new RequestViewAction(); 
    public static readonly EventAction Complete = new CompleteAction(); 
} 

public class AddAction : EventAction 
{ 
    public readonly int Value; 
    public AddAction(int value) => Value = value; 
} 

public class CompleteAction : EventAction 
{ 
} 

public class RequestViewAction : EventAction 
{ 
} 

Als nächstes werde ich eine Art AggregateView genannt schaffen es drei Rx Subject Werte halten:

• aggregator die die EventAction sammeln Ereignisse und verwalten eine aggregierte Lst<int> (Lst<int> ist eine unveränderliche Liste Typ von the language-ext functional language extensions library, aber Sie können auch ImmutableList verwenden).
• events, die einfach ein Strom der ganzzahligen Ereignisse
• views, die einen Strom von Lst<int> Ansichten

Hier ist die Klasse sein wird, wird sein:

using System; 
using LanguageExt; 
using static LanguageExt.Prelude; 
using System.Reactive.Linq; 
using System.Reactive.Subjects; 

public class AggregateView : IDisposable 
{ 
    readonly Subject<EventAction> aggregator = new Subject<EventAction>(); 
    readonly Subject<int> events = new Subject<int>(); 
    readonly Subject<Lst<int>> view = new Subject<Lst<int>>(); 

    readonly IDisposable subscription; 

    public AggregateView() 
    { 
     // Creates an aggregate view of the integers that responds to various control 
     // actions coming through. 
     subscription = aggregator.Aggregate(
      Lst<int>.Empty, 
      (list, action) => 
      { 
       switch(action) 
       { 
        // Adds an item to the aggregate list and passes it on to the 
        // events Subject 
        case AddAction add: 
         events.OnNext(add.Value); 
         return list.Add(add.Value); 

        // Clears the list and passes a list onto the views Subject 
        case CompleteAction complete: 
         view.OnNext(Lst<int>.Empty); 
         return Lst<int>.Empty; 

        // Gets the current aggregate list and passes it onto the 
        // views Subject 
        case RequestViewAction req: 
         view.OnNext(list); 
         return list; 

        default: 
         return list; 
       } 
      }) 
      .Subscribe(x => { }); 
    } 

    /// <summary> 
    /// Observable stream of integer events 
    /// </summary> 
    public IObservable<int> Events => 
     events; 

    /// <summary> 
    /// Observable stream of list views 
    /// </summary> 
    public IObservable<Lst<int>> Views => 
     view; 

    /// <summary> 
    /// Listener for plugging into an event 
    /// </summary> 
    public void Listener(int value) => 
     aggregator.OnNext(EventAction.Add(value)); 

    /// <summary> 
    /// Clears the aggregate view and post it to Views 
    /// </summary> 
    public void Complete() => 
     aggregator.OnNext(EventAction.Complete); 

    /// <summary> 
    /// Requests a the current aggregate view to be pushed through to 
    /// the Views subscribers 
    /// </summary> 
    public void RequestView() => 
     aggregator.OnNext(EventAction.RequestView); 

    /// <summary> 
    /// Dispose 
    /// </summary> 
    public void Dispose() 
    { 
     subscription?.Dispose(); 
     view?.OnCompleted(); 
     events?.OnCompleted(); 
     view?.Dispose(); 
     events?.Dispose(); 
    } 
} 

Es hat zwei IObservable Eigenschaften:

• Views - die Sie zu den aggregierten Listen
• Events abonnieren - die Sie zu den Integer-Ereignisse

Auch gibt es einige nützliche Methoden abonnieren:

• Listener - das ist, was Sie ‚ll-Stecker in Ihre event
• Complete - das wird die aggregierte Liste leeren und eine leere Liste bis zum View beobachtbaren
senden
• RequestView - dies sendet die aktuelle Aggregatliste an alle Teilnehmer der Views Observable.

Schließlich, es zu testen:

class Program 
{ 
    static event Action<int> eventTest; 

    static void Main(string[] args) 
    { 
     var aggregate = new AggregateView(); 
     eventTest += aggregate.Listener; 

     aggregate.Views.Subscribe(ReceiveList); 
     aggregate.Events.Subscribe(ReceiveValue); 

     eventTest(1); 
     eventTest(2); 
     eventTest(3); 
     eventTest(4); 
     eventTest(5); 

     aggregate.RequestView(); 
     aggregate.Complete(); 

     eventTest(6); 
     eventTest(7); 
     eventTest(8); 
     eventTest(9); 
     eventTest(10); 

     aggregate.RequestView(); 
    } 

    static void ReceiveList(Lst<int> list) => 
     Console.WriteLine($"Got list of {list.Count} items: {ListShow(list)}"); 

    static void ReceiveValue(int x) => 
     Console.WriteLine(x); 

    static string ListShow(Lst<int> list) => 
     String.Join(", ", list); 
} 

dies die funktionelle Art, wie ich mir vorstellen kann, wenn es mit Ereignissen zu tun. Action<int> sollte immer eine rote Flagge für jeden sein, der funktionell arbeiten möchte, weil es standardmäßig Nebenwirkungen hat und nicht rein ist. Also müssen Sie die Nebenwirkungen so gut wie möglich einkapseln und alles andere rein machen.

Übrigens können Sie diese ganze Sache verallgemeinern, um mit jedem Typ zu arbeiten. Was macht es viel nützlicher:

public enum EventActionTag 
{ 
    Add, 
    Complete, 
    RequestView 
} 

public class EventAction<T> 
{ 
    public readonly EventActionTag Tag; 

    public static EventAction<T> Add(T value) => new AddAction<T>(value); 
    public static readonly EventAction<T> RequestView = new RequestViewAction<T>(); 
    public static readonly EventAction<T> Complete = new CompleteAction<T>(); 

    public EventAction(EventActionTag tag) => 
     Tag = tag; 
} 

public class AddAction<T> : EventAction<T> 
{ 
    public readonly T Value; 
    public AddAction(T value) : base(EventActionTag.Add) => 
     Value = value; 
} 
public class CompleteAction<T> : EventAction<T> 
{ 
    public CompleteAction() : base(EventActionTag.Complete) 
    { } 
} 
public class RequestViewAction<T> : EventAction<T> 
{ 
    public RequestViewAction() : base(EventActionTag.RequestView) 
    { } 
} 

public class AggregateView<T> : IDisposable 
{ 
    readonly Subject<EventAction<T>> aggregator = new Subject<EventAction<T>>(); 
    readonly Subject<T> events = new Subject<T>(); 
    readonly Subject<Lst<T>> view = new Subject<Lst<T>>(); 

    readonly IDisposable subscription; 

    public AggregateView() 
    { 
     // Creates an aggregate view of the integers that responds to various control 
     // actions coming through. 
     subscription = aggregator.Aggregate(
      Lst<T>.Empty, 
      (list, action) => 
      { 
       switch(action.Tag) 
       { 
        // Adds an item to the aggregate list and passes it on to the 
        // events Subject 
        case EventActionTag.Add: 
         var add = (AddAction<T>)action; 
         events.OnNext(add.Value); 
         return list.Add(add.Value); 

        // Clears the list and passes a list onto the views Subject 
        case EventActionTag.Complete: 
         view.OnNext(Lst<T>.Empty); 
         return Lst<T>.Empty; 

        // Gets the current aggregate list and passes it onto the 
        // views Subject 
        case EventActionTag.RequestView: 
         view.OnNext(list); 
         return list; 

        default: 
         return list; 
       } 
      }) 
      .Subscribe(x => { }); 
    } 

    /// <summary> 
    /// Observable stream of integer events 
    /// </summary> 
    public IObservable<T> Events => 
     events; 

    /// <summary> 
    /// Observable stream of list views 
    /// </summary> 
    public IObservable<Lst<T>> Views => 
     view; 

    /// <summary> 
    /// Listener for plugging into an event 
    /// </summary> 
    public void Listener(T value) => 
     aggregator.OnNext(EventAction<T>.Add(value)); 

    /// <summary> 
    /// Clears the aggregate view and post it to Views 
    /// </summary> 
    public void Complete() => 
     aggregator.OnNext(EventAction<T>.Complete); 

    /// <summary> 
    /// Requests a the current aggregate view to be pushed through to 
    /// the Views subscribers 
    /// </summary> 
    public void RequestView() => 
     aggregator.OnNext(EventAction<T>.RequestView); 

    /// <summary> 
    /// Dispose 
    /// </summary> 
    public void Dispose() 
    { 
     subscription?.Dispose(); 
     view?.OnCompleted(); 
     events?.OnCompleted(); 
     view?.Dispose(); 
     events?.Dispose(); 
    } 
} 

Answer 2

Wenn ich vollständig verstehe, was Sie sagen, ist Ihr Ereignis auf Aktion gesperrt und Sie haben keine Kontrolle über diese Ereignissignatur. Sie möchten jeden int-Wert bis zu einem Punkt sammeln, an dem eine externe Anforderung zum Abrufen der Liste mit allen akkumulierten Ganzzahlen auftritt. An diesem Punkt wird die Liste zurückgesetzt und die Zeitinformationen über die Sammlung werden gespeichert. Ist das korrekt?

Was verwirrt mich, warum Sie funktionale Programmierung in einer Sprache erwähnen, die OOP sein soll? Sie könnten argumentieren, dass LINQ bis zu einem gewissen Grad funktional ist, aber sicherlich gibt es bessere Möglichkeiten für die funktionell Gesinnten? Weil dies scheint eine ziemlich einfache Lösung mit einem Akku-Manager-Klasse.

namespace bs 
{ 
struct CollectionEvent 
{ 
    public DateTime Retrieved { get; set; }  
    public String IP { get; set; } 
} 

static class Accumulator 
{ 
    private static List<int> Items { get; set; } = new List<int>(); 
    private static bool Mutex { get; set; } = false; 
    private static List<CollectionEvent> Collections { get; set; } = new List<CollectionEvent>(); 

    public static void Add(int i) 
    { 
     Sync(() => Items.Add(i)); 
    } 

    public static List<int> Retrieve(String IP) 
    { 
     Collections.Add(new CollectionEvent 
     { 
      Retrieved = DateTime.UtcNow, 
      IP = IP 
     }); 

     List<int> dataOut = null; 
     Sync(() => 
     { 
      dataOut = new List<int>(Items); 
      Items = new List<int>(); 
     }); 

     return dataOut; 
    } 

    public static void Sync(Action fn) 
    { 
     const int Threshold = 10; 
     int attempts = 0; 

     for (; Mutex && (attempts < Threshold); attempts++) 
      Thread.Sleep(100 * attempts); 

     if (attempts == Threshold) 
      throw new Exception(); // or something better I'm sure 

     Mutex = true; 
     fn(); 
     Mutex = false; 
    } 
} 

class Program 
{ 
    static void Main(string[] args) 
    { 
     var r = new Random(); 
     var m = r.Next(5, 10); 

     for (int i = 0; i < m; i++) 
     { 
      var datum = r.Next(100, 10000); 
      Console.WriteLine($"COLLECT {datum}"); 
      Accumulator.Add(datum); 
     } 

     Console.WriteLine("RETRIEVE"); 
     Accumulator.Retrieve("0.0.0.0").ForEach(i => Console.WriteLine($"\t{i}")); 

     m = r.Next(5, 10); 
     for (int i = 0; i < m; i++) 
     { 
      var datum = r.Next(100, 10000); 
      Console.WriteLine($"COLLECT {datum}"); 
      Accumulator.Add(datum); 
     } 

     Console.WriteLine("RETRIEVE"); 
     Accumulator.Retrieve("0.0.0.0").ForEach(i => Console.WriteLine($"\t{i}")); 

     Console.Read(); 
    } 
} 
}