ZeroMQ Test

Dev/OHVDriveLogger/OHVDriveLogger/OHVDriveLogger/FormMain.Designer.cs

@@ -30,6 +30,7 @@
         {
             this.lblPLCCommState = new System.Windows.Forms.Label();
             this.label55 = new System.Windows.Forms.Label();
+            this.button1 = new System.Windows.Forms.Button();
             this.SuspendLayout();
             // 
             // lblPLCCommState
@@ -59,12 +60,23 @@
             this.label55.Text = "PLC COMM";
             this.label55.TextAlign = System.Drawing.ContentAlignment.MiddleCenter;
             // 
+            // button1
+            // 
+            this.button1.Location = new System.Drawing.Point(154, 20);
+            this.button1.Name = "button1";
+            this.button1.Size = new System.Drawing.Size(124, 28);
+            this.button1.TabIndex = 13;
+            this.button1.Text = "button1";
+            this.button1.UseVisualStyleBackColor = true;
+            this.button1.Click += new System.EventHandler(this.button1_Click);
+            // 
             // FormMain
             // 
             this.AutoScaleDimensions = new System.Drawing.SizeF(7F, 12F);
             this.AutoScaleMode = System.Windows.Forms.AutoScaleMode.Font;
             this.BackColor = System.Drawing.SystemColors.ActiveBorder;
             this.ClientSize = new System.Drawing.Size(419, 72);
+            this.Controls.Add(this.button1);
             this.Controls.Add(this.lblPLCCommState);
             this.Controls.Add(this.label55);
             this.Name = "FormMain";
@@ -77,6 +89,7 @@
 
         public System.Windows.Forms.Label lblPLCCommState;
         private System.Windows.Forms.Label label55;
+        private System.Windows.Forms.Button button1;
     }
 }
 

Dev/OHVDriveLogger/OHVDriveLogger/OHVDriveLogger/FormMain.cs

@@ -17,6 +17,7 @@ namespace OHVDriveLogger
     {
         Logger logger = Logger.GetLogger();
         KncManager plc = new KncManager();
+        ZmqManager zmp = new ZmqManager();
         
         public FormMain()
         {
@@ -29,6 +30,8 @@ namespace OHVDriveLogger
         private void FormMain_FormClosing( object sender, FormClosingEventArgs e )
         {
             this.plc.Disconnect();
+
+            this.zmp.Dispose();
         }
 
         private void FormMain_Load( object sender, EventArgs e )
@@ -49,6 +52,8 @@ namespace OHVDriveLogger
             this.plc.OnDisconnect += Plc_OnDisconnect;
 
             this.plc.Connect();
+
+            this.zmp.CrateClientSocket();
         }
 
         private void Plc_OnDisconnect( string id )
@@ -67,5 +72,10 @@ namespace OHVDriveLogger
         {
             logger.I( $"[Bit Change] - {block.Name} / {block.IsBitOn}" );
         }
+
+        private void button1_Click( object sender, EventArgs e )
+        {
+            this.zmp.Request("10");
+        }
     }
 }

Dev/OHVDriveLogger/OHVDriveLogger/OHVDriveLogger/OHVDriveLogger.csproj

@@ -45,7 +45,23 @@
       <HintPath>..\..\..\OHV\Assambly\NetMq\NetMQ.dll</HintPath>
     </Reference>
     <Reference Include="System" />
+    <Reference Include="System.Buffers, Version=4.0.3.0, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51, processorArchitecture=MSIL">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>..\..\..\OHV\Assambly\NetMq\System.Buffers.dll</HintPath>
+    </Reference>
     <Reference Include="System.Core" />
+    <Reference Include="System.Memory, Version=4.0.1.1, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51, processorArchitecture=MSIL">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>..\..\..\OHV\Assambly\NetMq\System.Memory.dll</HintPath>
+    </Reference>
+    <Reference Include="System.Runtime.CompilerServices.Unsafe, Version=4.0.4.1, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>..\..\..\OHV\Assambly\NetMq\System.Runtime.CompilerServices.Unsafe.dll</HintPath>
+    </Reference>
+    <Reference Include="System.ValueTuple, Version=4.0.0.0, Culture=neutral, PublicKeyToken=cc7b13ffcd2ddd51, processorArchitecture=MSIL">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>..\..\..\OHV\Assambly\NetMq\System.ValueTuple.dll</HintPath>
+    </Reference>
     <Reference Include="System.Xml.Linq" />
     <Reference Include="System.Data.DataSetExtensions" />
     <Reference Include="Microsoft.CSharp" />

Dev/OHVDriveLogger/OHVDriveLogger/OHVDriveLogger/ZmqManager.cs

@@ -17,6 +17,10 @@ namespace OHVDriveLogger
         SubscriberSocket sub = null;
         RequestSocket req = null;
 
+        PublisherSocket pub = null;
+
+        NetMQPoller poller = null;
+
         ThreadCancel threadCancel = new ThreadCancel();
 
         public ZmqManager()
@@ -24,16 +28,34 @@ namespace OHVDriveLogger
             NetMQ.NetMQConfig.Cleanup();
         }
 
-        void CrateClientSocket()
+        public void CrateClientSocket()
         {
+            //pub = new PublisherSocket();
+            //pub.Bind( "tcp://127.0.0.1:5565" );
+
             sub = new SubscriberSocket();
             sub.Connect( "tcp://127.0.0.1:5565" );
-            sub.Subscribe(string.Empty); //All
+            sub.Subscribe(""); //All
 
             sub.ReceiveReady += Sub_ReceiveReady;
 
             req = new RequestSocket();
             req.Connect( "tcp://127.0.0.1:5567" );
+
+            this.poller = new NetMQPoller { this.sub };
+            this.poller.RunAsync();
+
+            //this.threadCancel.AddGo( Th_SubPoller );
+        }
+
+        public void Dispose()
+        {
+            this.poller.Dispose();
+            //this.pub.Dispose();
+            this.sub.Dispose();
+            this.req.Dispose();
+
+            this.threadCancel.Cancel();
         }
 
         private void Sub_ReceiveReady( object sender, NetMQ.NetMQSocketEventArgs e )
@@ -41,14 +63,32 @@ namespace OHVDriveLogger
             logger.I( e.Socket.ReceiveMultipartStrings() );
         }
 
-        string Request(string topic )
+        void Th_SubPoller()
+        {
+            while ( !this.threadCancel.Canceled )
+            {
+                LockUtils.Wait( 1000 );
+                this.pub.SendMoreFrame( "1000" ).SendFrame( "Test" );
+                LockUtils.Wait( 100 );
+
+                NetMQMessage msg = new NetMQMessage();
+                if ( this.sub.TryReceiveMultipartMessage( TimeSpan.FromSeconds( 1 ), ref msg ) )
+                {
+                    var m = msg;
+                }
+            }
+        }
+
+        public string Request(string topic )
         {
             this.req.SendFrame( topic );
 
             if ( this.req.Poll( TimeSpan.FromSeconds( 5 ) ) )
             {
-
+                return this.req.ReceiveFrameString();
             }
+            else
+                return string.Empty;
         }
     }
 }

Dev/OHVLoggerTest/OHVLoggerTest/Form1.cs

@@ -25,13 +25,15 @@ namespace OHVLoggerTest
         Logger logger = Logger.GetLogger();
 
         ThreadCancel threadCancel = new ThreadCancel();
-        KV8000 plc = new KV8000();
 
         List<point> Points = null;
         List<segment> Segments = null;
 
         Graph<int, string> graph = new Graph<int, string>();
 
+        ZmqManager zmq = null;
+        NetMq mq = null;
+
         public Form1()
         {
             InitializeComponent();
@@ -55,6 +57,9 @@ namespace OHVLoggerTest
 
             this.threadCancel.AddGo( Thread_Dowork );
             this.threadCancel.AddGo( Thread_DriveRun );
+
+            this.zmq = new ZmqManager();
+            //this.mq = new NetMq();
         }
 
         void Thread_Dowork()

Dev/OHVLoggerTest/OHVLoggerTest/NetMq.cs

@@ -0,0 +1,37 @@
+using GSG.NET.Concurrent;
+using NetMQ;
+using NetMQ.Sockets;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace OHVLoggerTest
+{
+    public class NetMq
+    {
+        PublisherSocket pub = null;
+        ThreadCancel threadCancel = new ThreadCancel();
+
+        public NetMq()
+        {
+            pub = new PublisherSocket();
+            pub.Bind( "tcp://127.0.0.1:5565" );
+
+            threadCancel.AddGo( Th_Pub );
+        }
+
+        void Th_Pub()
+        {
+            while ( !this.threadCancel.Canceled )
+            {
+                LockUtils.Wait( 1000 );
+                this.pub.SendMoreFrame( "1000" ).SendFrame( "Test" );
+                LockUtils.Wait( 100 );
+            }
+        }
+
+
+    }
+}

Dev/OHVLoggerTest/OHVLoggerTest/OHVLoggerTest.csproj

@@ -47,6 +47,9 @@
     <Reference Include="log4net">
       <HintPath>..\..\OHV\Assambly\log4net.dll</HintPath>
     </Reference>
+    <Reference Include="NetMQ">
+      <HintPath>..\..\OHV\Assambly\NetMq\NetMQ.dll</HintPath>
+    </Reference>
     <Reference Include="Newtonsoft.Json, Version=6.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
       <SpecificVersion>False</SpecificVersion>
       <HintPath>..\..\OHV\Assambly\Newtonsoft.Json.dll</HintPath>
@@ -72,6 +75,9 @@
     <Reference Include="System.Net.Http" />
     <Reference Include="System.Windows.Forms" />
     <Reference Include="System.Xml" />
+    <Reference Include="ZeroMQ">
+      <HintPath>..\..\..\..\..\Referance_Project\ZeroMQ\clrzmq4-master\bin\Debug\ZeroMQ.dll</HintPath>
+    </Reference>
   </ItemGroup>
   <ItemGroup>
     <Compile Include="Form1.cs">
@@ -80,9 +86,11 @@
     <Compile Include="Form1.Designer.cs">
       <DependentUpon>Form1.cs</DependentUpon>
     </Compile>
+    <Compile Include="NetMq.cs" />
     <Compile Include="Program.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
     <Compile Include="Redis.cs" />
+    <Compile Include="ZmqManager.cs" />
     <EmbeddedResource Include="Form1.resx">
       <DependentUpon>Form1.cs</DependentUpon>
     </EmbeddedResource>

Dev/OHVLoggerTest/OHVLoggerTest/ZmqManager.cs

@@ -0,0 +1,98 @@
+using GSG.NET.Concurrent;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using ZeroMQ;
+
+namespace OHVLoggerTest
+{
+    public class ZmqManager
+    {
+        ZContext ctx = new ZContext();
+        ZSocket rep = null;
+        ZSocket pub = null;
+        ZSocket sub = null;
+
+        ThreadCancel threadCancel = new ThreadCancel();
+
+        public ZmqManager()
+        {
+            CreateZeroMQ();
+        }
+
+        void CreateZeroMQ()
+        {
+            this.pub = new ZSocket( ctx, ZSocketType.PUB );
+            //this.pub.Bind( "tcp://127.0.0.1:5566" );
+            this.pub.Bind( "tcp://127.0.0.1:5565" );
+
+            this.sub = new ZSocket( ctx, ZSocketType.SUB );
+            this.sub.Connect( "tcp://127.0.0.1:5565" );
+            this.sub.SubscribeAll();
+
+            this.rep = new ZSocket( ctx, ZSocketType.REP );
+            rep.Bind( "tcp://127.0.0.1:5567" );
+
+            this.threadCancel.AddGo( Th_Respens );
+            this.threadCancel.AddGo( Th_Publish );
+        }
+
+        void Th_Respens()
+        {
+            while ( !this.threadCancel.Canceled )
+            {
+                ZMessage message;
+                ZError error;
+                if ( null != (message = this.sub.ReceiveMessage(out error ) ))
+                {
+                    using ( message )
+                    {
+                        Console.WriteLine( $"{message[0].ReadString()} / {message[1].ReadString()}" );
+                    }
+                }
+                else
+                {
+                    if ( error == ZError.ETERM )
+                        return; // Interrupted
+                    throw new ZException( error );
+                }
+
+                //if ( null != ( message = this.rep.ReceiveMessage( out error ) ) )
+                //{
+                //    using ( message )
+                //    {
+                //        Console.WriteLine( $"{message}" );
+                //        rep.Send( new ZFrame( "rep" ) );
+                //    }
+                //}
+                //else
+                //{
+                //    if ( error == ZError.ETERM )
+                //        return; // Interrupted
+                //    throw new ZException( error );
+                //}
+            }
+        }
+
+        void Th_Publish()
+        {
+            while ( !this.threadCancel.Canceled )
+            {
+                LockUtils.Wait( 1000 );
+                var msg = new ZMessage();
+                msg.Add( new ZFrame( "1000" ) );
+                msg.Add( new ZFrame( "Test" ) );
+                this.pub.Send( msg );
+                //this.pub.SendMore( new ZFrame( "1000" ) );
+                //this.pub.Send( new ZFrame( "Test Publish" ) );
+            }
+        }
+
+        void Dispese()
+        {
+            this.threadCancel.Cancel();
+        }
+    }
+}