-
-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reading messages from Azure IoT Hub is done using the Event Hubs compatible endpoint by default. This Pr adds convenience extensions for working with Azure IoT Hub in the `Tingle.EventBus.Transports.EventHubs` project/package. In addition, a functional sample is added. Other changes: 1. Logging in EventHubs is streamlined further. 2. `MessageId, `CorrelationId`, and `ContentType` headers in the `EventData` are now used instead of own properties.
- Loading branch information
1 parent
6214698
commit 67d05d0
Showing
17 changed files
with
493 additions
and
52 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| namespace AzureIotHub; | ||
|
|
||
| internal class AzureIotEventsConsumer : IEventConsumer<MyIotHubEvent> | ||
| { | ||
| private readonly ILogger logger; | ||
|
|
||
| public AzureIotEventsConsumer(ILogger<AzureIotEventsConsumer> logger) | ||
| { | ||
| this.logger = logger ?? throw new ArgumentNullException(nameof(logger)); | ||
| } | ||
|
|
||
| public Task ConsumeAsync(EventContext<MyIotHubEvent> context, CancellationToken cancellationToken) | ||
| { | ||
| var deviceId = context.GetIotHubDeviceId(); | ||
| var source = context.GetIotHubMessageSource(); | ||
| var enqueued = context.GetIotHubEnqueuedTime(); | ||
|
|
||
| logger.LogInformation("Received {Source} from {DeviceId}\r\nEnqueued: {EnqueuedTime}\r\nTimestamped: {Timestamp}.", | ||
| source, | ||
| deviceId, | ||
| enqueued, | ||
| context.Event.Timestamp); | ||
|
|
||
| return Task.CompletedTask; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk.Worker"> | ||
|
|
||
| <PropertyGroup> | ||
| <UserSecretsId>710bcdca-052b-456d-be46-4efede662bb2</UserSecretsId> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="Microsoft.Extensions.Configuration.UserSecrets" Version="6.0.0" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\Tingle.EventBus.Transports.Azure.EventHubs\Tingle.EventBus.Transports.Azure.EventHubs.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| using System.Text.Json.Nodes; | ||
| using System.Text.Json.Serialization; | ||
| using Tingle.EventBus.Configuration; | ||
|
|
||
| namespace AzureIotHub; | ||
|
|
||
| [EventSerializer(typeof(MyIotHubEventSerializer))] | ||
| internal class MyIotHubEvent | ||
| { | ||
| public DateTimeOffset Timestamp { get; set; } | ||
|
|
||
| [JsonExtensionData] | ||
| public JsonObject? Extras { get; set; } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| using Microsoft.Extensions.Options; | ||
| using System.Net.Mime; | ||
| using System.Text.Json; | ||
| using Tingle.EventBus.Serialization; | ||
|
|
||
| namespace AzureIotHub; | ||
|
|
||
| internal class MyIotHubEventSerializer : AbstractEventSerializer | ||
| { | ||
| public MyIotHubEventSerializer(IOptionsMonitor<EventBusOptions> optionsAccessor, | ||
| ILoggerFactory loggerFactory) | ||
| : base(optionsAccessor, loggerFactory) { } | ||
|
|
||
| /// <inheritdoc/> | ||
| protected override IList<string> SupportedMediaTypes => JsonContentTypes; | ||
|
|
||
| /// <inheritdoc/> | ||
| protected override async Task<IEventEnvelope<T>?> DeserializeToEnvelopeAsync<T>(Stream stream, | ||
| ContentType? contentType, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| var serializerOptions = OptionsAccessor.CurrentValue.SerializerOptions; | ||
| var @event = await JsonSerializer.DeserializeAsync<T>(utf8Json: stream, | ||
| options: serializerOptions, | ||
| cancellationToken: cancellationToken); | ||
|
|
||
| return new EventEnvelope<T> { Event = @event, }; | ||
| } | ||
|
|
||
| /// <inheritdoc/> | ||
| protected override Task SerializeEnvelopeAsync<T>(Stream stream, | ||
| EventEnvelope<T> envelope, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| throw new NotSupportedException("Serialization of IotHubEvent events should never happen."); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| using AzureIotHub; | ||
| using Tingle.EventBus.Configuration; | ||
|
|
||
| var host = Host.CreateDefaultBuilder(args) | ||
| .ConfigureServices((hostContext, services) => | ||
| { | ||
| var configuration = hostContext.Configuration; | ||
|
|
||
| services.AddEventBus(builder => | ||
| { | ||
| builder.Configure(o => o.ConfigureEvent<MyIotHubEvent>(reg => reg.ConfigureAsIotHubEvent(configuration["IotHubEventHubName"]))); | ||
|
|
||
| builder.AddConsumer<AzureIotEventsConsumer>(); | ||
|
|
||
| // setup extra serializers | ||
| builder.Services.AddSingleton<MyIotHubEventSerializer>(); | ||
|
|
||
| // Transport specific configuration | ||
| builder.AddAzureEventHubsTransport(options => | ||
| { | ||
| options.Credentials = configuration.GetConnectionString("EventHub"); | ||
| options.BlobStorageCredentials = configuration.GetConnectionString("AzureStorage"); | ||
| }); | ||
| }); | ||
| }) | ||
| .Build(); | ||
|
|
||
| await host.RunAsync(); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| { | ||
| "profiles": { | ||
| "AzureIotHub": { | ||
| "commandName": "Project", | ||
| "environmentVariables": { | ||
| "DOTNET_ENVIRONMENT": "Development" | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| { | ||
| "Logging": { | ||
| "LogLevel": { | ||
| "Default": "Debug", | ||
| "Microsoft": "Information", | ||
| "System": "Information" | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| { | ||
| "Logging": { | ||
| "LogLevel": { | ||
| "Default": "Information", | ||
| "Microsoft": "Warning", | ||
| "Microsoft.Hosting.Lifetime": "Information" | ||
| } | ||
| }, | ||
| "AllowedHosts": "*", | ||
|
|
||
| "ConnectionStrings:AzureStorage": "UseDevelopmentStorage=true;", | ||
| "ConnectionStrings:EventHub": "Endpoint=sb://abcd.servicebus.windows.net/;SharedAccessKeyName=xyz;SharedAccessKey=AAAAAAAAAAAAAAAAAAAAAA==", | ||
| "IotHubEventHubName": "iothub-ehub-test-dev-0000000-0aaaa000aa" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.