Consume messages from the Event Hub

To consume messages from the Event Hub:

1. Activate a group inIoT. The Lifecycle, Twin and Error Event Hub endpoints will be returned in the response.

   Response

   deliveryEndpoints": {
    "lifecycle": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-lifecycle",
    "shadow": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-twin",
    "error": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-error"
    }

2. Activate devices in IoT.

3. Create Routes for the group. The Event Hub endpoint will be returned in the response.

   deliveryEndpoints": {
    "lifecycle": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-lifecycle",
    "shadow": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-twin",
    "error": "iot-ea5d610b45dc4fc89da1f5f4a96ba3a-error"
    }

4. Retrieve Access keys using the Access Keys API : AccessKeys API allows to retrieve credentials for accessing the delivery endpoints. There are two types of credentials:

   • Session Token: This credential will expire in one hour.

   {
     "namespace": "testhub.servicebus.net",
     "expiryUtc": "2020-04-06T11:37:53Z",
     "sessionToken": "SharedAccessSignature sr=test.servicebus.windows.net&sig=rrFiZgryeieVdvpbkiU3m%2frzpI3xIr88pjLmzgvw%3d&se=1644219935&skn=ListenPolicy"
     }

   • Connection String

   {
     "connectionString": "Endpoint=sb://tcp-iot-cgtest.servicebus.windows.net/;SharedAccessKeyName=ListenPolicy;SharedAccessKey=0HYhWKVpHjNYb54eGcFUv/09tC4bYf/AJE2VZV5cpSU="
     }

5. Create Event Hub client using the above credentials and receive messages from the event hub.

   • Refer to the code provided in languages: C# , Python, and Java.

   • Sample Kafka Consumer

   using Confluent.Kafka;
   using System.Configuration;
   
   //Please add the below configurations in AppConfig file
   string? brokerList = ConfigurationManager.AppSettings["<EventHubEndpoint>"];
   string? connectionString = ConfigurationManager.AppSettings["<EventHubConnectionString>"];
   string? topic = ConfigurationManager.AppSettings["<EventHubName>"];
   string? caCertLocation = ConfigurationManager.AppSettings["Server_Ca_Path"];
   string? consumerGroup = ConfigurationManager.AppSettings["<ConsumerGroupName>"];
   
   //Console.WriteLine("Initializing Producer");
   //Worker.Producer(brokerList, connectionString, topic, caCertLocation).Wait();
   //Console.WriteLine();
   Console.WriteLine("Initializing Consumer");
   Worker.Consumer(brokerList, connectionString, consumerGroup, topic, caCertLocation);
   Console.ReadKey();
   
   class Worker
   {
       public static void Consumer(string? brokerList, string? connStr, string? consumergroup, string? topic, string? cacertlocation)
       {
           var config = new ConsumerConfig
           {
               BootstrapServers = brokerList,
               SecurityProtocol = SecurityProtocol.SaslSsl,
               SocketTimeoutMs = 60000,                //this corresponds to the Consumer config `request.timeout.ms`
               SessionTimeoutMs = 30000,
               SaslMechanism = SaslMechanism.Plain,
               SaslUsername = "$ConnectionString",
               SaslPassword = connStr,
               SslCaLocation = cacertlocation,
               GroupId = consumergroup,
               AutoOffsetReset = AutoOffsetReset.Earliest,
               BrokerVersionFallback = "1.0.0",
           };
   
           using (var consumer = new ConsumerBuilder<string, string>(config).SetKeyDeserializer(Deserializers.Utf8).SetValueDeserializer(Deserializers.Utf8).Build())
           {
               CancellationTokenSource cts = new CancellationTokenSource();
               Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };
   
               consumer.Subscribe(topic);
   
               Console.WriteLine("Consuming messages from topic: " + topic + ", broker(s): " + brokerList);
   
               while (true)
               {
                   try
                   {
                       var msg = consumer.Consume(cts.Token);
                       if (msg != null)
                       {
                           Console.WriteLine($"Received: '{msg.Message.Value}'");
                       }
                   }
                   catch (ConsumeException e)
                   {
                       Console.WriteLine($"Consume error: {e.Error.Reason}");
                   }
                   catch (Exception e)
                   {
                       Console.WriteLine($"Error: {e.Message}");
                   }
               }
           }
       }
   }

6. Optionally, consumer groups can be created and the name of the consumer group will be returned in the response. This name can be used while creating the Event Hub client to read messages from the consumer group.