OHV/Dev/OHVLoggerTest/OHVLoggerTest/ZmqManager.cs

using GSG.NET.Concurrent;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using ZeroMQ;

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