c# - Azure IoT 中心中的 TCP 支持

Question

Azure IoT 中心支持 AMQP、MQTT、HTTP 协议(protocol)。为了自定义这些协议(protocol)，我们有 Azure IoT 协议(protocol)网关。我可以找到关于 MQTT 协议(protocol)定制的好示例。我需要一些使用 Azure IoT 协议(protocol)网关进行基于 TCP 的协议(protocol)自定义的示例代码。

编辑(为了得到答案):OP所询问的是一个使用Azure协议(protocol)网关支持基于TCP的专有协议(protocol)的示例。目前 IoT 中心仅支持 AMQP、MQTT 和 HTTP。虽然这些协议(protocol)实际上依赖于 TCP，但如果没有额外的 AMQP、MQTT 或 HTTP 层，集线器不支持直接 TCP 连接。正如所解释的here ，我们需要一个基于自定义 TCP 协议(protocol)的基本示例。

想象一个基本设备，它只能在给定的 IP 地址/端口上通过 TCP 发送一些专有的有效负载:我们需要一个网关自定义示例，允许该设备将数据发送到集线器。

协议(protocol)网关当前代码设计不佳，严重依赖 MQTT。

还添加一些赏金。

Answer 1

由于所有 MQTT 代码，默认协议(protocol)网关示例确实有些令人困惑。协议(protocol)网关的工作原理是为连接到网关的每个自定义协议(protocol)设备“模拟”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 回调方法应按如下所示和 this line 中所示实现。在 Github 的 ProtocolGateway 存储库中。

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

当 Tcp 设备连接到该协议(protocol)时，您应该创建此 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);
}