devbox
盐析白兔
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

基于IPC-CFX的点对点通信C#_ipc cfx

作者：盐析白兔 | 2024-07-18 21:20:03

赞

踩

ipc cfx

        IPC-CFX有两种主要的通信方式，可以通过RabbitMQ发布和订阅，也可以通过request和response进行点对点的通信，本文主要讲的是点对点的通信方式。

        在vscode里建立新的dotnet项目，可以通过终端输入dotnet new console来建立，文件目录为CFXDemo->machine1和CFXDemo->machine2。

        通过nuget插件分别为两个项目都安装CFX.CFXSDK、AMQPNetLite.Core和Newtonsoft.Json这几个metapackage。

         我们将machine1作为发送端（sendRequest)，machine2作为接收端(Receive)，则Machine1的代码如下所示：

  1. using System.Threading;
  2. using CFX;
  3. using CFX.Transport;
  4. using System;
  5. using System.Security.Principal;
  6.  
  7. namespace machine1
  8. {
  9.     class Program
  10.     {
  11.  
  12.         static void Main(string[] args)
  13.         {
  14.  
  15.             OpenRequest();
  16.             Console.ReadLine();
  17.             for(int i = 0;i<5;i++){
  18.                 SendRequest();
  19.                 Thread.Sleep(2000);
  20.             }
  21.             
  22.         }
  23.  
  24.         static string sendCFXHandle = "a.b.001";     
  25.         static string receiveCFXHandle = "a.b.002";
  26.         static string sendRequestUri = string.Format("amqp://127.0.0.1:1235");
  27.         static string receivRequestUri = string.Format("amqp://127.0.0.1:1234");
  28.  
  29.  
  30.         #region send request
  31.  
  32.         static AmqpCFXEndpoint endpointSendRequest;
  33.  
  34.         static void OpenRequest()
  35.         {
  36.  
  37.  
  38.             if (endpointSendRequest != null)
  39.             {
  40.                 endpointSendRequest.Close();
  41.                 endpointSendRequest = null;
  42.  
  43.             }
  44.  
  45.             endpointSendRequest = new AmqpCFXEndpoint();
  46.  
  47.             if (!endpointSendRequest.IsOpen)
  48.             {
  49.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointSendRequest.IsOpen.ToString());
  50.                 endpointSendRequest.Open(sendCFXHandle);    //这一步会绑定endpointSendRequest里的CFXHandle，即sendCFXHandle的值
  51.                 Console.WriteLine("Request.Source is : {0}",endpointSendRequest.CFXHandle);
  52.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointSendRequest.IsOpen.ToString());
  53.  
  54.             }
  55.  
  56.             // Set a timeout of 20 seconds.  If the target endpoint does not
  57.             // respond in this time, the request will time out.
  58.             AmqpCFXEndpoint.RequestTimeout = TimeSpan.FromSeconds(20);
  59.  
  60.         }
  61.         static void SendRequest()
  62.         {
  63.  
  64.             // Build a GetEndpointInfomation Request
  65.             CFXEnvelope request = CFXEnvelope.FromCFXMessage(new GetEndpointInformationRequest()
  66.             {
  67.                 CFXHandle = receiveCFXHandle
  68.             });
  69.             request.Source = endpointSendRequest.CFXHandle;
  70.             request.Target = receiveCFXHandle;
  71.  
  72.             try
  73.             {
  74.                 CFXEnvelope response = endpointSendRequest.ExecuteRequest(receivRequestUri, request);
  75.                 Console.WriteLine($"response:\n{response.ToJson(true)}");
  76.             }
  77.             catch (Exception ex)
  78.             {
  79.                 Console.WriteLine(ex.Message);
  80.             }
  81.         }
  82.  
  83.         #endregion send request
  84.  
  85.  
  86.         #region receive request
  87.         static AmqpCFXEndpoint endpointReceiveRequest;
  88.  
  89.         static void OpenReceive()
  90.         {
  91.             if (endpointReceiveRequest != null)
  92.             {
  93.                 endpointReceiveRequest.Close();
  94.                 endpointReceiveRequest = null;
  95.             }
  96.  
  97.             endpointReceiveRequest = new AmqpCFXEndpoint();
  98.  
  99.             endpointReceiveRequest.OnRequestReceived -= Endpoint_OnRequestReceived;
  100.             endpointReceiveRequest.OnRequestReceived += Endpoint_OnRequestReceived;
  101.  
  102.             if (!endpointReceiveRequest.IsOpen)
  103.             {
  104.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointReceiveRequest.IsOpen.ToString());
  105.                 endpointReceiveRequest.Open(receiveCFXHandle, new Uri(receivRequestUri));
  106.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointReceiveRequest.IsOpen.ToString());
  107.             }
  108.         }
  109.  
  110.  
  111.         static CFXEnvelope Endpoint_OnRequestReceived(CFXEnvelope request)
  112.         {
  113.             Console.WriteLine($"Endpoint_OnRequestReceived:  { request.ToString()}");
  114.  
  115.             // Process request.  Return Result.
  116.             if (request.MessageBody is WhoIsThereRequest)
  117.             {
  118.                 CFXEnvelope result = CFXEnvelope.FromCFXMessage(new WhoIsThereResponse()
  119.                 { CFXHandle = receiveCFXHandle, RequestNetworkUri = receivRequestUri, RequestTargetAddress = "" });
  120.                 result.Source = receiveCFXHandle;
  121.                 result.Target = request.Source;
  122.                 result.TimeStamp = DateTime.Now;
  123.                 return result;
  124.             }
  125.  
  126.             if (request.MessageBody is GetEndpointInformationRequest)
  127.             {
  128.                 CFXEnvelope result = CFXEnvelope.FromCFXMessage(new WhoIsThereResponse()
  129.                 { CFXHandle = receiveCFXHandle, RequestNetworkUri = receivRequestUri, RequestTargetAddress = "...." });
  130.                 result.Source = receiveCFXHandle;
  131.                 result.Target = request.Source;
  132.                 result.TimeStamp = DateTime.Now;
  133.                 return result;
  134.             }
  135.  
  136.             return null;
  137.         }
  138.  
  139.         #endregion receive request
  140.  
  141.  
  142.  
  143.  
  144.     }
  145. }

        作为接收端，machine2的代码如下所示：

  1. using CFX;
  2. using CFX.Transport;
  3. using System;
  4.  
  5. namespace machine2
  6. {
  7.     class Program
  8.     {
  9.  
  10.  
  11.         static void Main(string[] args)
  12.         {
  13.  
  14.             Console.WriteLine("ReceivEndPoint is waiting Request......");
  15.             OpenReceive();
  16.             Console.WriteLine("Press Enter Key to end the App");
  17.             Console.ReadKey();
  18.  
  19.         }
  20.  
  21.         static string sendCFXHandle = "a.b.001";
  22.         static string receiveCFXHandle = "a.b.002";
  23.         static string sendRequestUri = string.Format("amqp://127.0.0.1:1235");
  24.         static string receivRequestUri = string.Format("amqp://127.0.0.1:1234");
  25.  
  26.  
  27.         #region send request
  28.  
  29.         static AmqpCFXEndpoint endpointSendRequest;
  30.  
  31.         static void OpenRequest()
  32.         {
  33.  
  34.  
  35.             if (endpointSendRequest != null)
  36.             {
  37.                 endpointSendRequest.Close();
  38.                 endpointSendRequest = null;
  39.  
  40.             }
  41.  
  42.             endpointSendRequest = new AmqpCFXEndpoint();
  43.  
  44.             if (!endpointSendRequest.IsOpen)
  45.             {
  46.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointSendRequest.IsOpen.ToString());
  47.                 endpointSendRequest.Open(receiveCFXHandle, new Uri(receivRequestUri));
  48.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointSendRequest.IsOpen.ToString());
  49.  
  50.             }
  51.  
  52.             // Set a timeout of 20 seconds.  If the target endpoint does not
  53.             // respond in this time, the request will time out.
  54.             AmqpCFXEndpoint.RequestTimeout = TimeSpan.FromSeconds(20);
  55.  
  56.         }
  57.         static void SendRequest()
  58.         {
  59.  
  60.             // Build a GetEndpointInfomation Request
  61.             CFXEnvelope request = CFXEnvelope.FromCFXMessage(new GetEndpointInformationRequest()
  62.             {
  63.                 CFXHandle = receiveCFXHandle
  64.             });
  65.             request.Source = endpointSendRequest.CFXHandle;
  66.             request.Target = receiveCFXHandle;
  67.  
  68.             try
  69.             {
  70.                 CFXEnvelope response = endpointSendRequest.ExecuteRequest(receivRequestUri, request);
  71.                 Console.WriteLine($"response:\n{response.ToJson(true)}");
  72.             }
  73.             catch (Exception ex)
  74.             {
  75.                 Console.WriteLine(ex.Message);
  76.             }
  77.         }
  78.  
  79.         #endregion send request
  80.  
  81.  
  82.         #region receive request
  83.         static AmqpCFXEndpoint endpointReceiveRequest;
  84.  
  85.         static void OpenReceive()
  86.         {
  87.             if (endpointReceiveRequest != null)
  88.             {
  89.                 endpointReceiveRequest.Close();
  90.                 endpointReceiveRequest = null;
  91.             }
  92.  
  93.             endpointReceiveRequest = new AmqpCFXEndpoint();
  94.  
  95.             endpointReceiveRequest.OnRequestReceived -= Endpoint_OnRequestReceived;
  96.             endpointReceiveRequest.OnRequestReceived += Endpoint_OnRequestReceived;
  97.  
  98.             if (!endpointReceiveRequest.IsOpen)
  99.             {
  100.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointReceiveRequest.IsOpen.ToString());
  101.                 endpointReceiveRequest.Open(receiveCFXHandle, new Uri(receivRequestUri));
  102.                 //endpointReceiveRequest.Open(receiveCFXHandle);
  103.                 Console.WriteLine("endpointSendRequest.IsOpen:" + endpointReceiveRequest.IsOpen.ToString());
  104.             }
  105.         }
  106.  
  107.  
  108.         static CFXEnvelope Endpoint_OnRequestReceived(CFXEnvelope request)
  109.         {
  110.             Console.WriteLine($"Endpoint_OnRequestReceived:  { request.ToString()}");
  111.             Console.WriteLine($"request:\n{request.ToJson(true)}");
  112.  
  113.             // Process request.  Return Result.
  114.             if (request.MessageBody is WhoIsThereRequest)
  115.             {
  116.                 CFXEnvelope result = CFXEnvelope.FromCFXMessage(new WhoIsThereResponse()
  117.                 { CFXHandle = receiveCFXHandle, RequestNetworkUri = receivRequestUri, RequestTargetAddress = "" });
  118.                 result.Source = receiveCFXHandle;
  119.                 result.Target = request.Source;
  120.                 result.TimeStamp = DateTime.Now;
  121.                 return result;
  122.             }
  123.  
  124.             if (request.MessageBody is GetEndpointInformationRequest)
  125.             {
  126.                 CFXEnvelope result = CFXEnvelope.FromCFXMessage(new WhoIsThereResponse()
  127.                 { CFXHandle = receiveCFXHandle, RequestNetworkUri = receivRequestUri, RequestTargetAddress = "..." });
  128.                 result.Source = receiveCFXHandle;
  129.                 result.Target = request.Source;
  130.                 result.TimeStamp = DateTime.Now;
  131.                 return result;
  132.             }
  133.  
  134.             return null;
  135.         }
  136.  
  137.         #endregion receive request
  138.  
  139.  
  140.  
  141.  
  142.     }
  143. }

         运行结果，可以用json格式对response和request的内容进行解析。

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/盐析白兔/article/detail/847780
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号