C＃intra-app Generic Pub/Sub子集模式

Question

我並不完全確定我正在尋找的正確術語，因此請耐心等待。在我維護的應用程序中，我們有一個標準的PubSub實現，用於從多個不同的發佈者獲取數據到應用程序中。我們有一個「服務」用於訂閱處理PubSub協商的數據，不同種類的數據有不同的服務（例如產品，訂單，供應商等）。這些服務通常通過普通的vanilla .NET事件通知消息的應用程序。

然而並非所有應用程序的所有部分都對所有事件感興趣。例如，有些地方可能只對產品A，B和C感興趣，而應用程序中的另一個地方則需要產品B，D和E，而且通常由用戶輸入驅動。結果發生的是，在應用程序的每個位置，處理程序都被添加到服務的事件中，並且該處理程序基於基於用戶輸入設置過濾器的其他代碼來過濾不需要的事件。毫不奇怪，用於過濾和設置過濾器的代碼在任何地方都基本相同。

這似乎是一個通用的PubSub對象坐落在服務和想要數據的地方之間的好地方。我可以根據需要爲每個數據場景設置對象的具體版本，並將它們添加到應用程序中各個位置的服務中以供使用。不想重新發明輪子，我一直在使用PubSub，我能找到的最接近的是主題，但這些不適合這種情況，因爲它需要爲每個產品設置一個主題。

有沒有任何資源可以讓任何人知道？我正在使用C＃，但如果需要，我可以根據其他語言進行調整。主要的要求是這不能是一個單獨的服務或過程，例如WCF;它需要駐留在應用程序中並使用現有的基礎架構，大概是通過附加到服務的事件，然後通知訂戶。另外值得關注的事實是，這個應用程序是多線程的，所以對線程安全的傾向會很有幫助，儘管我可以根據需要進行調整。

在此先感謝！

Answer 1

我會看看DDS產品（數據分配服務），如RTI。他們提供所有你需要的和更多。 見： http://www.rti.com/products/dds/

Answer 2

嗯..不知道如果我理解正確。那樣的事情呢？使用模式

public void handleui(dynamic s) 
{ 
    Application.Current.Dispatcher.Invoke(delegate 
               { 
                btn1.Content = s.ToString(); 
               }); 
} 
Globals.Events["error"] += msg=> Console.WriteLine(msg);//logger perhaps 
Globals.Events["productb"] += handleui;//sub 
Globals.Events["productb"] -= handleui;//unsub  
Globals.Events["productb"].Send("productbdata");//raise the event or publish to productb channel subscribers 
Globals.Events.Send("broadcast?"); 

我會想象你可以做一個單一的過濾器，只需將所有事件活動[「產品A」]，事件[「產品B」]，事件[「productc」]等，以及部分可以當他們想要它時，只需要sub/unsub。

執行。

using System; 
using System.Collections.Concurrent; 


public class Globals 
{ 
    public static MsgBus Events = new MsgBus(); 
} 

public class MsgBus 
{ 
    private readonly ConcurrentDictionary<dynamic, MsgChannel> channels = new ConcurrentDictionary<dynamic, MsgChannel>(); 

    public MsgChannel this[dynamic channel] 
    { 
     set { channels[channel] = value; } 
     get 
     { 
      var ch = (MsgChannel)channels.GetOrAdd(channel, new MsgChannel()); 
      return ch; 
     } 
    } 

    private MsgChannel broadcast = new MsgChannel(); 
    public void Send(dynamic msg) 
    { 
     broadcast.Send(msg); 
    } 

    public static MsgBus operator +(MsgBus left, Action<dynamic> right) 
    { 
     left.broadcast += right; 
     return left; 
    } 

    public static MsgBus operator -(MsgBus left, Action<dynamic> right) 
    { 
     left.broadcast -= right; 
     return left; 
    } 
} 

public class MsgChannel 
{ 
    ConcurrentDictionary<Action<dynamic>, int> observers = new ConcurrentDictionary<Action<dynamic>, int>(); 
    public void Send(dynamic msg) 
    { 
     foreach (var observer in observers) 
     { 
      for (int i = 0; i < observer.Value; i++) 
      { 
       observer.Key.Invoke(msg); 
      } 
     } 
    } 
    public static MsgChannel operator +(MsgChannel left, Action<dynamic> right) 
    { 

     if (!left.observers.ContainsKey(right)) 
     { 
      left.observers.GetOrAdd(right, 0); 
     } 

     left.observers[right]++; 
     return left; 
    } 

    public static MsgChannel operator -(MsgChannel left, Action<dynamic> right) 
    { 
     if (left.observers.ContainsKey(right) && 
      left.observers[right] > 0) 
     { 
      left.observers[right]--; 
      int dummy; 
      if (left.observers[right] <= 0) left.observers.TryRemove(right, out dummy); 
     } 

     return left; 
    } 
}