Skip to content

Commit b99eb63

Browse files
pakrympakrym
authored
Add EventGrid WebJobs extension (Azure#15058)
1 parent 9caa2a0 commit b99eb63

28 files changed

+1713
-1
lines changed

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/CHANGELOG.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
# Release History
2+
3+
## 3.0.0-beta.1 (Unreleased)

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/Directory.Build.props‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
<Project ToolsVersion="15.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
2+
<PropertyGroup>
3+
<IsClientLibrary>true</IsClientLibrary>
4+
</PropertyGroup>
5+
6+
<Import Project="$([MSBuild]::GetDirectoryNameOfFileAbove($(MSBuildThisFileDirectory)..\, Directory.Build.props))\Directory.Build.props" />
7+
</Project>

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/README.md‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
# Azure WebJobs EventGrid client library for .NET
2+
3+
TODO
4+
5+
## Getting started
6+
7+
### Install the package
8+
9+
10+
### Prerequisites
11+
12+
13+
### Authenticate the client
14+
15+
16+
## Key concepts
17+
18+
TODO
19+
20+
## Examples
21+
22+
TODO
23+
24+
## Troubleshooting
25+
26+
TODO
27+
28+
## Next steps
29+
30+
TODO
31+
32+
## Contributing
33+
34+
TODO

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/AssemblyInfo.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
using System.Runtime.CompilerServices;
5+
6+
[assembly: InternalsVisibleTo("Microsoft.Azure.WebJobs.Extensions.EventGrid.Tests, PublicKey=0024000004800000940000000602000000240000525341310004000001000100cd1dabd5a893b40e75dc901fe7293db4a3caf9cd4d3e3ed6178d49cd476969abe74a9e0b7f4a0bb15edca48758155d35a4f05e6e852fff1b319d103b39ba04acbadd278c2753627c95e1f6f6582425374b92f51cca3deb0d2aab9de3ecda7753900a31f70a236f163006beefffe282888f85e3c76d1205ec7dfef7fa472a17b1")]

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/EventGridExtensionConfigProvider.cs‎

Lines changed: 220 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,220 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Net;
8+
using System.Net.Http;
9+
using System.Threading;
10+
using System.Threading.Tasks;
11+
using System.Web;
12+
using Azure.Messaging.EventGrid;
13+
using Microsoft.Azure.WebJobs.Description;
14+
using Microsoft.Azure.WebJobs.Host.Bindings;
15+
using Microsoft.Azure.WebJobs.Host.Config;
16+
using Microsoft.Azure.WebJobs.Host.Executors;
17+
using Microsoft.Azure.WebJobs.Logging;
18+
using Microsoft.Extensions.Logging;
19+
using Newtonsoft.Json;
20+
using Newtonsoft.Json.Linq;
21+
22+
namespace Microsoft.Azure.WebJobs.Extensions.EventGrid
23+
{
24+
/// <summary>
25+
/// Defines the configuration options for the EventGrid binding.
26+
/// </summary>
27+
[Extension("EventGrid", "EventGrid")]
28+
internal class EventGridExtensionConfigProvider : IExtensionConfigProvider,
29+
IAsyncConverter<HttpRequestMessage, HttpResponseMessage>
30+
{
31+
private ILogger _logger;
32+
private readonly ILoggerFactory _loggerFactory;
33+
private readonly Func<EventGridAttribute, IAsyncCollector<EventGridEvent>> _converter;
34+
35+
// for end to end testing
36+
internal EventGridExtensionConfigProvider(Func<EventGridAttribute, IAsyncCollector<EventGridEvent>> converter, ILoggerFactory loggerFactory)
37+
{
38+
_converter = converter;
39+
_loggerFactory = loggerFactory;
40+
}
41+
42+
// default constructor
43+
public EventGridExtensionConfigProvider(ILoggerFactory loggerFactory)
44+
{
45+
_converter = (attr => new EventGridAsyncCollector(new EventGridPublisherClient(new Uri(attr.TopicEndpointUri), new EventGridSharedAccessSignatureCredential(attr.TopicKeySetting))));
46+
_loggerFactory = loggerFactory;
47+
}
48+
49+
public void Initialize(ExtensionConfigContext context)
50+
{
51+
if (context == null)
52+
{
53+
throw new ArgumentNullException(nameof(context));
54+
}
55+
56+
_logger = _loggerFactory.CreateLogger(LogCategories.CreateTriggerCategory("EventGrid"));
57+
58+
#pragma warning disable 618
59+
Uri url = context.GetWebhookHandler();
60+
#pragma warning restore 618
61+
_logger.LogInformation($"registered EventGrid Endpoint = {url?.GetLeftPart(UriPartial.Path)}");
62+
63+
// Register our extension binding providers
64+
// use converterManager as a hashTable
65+
// also take benefit of identity converter
66+
context
67+
.AddBindingRule<EventGridTriggerAttribute>() // following converters are for EventGridTriggerAttribute only
68+
.AddConverter<JToken, string>((jtoken) => jtoken.ToString(Formatting.Indented))
69+
.AddConverter<JToken, string[]>((jarray) => jarray.Select(ar => ar.ToString(Formatting.Indented)).ToArray())
70+
.AddConverter<JToken, DirectInvokeString>((jtoken) => new DirectInvokeString(null))
71+
.AddConverter<JToken, EventGridEvent>((jobject) => jobject.ToObject<EventGridEvent>()) // surface the type to function runtime
72+
.AddConverter<JToken, EventGridEvent[]>((jobject) => jobject.ToObject<EventGridEvent[]>()) // surface the type to function runtime
73+
.AddOpenConverter<JToken, OpenType.Poco>(typeof(JTokenToPocoConverter<>))
74+
.AddOpenConverter<JToken, OpenType.Poco[]>(typeof(JTokenToPocoConverter<>))
75+
.BindToTrigger<JToken>(new EventGridTriggerAttributeBindingProvider(this));
76+
77+
78+
// Register the output binding
79+
var rule = context
80+
.AddBindingRule<EventGridAttribute>()
81+
.AddConverter<string, EventGridEvent>((str) => EventGridEvent.Parse(str).Single())
82+
.AddConverter<JObject, EventGridEvent>((jobject) => EventGridEvent.Parse(jobject.ToString()).Single());
83+
rule.BindToCollector(_converter);
84+
rule.AddValidator((a, t) =>
85+
{
86+
// if app setting is missing, it will be caught by runtime
87+
// this logic tries to validate the practicality of attribute properties
88+
if (string.IsNullOrWhiteSpace(a.TopicKeySetting))
89+
{
90+
throw new InvalidOperationException($"The '{nameof(EventGridAttribute.TopicKeySetting)}' property must be the name of an application setting containing the Topic Key");
91+
}
92+
93+
if (!Uri.IsWellFormedUriString(a.TopicEndpointUri, UriKind.Absolute))
94+
{
95+
throw new InvalidOperationException($"The '{nameof(EventGridAttribute.TopicEndpointUri)}' property must be a valid absolute Uri");
96+
}
97+
});
98+
}
99+
100+
private Dictionary<string, EventGridListener> _listeners = new Dictionary<string, EventGridListener>();
101+
102+
internal void AddListener(string key, EventGridListener listener)
103+
{
104+
_listeners.Add(key, listener);
105+
}
106+
107+
public async Task<HttpResponseMessage> ConvertAsync(HttpRequestMessage input, CancellationToken cancellationToken)
108+
{
109+
var response = ProcessAsync(input);
110+
return await response.ConfigureAwait(false);
111+
}
112+
113+
private async Task<HttpResponseMessage> ProcessAsync(HttpRequestMessage req)
114+
{
115+
// webjobs.script uses req.GetQueryNameValuePairs();
116+
// which requires webapi.core...but this does not work for .netframework2.0
117+
// TODO change this once webjobs.script is migrated
118+
var functionName = HttpUtility.ParseQueryString(req.RequestUri.Query)["functionName"];
119+
if (String.IsNullOrEmpty(functionName) || !_listeners.ContainsKey(functionName))
120+
{
121+
_logger.LogInformation($"cannot find function: '{functionName}', available function names: [{string.Join(", ", _listeners.Keys.ToArray())}]");
122+
return new HttpResponseMessage(HttpStatusCode.NotFound) { Content = new StringContent($"cannot find function: '{functionName}'") };
123+
}
124+
125+
IEnumerable<string> eventTypeHeaders = null;
126+
string eventTypeHeader = null;
127+
if (req.Headers.TryGetValues("aeg-event-type", out eventTypeHeaders))
128+
{
129+
eventTypeHeader = eventTypeHeaders.First();
130+
}
131+
132+
if (String.Equals(eventTypeHeader, "SubscriptionValidation", StringComparison.OrdinalIgnoreCase))
133+
{
134+
string jsonArray = await req.Content.ReadAsStringAsync().ConfigureAwait(false);
135+
SubscriptionValidationEvent validationEvent = null;
136+
List<JObject> events = JsonConvert.DeserializeObject<List<JObject>>(jsonArray);
137+
// TODO remove unnecessary serialization
138+
validationEvent = ((JObject)events[0]["data"]).ToObject<SubscriptionValidationEvent>();
139+
SubscriptionValidationResponse validationResponse = new SubscriptionValidationResponse { ValidationResponse = validationEvent.ValidationCode };
140+
var returnMessage = new HttpResponseMessage(HttpStatusCode.OK);
141+
returnMessage.Content = new StringContent(JsonConvert.SerializeObject(validationResponse));
142+
_logger.LogInformation($"perform handshake with eventGrid for function: {functionName}");
143+
return returnMessage;
144+
}
145+
else if (String.Equals(eventTypeHeader, "Notification", StringComparison.OrdinalIgnoreCase))
146+
{
147+
JArray events = null;
148+
string requestContent = await req.Content.ReadAsStringAsync().ConfigureAwait(false);
149+
var token = JToken.Parse(requestContent);
150+
if (token.Type == JTokenType.Array)
151+
{
152+
// eventgrid schema
153+
events = (JArray)token;
154+
}
155+
else if (token.Type == JTokenType.Object)
156+
{
157+
// cloudevent schema
158+
events = new JArray
159+
{
160+
token
161+
};
162+
}
163+
164+
List<Task<FunctionResult>> executions = new List<Task<FunctionResult>>();
165+
166+
// Single Dispatch
167+
if (_listeners[functionName].SingleDispatch)
168+
{
169+
foreach (var ev in events)
170+
{
171+
// assume each event is a JObject
172+
TriggeredFunctionData triggerData = new TriggeredFunctionData
173+
{
174+
TriggerValue = ev
175+
};
176+
executions.Add(_listeners[functionName].Executor.TryExecuteAsync(triggerData, CancellationToken.None));
177+
}
178+
await Task.WhenAll(executions).ConfigureAwait(false);
179+
}
180+
// Batch Dispatch
181+
else
182+
{
183+
TriggeredFunctionData triggerData = new TriggeredFunctionData
184+
{
185+
TriggerValue = events
186+
};
187+
executions.Add(_listeners[functionName].Executor.TryExecuteAsync(triggerData, CancellationToken.None));
188+
}
189+
190+
// FIXME without internal queuing, we are going to process all events in parallel
191+
// and return 500 if there's at least one failure...which will cause EventGrid to resend the entire payload
192+
foreach (var execution in executions)
193+
{
194+
if (!execution.Result.Succeeded)
195+
{
196+
return new HttpResponseMessage(HttpStatusCode.InternalServerError) { Content = new StringContent(execution.Result.Exception.Message) };
197+
}
198+
}
199+
200+
return new HttpResponseMessage(HttpStatusCode.Accepted);
201+
}
202+
else if (String.Equals(eventTypeHeader, "Unsubscribe", StringComparison.OrdinalIgnoreCase))
203+
{
204+
// TODO disable function?
205+
return new HttpResponseMessage(HttpStatusCode.Accepted);
206+
}
207+
208+
return new HttpResponseMessage(HttpStatusCode.BadRequest);
209+
210+
}
211+
212+
private class JTokenToPocoConverter<T> : IConverter<JToken, T>
213+
{
214+
public T Convert(JToken input)
215+
{
216+
return input.ToObject<T>();
217+
}
218+
}
219+
}
220+
}

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/EventGridWebJobsBuilderExtensions.cs‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
using System;
5+
6+
namespace Microsoft.Azure.WebJobs.Extensions.EventGrid
7+
{
8+
/// <summary>
9+
/// Extension methods for EventGrid integration
10+
/// </summary>
11+
public static class EventGridWebJobsBuilderExtensions
12+
{
13+
14+
/// <summary>
15+
/// Adds the EventGrid extension to the provided <see cref="IWebJobsBuilder"/>.
16+
/// </summary>
17+
/// <param name="builder">The <see cref="IWebJobsBuilder"/> to configure.</param>
18+
public static IWebJobsBuilder AddEventGrid(this IWebJobsBuilder builder)
19+
{
20+
if (builder == null)
21+
{
22+
throw new ArgumentNullException(nameof(builder));
23+
}
24+
25+
builder.AddExtension<EventGridExtensionConfigProvider>();
26+
return builder;
27+
}
28+
}
29+
}

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/EventGridWebJobsStartup.cs‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
using Microsoft.Azure.WebJobs.Extensions.EventGrid;
5+
using Microsoft.Azure.WebJobs.Hosting;
6+
using Microsoft.Extensions.Hosting;
7+
8+
[assembly: WebJobsStartup(typeof(EventGridWebJobsStartup))]
9+
10+
namespace Microsoft.Azure.WebJobs.Extensions.EventGrid
11+
{
12+
internal class EventGridWebJobsStartup : IWebJobsStartup
13+
{
14+
public void Configure(IWebJobsBuilder builder)
15+
{
16+
builder.AddEventGrid();
17+
}
18+
}
19+
}

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/Microsoft.Azure.WebJobs.Extensions.EventGrid.csproj‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
<PropertyGroup>
3+
<TargetFrameworks>$(RequiredTargetFrameworks)</TargetFrameworks>
4+
<Description>This extension adds bindings for EventGrid</Description>
5+
<Version>3.0.0-beta.1</Version>
6+
<NoWarn>$(NoWarn);AZC0001;CS1591</NoWarn>
7+
</PropertyGroup>
8+
9+
<ItemGroup>
10+
<ProjectReference Include="../../../extensions/Microsoft.Azure.WebJobs.Extensions.Clients/src/Microsoft.Azure.WebJobs.Extensions.Clients.csproj" />
11+
<ProjectReference Include="../../Azure.Messaging.EventGrid/src/Azure.Messaging.EventGrid.csproj" />
12+
</ItemGroup>
13+
14+
</Project>

‎sdk/eventgrid/Microsoft.Azure.WebJobs.Extensions.EventGrid/src/OutputBinding/EventGridAsyncCollector.cs‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
using System;
5+
using System.Collections.Generic;
6+
using System.Linq;
7+
using System.Threading;
8+
using System.Threading.Tasks;
9+
using Azure.Messaging.EventGrid;
10+
namespace Microsoft.Azure.WebJobs.Extensions.EventGrid
11+
{
12+
internal sealed class EventGridAsyncCollector : IAsyncCollector<EventGridEvent>
13+
{
14+
// use EventGridPublisherClient for mocking test
15+
private readonly EventGridPublisherClient _client;
16+
private readonly object _syncroot = new object();
17+
18+
private IList<EventGridEvent> _eventsToSend = new List<EventGridEvent>();
19+
20+
public EventGridAsyncCollector(EventGridPublisherClient client)
21+
{
22+
_client = client;
23+
}
24+
25+
public Task AddAsync(EventGridEvent item, CancellationToken cancellationToken = default(CancellationToken))
26+
{
27+
lock (_syncroot)
28+
{
29+
// Don't let FlushAsyc take place while we're doing this
30+
_eventsToSend.Add(item);
31+
}
32+
33+
return Task.CompletedTask;
34+
}
35+
36+
public async Task FlushAsync(CancellationToken cancellationToken = default(CancellationToken))
37+
{
38+
IList<EventGridEvent> events;
39+
var newEventList = new List<EventGridEvent>();
40+
lock (_syncroot)
41+
{
42+
// swap the events to send out with a new list; locking so 'AddAsync' doesn't take place while we do this
43+
events = _eventsToSend;
44+
_eventsToSend = newEventList;
45+
}
46+
47+
if (events.Any())
48+
{
49+
await _client.SendEventsAsync(events, cancellationToken).ConfigureAwait(false);
50+
}
51+
}
52+
}
53+
}

0 commit comments

Comments
 (0)