Skip to content
This repository was archived by the owner on Oct 12, 2023. It is now read-only.

Commit 64467e9

Browse files
robinshrobinsh
authored
Merge pull request #24 from YijunXieMS/master
Update sample to use latest version of Python Event Hubs package
2 parents 4a7634e + a45f6ee commit 64467e9

File tree

4 files changed

+184
-1
lines changed

4 files changed

+184
-1
lines changed

‎iot-hub/Quickstarts/read-d2c-messages/Readme.md‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,24 @@ For an example that uses checkpointing, use the checkpoint store from
4040

4141
The above links have documentation on samples on how to use the checkpoint store.
4242

43-
43+
## WebSocket and proxy
44+
45+
To use web socket, you need to specify param `transport_type` when creating the `EventHubConsumerClient`.
46+
To use http proxy, you need to specify param `http_proxy` when creating the `EventHubConsumerClient`.
47+
```python
48+
from azure.eventhub import TrasnportType
49+
from azure.eventhub.aio import EventHubConsumerClient # This is async API. For sync API, remove ".aio"
50+
client = EventHubConsumerClient.from_connection_string(
51+
conn_str=CONNECTION_STR,
52+
consumer_group="$default",
53+
transport_type=TransportType.AmqpOverWebsocket,
54+
http_proxy={
55+
'proxy_hostname': '<proxy host>',
56+
'proxy_port': 3128,
57+
'username': '<proxy user name>',
58+
'password': '<proxy password>'
59+
}
60+
)
61+
```
4462

4563

‎iot-hub/Quickstarts/read-d2c-messages/read_device_to_cloud_messages_async.py‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
# --------------------------------------------------------------------------------------------
2+
# Copyright (c) Microsoft Corporation. All rights reserved.
3+
# Licensed under the MIT License. See License.txt in the project root for license information.
4+
# --------------------------------------------------------------------------------------------
5+
6+
"""
7+
This sample demonstrates how to use the Microsoft Azure Event Hubs Client for Python async API to
8+
read messages sent from a device. Please see the documentation for @azure/event-hubs package
9+
for more details at https://pypi.org/project/azure-eventhub/
10+
11+
For an example that uses checkpointing, follow up this sample with the sample in the
12+
azure-eventhub-checkpointstoreblob package on GitHub at the following link:
13+
14+
https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/samples/receive_events_using_checkpoint_store_async.py
15+
"""
16+
17+
18+
import asyncio
19+
from azure.eventhub import TransportType
20+
from azure.eventhub.aio import EventHubConsumerClient
21+
22+
23+
# Event Hub-compatible endpoint
24+
# az iot hub show --query properties.eventHubEndpoints.events.endpoint --name {your IoT Hub name}
25+
EVENTHUB_COMPATIBLE_ENDPOINT = "{your Event Hubs compatible endpoint}"
26+
27+
# Event Hub-compatible name
28+
# az iot hub show --query properties.eventHubEndpoints.events.path --name {your IoT Hub name}
29+
EVENTHUB_COMPATIBLE_PATH = "{your Event Hubs compatible name}"
30+
31+
# Primary key for the "service" policy to read messages
32+
# az iot hub policy show --name service --query primaryKey --hub-name {your IoT Hub name}
33+
IOTHUB_SAS_KEY = "{your service primary key}"
34+
35+
# If you have access to the Event Hub-compatible connection string from the Azure portal, then
36+
# you can skip the Azure CLI commands above, and assign the connection string directly here.
37+
CONNECTION_STR = f'Endpoint={EVENTHUB_COMPATIBLE_ENDPOINT}/;SharedAccessKeyName=service;SharedAccessKey={IOTHUB_SAS_KEY};EntityPath={EVENTHUB_COMPATIBLE_PATH}'
38+
39+
# Define callbacks to process events
40+
async def on_event_batch(partition_context, events):
41+
for event in events:
42+
print("Received event from partition: {}.".format(partition_context.partition_id))
43+
print("Telemetry received: ", event.body_as_str())
44+
print("Properties (set by device): ", event.properties)
45+
print("System properties (set by IoT Hub): ", event.system_properties)
46+
print()
47+
await partition_context.update_checkpoint()
48+
49+
async def on_error(partition_context, error):
50+
# Put your code here. partition_context can be None in the on_error callback.
51+
if partition_context:
52+
print("An exception: {} occurred during receiving from Partition: {}.".format(
53+
partition_context.partition_id,
54+
error
55+
))
56+
else:
57+
print("An exception: {} occurred during the load balance process.".format(error))
58+
59+
60+
def main():
61+
loop = asyncio.get_event_loop()
62+
client = EventHubConsumerClient.from_connection_string(
63+
conn_str=CONNECTION_STR,
64+
consumer_group="$default",
65+
# transport_type=TransportType.AmqpOverWebsocket, # uncomment it if you want to use web socket
66+
# http_proxy={ # uncomment if you want to use proxy
67+
# 'proxy_hostname': '127.0.0.1', # proxy hostname.
68+
# 'proxy_port': 3128, # proxy port.
69+
# 'username': '<proxy user name>',
70+
# 'password': '<proxy password>'
71+
# }
72+
)
73+
try:
74+
loop.run_until_complete(client.receive_batch(on_event_batch=on_event_batch, on_error=on_error))
75+
except KeyboardInterrupt:
76+
print("Receiving has stopped.")
77+
finally:
78+
loop.run_until_complete(client.close())
79+
loop.stop()
80+
81+
82+
if __name__ == '__main__':
83+
main()

‎iot-hub/Quickstarts/read-d2c-messages/read_device_to_cloud_messages_sync.py‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
# --------------------------------------------------------------------------------------------
2+
# Copyright (c) Microsoft Corporation. All rights reserved.
3+
# Licensed under the MIT License. See License.txt in the project root for license information.
4+
# --------------------------------------------------------------------------------------------
5+
6+
"""
7+
This sample demonstrates how to use the Microsoft Azure Event Hubs Client for Python sync API to
8+
read messages sent from a device. Please see the documentation for @azure/event-hubs package
9+
for more details at https://pypi.org/project/azure-eventhub/
10+
11+
For an example that uses checkpointing, follow up this sample with the sample in the
12+
azure-eventhub-checkpointstoreblob package on GitHub at the following link:
13+
14+
https://github.com/Azure/azure-sdk-for-python/blob/master/sdk/eventhub/azure-eventhub-checkpointstoreblob/samples/receive_events_using_checkpoint_store.py
15+
"""
16+
17+
18+
from azure.eventhub import TransportType
19+
from azure.eventhub import EventHubConsumerClient
20+
21+
22+
# Event Hub-compatible endpoint
23+
# az iot hub show --query properties.eventHubEndpoints.events.endpoint --name {your IoT Hub name}
24+
EVENTHUB_COMPATIBLE_ENDPOINT = "{your Event Hubs compatible endpoint}"
25+
26+
# Event Hub-compatible name
27+
# az iot hub show --query properties.eventHubEndpoints.events.path --name {your IoT Hub name}
28+
EVENTHUB_COMPATIBLE_PATH = "{your Event Hubs compatible name}"
29+
30+
# Primary key for the "service" policy to read messages
31+
# az iot hub policy show --name service --query primaryKey --hub-name {your IoT Hub name}
32+
IOTHUB_SAS_KEY = "{your service primary key}"
33+
34+
# If you have access to the Event Hub-compatible connection string from the Azure portal, then
35+
# you can skip the Azure CLI commands above, and assign the connection string directly here.
36+
CONNECTION_STR = f'Endpoint={EVENTHUB_COMPATIBLE_ENDPOINT}/;SharedAccessKeyName=service;SharedAccessKey={IOTHUB_SAS_KEY};EntityPath={EVENTHUB_COMPATIBLE_PATH}'
37+
38+
# Define callbacks to process events
39+
def on_event_batch(partition_context, events):
40+
for event in events:
41+
print("Received event from partition: {}.".format(partition_context.partition_id))
42+
print("Telemetry received: ", event.body_as_str())
43+
print("Properties (set by device): ", event.properties)
44+
print("System properties (set by IoT Hub): ", event.system_properties)
45+
print()
46+
partition_context.update_checkpoint()
47+
48+
def on_error(partition_context, error):
49+
# Put your code here. partition_context can be None in the on_error callback.
50+
if partition_context:
51+
print("An exception: {} occurred during receiving from Partition: {}.".format(
52+
partition_context.partition_id,
53+
error
54+
))
55+
else:
56+
print("An exception: {} occurred during the load balance process.".format(error))
57+
58+
59+
def main():
60+
client = EventHubConsumerClient.from_connection_string(
61+
conn_str=CONNECTION_STR,
62+
consumer_group="$default",
63+
# transport_type=TransportType.AmqpOverWebsocket, # uncomment it if you want to use web socket
64+
# http_proxy={ # uncomment if you want to use proxy
65+
# 'proxy_hostname': '127.0.0.1', # proxy hostname.
66+
# 'proxy_port': 3128, # proxy port.
67+
# 'username': '<proxy user name>',
68+
# 'password': '<proxy password>'
69+
# }
70+
)
71+
try:
72+
with client:
73+
client.receive_batch(
74+
on_event_batch=on_event_batch,
75+
on_error=on_error
76+
)
77+
except KeyboardInterrupt:
78+
print("Receiving has stopped.")
79+
80+
if __name__ == '__main__':
81+
main()

‎iot-hub/Quickstarts/read-d2c-messages/requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
azure-eventhub

0 commit comments

Comments
 (0)