Azure IoT Hub中的TCP支持

1 回答

• 2

  由于所有MQTT代码，默认的协议网关示例确实有些令人困惑 . 协议网关通过为连接到网关的每个自定义协议设备“模拟”IoTHub连接来工作 .

  要从TCP设备到IoTHub设备执行此转换，首先需要代表设备连接到IoTHub . 这是网关部分 . 以下是此IoTHubConnection的核心要点 .

  namespace GatewayTest
  {
      using System;
      using System.Text;
      using System.Threading;
      using System.Threading.Tasks;
      using DotNetty.Buffers;
      using Microsoft.Azure.Devices.ProtocolGateway.Identity;
      using Microsoft.Azure.Devices.ProtocolGateway.IotHubClient;
      using Microsoft.Azure.Devices.ProtocolGateway.Messaging;
  
      public class IoTHubConnection : IMessagingChannel<IMessage>
      {
          private readonly string iotHubHostName;
          private readonly Func<IDeviceIdentity, Task<IMessagingServiceClient>> deviceClientFactory;
          private readonly Func<string, Task> onMessage;
          private IMessagingServiceClient deviceClient;
          private IDeviceIdentity deviceIdentity;
  
          public IoTHubConnection(
              string iotHubHostName,
              Func<IDeviceIdentity, Task<IMessagingServiceClient>> deviceClientFactory,
              Func<string, Task> onMessage)
          {
              this.iotHubHostName = iotHubHostName;
              this.deviceClientFactory = deviceClientFactory;
              this.onMessage = onMessage;
          }
  
          public event EventHandler CapabilitiesChanged;
  
          public async Task OpenAsync(string deviceId, string deviceKey)
          {
              this.deviceIdentity = this.GetDeviceIdentity(deviceId, deviceKey);
              if (this.deviceIdentity != UnauthenticatedDeviceIdentity.Instance)
              {
                  this.deviceClient = await this.deviceClientFactory(this.deviceIdentity);
                  this.deviceClient.BindMessagingChannel(this);
              }
          }
  
          public async Task CloseAsync()
          {
              await this.deviceClient.DisposeAsync(null);
              this.deviceClient = null;
          }
  
          public void Handle(IMessage message)
          {
              var messageBody = message.Payload.ToString(Encoding.UTF8);
  
              this.onMessage(messageBody);
  
              this.deviceClient.CompleteAsync(message.Id);
          }
  
          public Task SendMessage(string message)
          {
              var buffer = Unpooled.WrappedBuffer(Encoding.UTF8.GetBytes(message));
              var deviceMessage = this.deviceClient.CreateMessage($"devices/{this.deviceIdentity.Id}/messages/events", buffer);
              return this.deviceClient.SendAsync(deviceMessage);
          }
  
          protected virtual void OnCapabilitiesChanged(EventArgs e)
          {
              this.CapabilitiesChanged?.Invoke(this, e);
          }
  
          private IDeviceIdentity GetDeviceIdentity(string userName, string deviceKey)
          {
              IotHubDeviceIdentity ideviceIdentity;
              if (!IotHubDeviceIdentity.TryParse($"{this.iotHubHostName}/{userName}", out ideviceIdentity))
              {
                  return UnauthenticatedDeviceIdentity.Instance;
              }
  
              ideviceIdentity.WithDeviceKey(deviceKey);
              return ideviceIdentity;
          }
      }
  }
  

  deviceClientFactory回调方法应该如下所示实现，并在Github的ProtocolGateway repo中的this line中实现 .

  deviceClientFactory = IotHubClient.PreparePoolFactory(
      "IotHubConnectionString",
      400,
      TimeSpan.FromMinutes(3),
      iotHubClientSettings,
      PooledByteBufferAllocator.Default,
      new ConfigurableMessageAddressConverter("TopicNameConversion"));
  

  当Tcp设备连接到协议时，您应该创建此IoTHubConnection的实例，并将消息从设备发送到IoTHubConnection，反之亦然 . 下面的代码显示了如何完成此操作的非常简单的版本 .

  private const int BufferSize = 1024;
  private byte[] buffer = new byte[BufferSize];
  private IoTHubConnection ioTHubConnection;
  private NetworkStream stream;
  
  private async Task Start()
  {
      listener = new TcpListener(IPAddress.Any, port);
      listener.Start();
  
      var client = await listener.AcceptTcpClientAsync();
      ioTHubConnection = new IoTHubConnection("IoTHubName", deviceClientFactory, OnIoTHubMessage);
      stream = client.GetStream();
  
      // Read DeviceId and DeviceKey from some sort of StartConnection-message send by the TcpClient.
      await ioTHubConnection.OpenAsync("DeviceId", "DeviceKey");
  
      stream.BeginRead(buffer, 0, BufferSize, ReadTcpStreamCallback, null);
  }
  
  private void ReadTcpStreamCallback(IAsyncResult ar)
  {
      var bytesRead = stream.EndRead(ar);
  
      if (bytesRead > 0)
      {
          var message = System.Text.Encoding.ASCII.GetString(result);
  
          ioTHubConnection.SendMessage(message);
  
          // Read again.
          stream.BeginRead(buffer, 0, BufferSize, ReadTcpStreamCallback, null);
      }
  }
  
  private async Task OnIoTHubMessage(string message)
  {
      // Potentially do some translation on the IoTHub message
      // and send it to the Device
  
      var byteData = Encoding.UTF8.GetBytes(message);
      stream.BeginWrite(byteData, 0, byteData.Length, SendTcpCallback, null);
  }
  
  private void SendTcpCallback(IAsyncResult ar)
  {
      stream.EndWrite(ar);
  }
  

  回复于 2024-04-24T15:01:06+08:00