change Pub/Sub

IEC60870/IEC60870.csproj

@@ -31,10 +31,6 @@
     <WarningLevel>4</WarningLevel>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="PubSub, Version=1.4.1.0, Culture=neutral, processorArchitecture=MSIL">
-      <HintPath>..\packages\PubSub.1.4.1\lib\portable-net45+sl5+netcore45+wpa81+wp8+MonoAndroid1+MonoTouch1\PubSub.dll</HintPath>
-      <Private>True</Private>
-    </Reference>
     <Reference Include="System" />
     <Reference Include="System.Core" />
     <Reference Include="System.Xml.Linq" />
@@ -102,12 +98,11 @@
     <Compile Include="SAP\ServerSAP.cs" />
     <Compile Include="Util\CountDownLatch.cs" />
     <Compile Include="Util\PeriodicTaskFactory.cs" />
+    <Compile Include="Util\PubSubExtensions.cs" />
+    <Compile Include="Util\PubSubHub.cs" />
     <Compile Include="Util\ThreadBase.cs" />
   </ItemGroup>
   <ItemGroup />
-  <ItemGroup>
-    <None Include="packages.config" />
-  </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.

IEC60870/SAP/ServerSAP.cs

@@ -5,7 +5,6 @@
 using System.IO;
 using System.Net;
 using System.Net.Sockets;
-using PubSub;
 
 namespace IEC60870.SAP
 {
@@ -19,25 +18,25 @@ public ConnectionHandler(Socket socket, ConnectionSettings settings): base()
             _socket = socket;
             _settings = settings;
 
-            this.Subscribe<ASdu>(asdu =>
+            this.Subscribe<ASdu>("send", asdu =>
             {
                 try
                 {
                     serverConnection.Send(asdu);
                 }
                 catch (Exception e)
                 {
-                    Console.WriteLine(e);
+                    this.Publish("error", e);
                 }               
             });
         }
 
         public override void Run()
         {
             serverConnection = new Connection(_socket, _settings);
-            serverConnection.ConnectionClosed += (a) =>
+            serverConnection.ConnectionClosed += e =>
             {
-                Console.WriteLine(a);
+                this.Publish<Exception>("error", e);
             };
 
             serverConnection.WaitForStartDT(5000);
@@ -73,11 +72,11 @@ public override void Run()
                     }
                     catch (IOException e)
                     {
-                        Console.WriteLine(e);
+                        this.Publish<Exception>("error", e);
                     }   
                     catch (Exception e)
                     {
-                        Console.WriteLine(e);
+                        this.Publish("error", e);
                         break;
                     }                
                 }
@@ -132,7 +131,7 @@ public void StartListen(int backlog)
 
         public void SendASdu(ASdu asdu)
         {
-            this.Publish(asdu);
+            this.Publish("send", asdu);
         }
 
         public void SetMessageFragmentTimeout(int timeout)

IEC60870/Util/PubSubExtensions.cs

@@ -0,0 +1,19 @@
+using System;
+
+namespace IEC60870.Util
+{
+    static public class PubSubExtensions
+    {
+        static private readonly PubSubHub hub = new PubSubHub();
+
+        static public void Publish<T>(this object obj, string topic, T data)
+        {
+            hub.Publish(obj, topic, data);
+        }
+
+        static public void Subscribe<T>(this object obj, string topic, Action<T> handler)
+        {
+            hub.Subscribe(obj, topic, handler);
+        }
+    }
+}

IEC60870/Util/PubSubHub.cs

@@ -0,0 +1,66 @@
+using System;
+using System.Collections.Generic;
+
+namespace IEC60870.Util
+{
+    public class PubSubHub
+    {
+        internal class Handler
+        {
+            public Delegate Action { get; set; }
+            public WeakReference Sender { get; set; }
+            public Type Type { get; set; }
+        }
+
+        internal object locker = new object();
+        internal Dictionary<string, List<Handler>> handlers = new Dictionary<string, List<Handler>>();
+
+        public void Publish<T>(object sender, string topic, T data)
+        {
+            if (handlers.ContainsKey(topic))
+            {
+                var listHandler = handlers[topic];
+                listHandler.ForEach(handler =>
+                {
+                    if (handler.Sender.IsAlive)
+                    {
+                        ((Action<T>)handler.Action)(data);
+                    }
+                    else
+                    {
+                        lock (locker)
+                        {
+                            handlers[topic].Remove(handler);
+                        }
+                    }
+                });
+            }
+        }
+
+        public void Subscribe<T>(object sender, string topic, Action<T> handler)
+        {
+            var item = new Handler
+            {
+                Action = handler,
+                Sender = new WeakReference(sender),
+                Type = typeof(T)
+            };
+
+            if (handlers.ContainsKey(topic))
+            {
+                lock (locker)
+                {
+                    handlers[topic].Add(item);
+                }
+            }
+            else
+            {
+                lock (locker)
+                {
+                    handlers[topic] = new List<Handler>();
+                    handlers[topic].Add(item);
+                }
+            }
+        }
+    }
+}