收集事件中的值

Question

我有一个事件。 该事件不时被触发，它调用一个事件处理程序，它是一个Action<int>。

现在我想“收集”这些事件发送给我的整数并将它们保存在列表中。我也想指定列表完成并开始新列表的时刻。

我想到的天真的解决方案是属性List<int> ValList。 事件处理程序每​​次调用时都会添加一个值。 消费者端在需要时会列出并且会不时说出ValList = new List<int>(); 为了避免线程同步的问题，我还需要一个锁。

我觉得这个解决方案非常丑陋，想知道替代品。随着时间的推移，我变得越来越多的功能程序员，我经常使用它。 但是，当涉及到这样的问题时，我仍然在思考程序问题。 我真的想避免一个可变列表（我仍然使用System.Collections.Immutable）。

是否有一个很好的功能解决方案，没有可变和副作用？

Answer 1

您应该考虑对此使用Reactive Extensions。它处理值（事件）流，并可以删除对锁的需求。

首先，我要为Add,Complete和RequestView动作定义一些动作类。这是为了表现得像在F＃例如识别联合：

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 
{ 
} 

接下来我会创建一个名为AggregateView类型将举行三个的Rx Subject值：

• aggregator将收集EventAction事件和管理汇总Lst<int>（Lst<int>是the language-ext functional language extensions library的不可变列表类型，但您也可以使用ImmutableList）。
• events，这将仅仅是整数事件
• views流，这将是Lst<int>意见

这里流在班上是：

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(); 
    } 
} 

它有两个IObservable属性：

• Views - 它允许您订阅的总名单
• Events - 它允许您订阅整数事件

也有一些有用的方法：

• Listener - 这就是你将插入您的event
• Complete - 这将清空聚合列表并发送一个空列表到View可观察
• RequestView - 这会将当前聚集列表发送给Views可观察对象的所有订阅者。

最后进行测试：

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); 
} 

这是我能想到的事件打交道时最实用的方式。 Action<int>应该始终是红旗对于任何想要在功能上工作的人，因为默认情况下它有副作用并且不纯。所以你需要尽可能地封装副作用，并使其他一切都变得纯洁。

顺便说一句，你可以概括这整个事情与任何类型的工作。这使它更有用：

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

如果我完全理解你说的事情，你的事件被锁定到Action，并且你无法控制该事件签名。您希望收集每个传递的int值，直到某个外部请求用来检索具有任何累积整数的列表为止，此时列表将重置列表并保存关于该集合的时间信息。那是对的吗？

让我困惑的是为什么你提到函数式编程的意思是OOP的语言？你可能会认为LINQ在某种程度上是功能性的，但是肯定有更好的功能选择。因为这对于累加器管理器类似乎是一个非常简单的解决方案。

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(); 
    } 
} 
}