Download source - 2.9 KB

Introduction

Wouldn't it be nice to be able to react from code instantly to events arising in applications external to your solution? Of course. Well, in some cases, you simply can't do that, in others, you have to hack that system somehow to catch that event and notify your application. Fortunately, some - more and more - platforms provide more or less standard solutions to interact with them - but backward communication is not that widespread. Even Azure APIs have limited support for that. For example - besides polling - at the time this article was written, Office 365 APIs provide two somewhat limited subscription options:

Webhook (https callback) based (both Graph API and Outlook API)
Streaming (Outlook API only)

For the first option to work, you need to have an open http(s) endpoint. This is not possible in some cases, especially in the case of on-premise solutions, mostly because of security reasons; but it is also not usable if fixed IP is not available.

Thus getting notified over a stream channel opened by the client seems a reasonable option, and even a more responsive one than the callback based. To have a general overview, please take a look at the documentation in bold link above.

Please note, that the API used is officially in Beta (preview) stage!

Background

For my recent project, the customer asked for a demonstration application that subscribes to the calendar events of a specific Azure AD user. I have prepared a LinqPad 5 sketch for this purpose. This is why you won't find attached any library or complete application to download, just this sketch. In this article, I will present the particularities encountered during the development of this proof-of-concept solution.

I have used NuGet packages, which are supported only in the registered version of LinqPad. If you are using the free one, you will have to download the packages yourself.

Using the Code

Before you begin, you need to have an Azure AD tenant and a linked Office 365 user - whose events you will be watching. If you don't have these, you will have to proceed for a trial account.

The client ID used in this article is registered (but I don't know for how long) as multi tenant native application, thus you might be able to run directly with your tenant. If not, you will need to register for yourself following this article. Please note that the application is using ADAL, thus endpoint V1 is addressed. The application will request calendar.read delegated scope.

You will absolutely need to change the tenant to yours. Both the tenant and client ID can be found in the Query Properties, on the App.Config page. And of course, you will need to specify your tenant's UPN and authorize this application for it. You will notice that device profile flow is used for authorization. This is because of the final solution targeted. And you will also notice some simplifications too (like missing ConfigureAwait-s, no DI/IoC, etc..), but as I stated before, the intention was not to provide a full-featured library, just a proof-of-concept with a specific goal in mind.

Even if this code is focusing on calendar events, email, contact, and task events need to be handled similarly.

Some Words about the API

Briefly, there are two steps the client has to complete to get streaming notifications:

Initiate a subscription by posting a request.
Start accessing the stream using the subscription ID returned by the server in the previous step.

Quite simple, isn't it?

Yes indeed, but with one annoying limitation: even if the "impersonated" user has access to other user's calendar (shared calendar), you can subscribe to the events of the current user only. So if you need to watch multiple calendars at once, you do it either impersonating those users parallelly or, you have to poll with Graph API. I hope this will change in the future...

Following OData format, you can request simple and rich subscriptions: in the latter case, the notifications will contain portions of the object that triggered the notification itself. Details of the event in my case. And of course, you can specify some filtering if needed.

The classes in the code representing the subscription request are the following:

public abstract class StreamingSubscriptionBase
{
    [JsonProperty(PropertyName = "@odata.type")]
    public string ODataType {get; protected set;}
    
    public string Resource {get; protected set;}
    
    public string ChangeType {get; protected set;}
}

public class SimpleStreamingEventSubscription : StreamingSubscriptionBase
{
    public SimpleStreamingEventSubscription()
    {
        this.ODataType = "#Microsoft.OutlookServices.StreamingSubscription";
        this.Resource = "https://outlook.office.com/api/beta/me/events";
        this.ChangeType = "Acknowledgment, Created, Updated, Deleted, Missed";
    }
}

Note that I have subscribed for all possible changes. The expected response will be deserialized in this class:

public class StreamingSubscriptionResponse
{
    [JsonProperty(PropertyName = "@odata.context")]
    public string ODataContext { get; set; }
    
    [JsonProperty(PropertyName = "@odata.type")]
    public string ODataType { get; set; }

    [JsonProperty(PropertyName = "@odata.id")]
    public string ODataId { get; set; }

    public string Id { get; set; }

    public string Resource { get; set; }

    public string ChangeType { get; set; }
}

Most of this is not really useful, except the Id property, which is the reference of the subscription itself.

In the API documentation, you will find the notion of subscription expiration. You would expect to get the expiration timestamp at the moment your subscription is created. But, for some reason, you will first see it with the first actual event change notification. Which means, if nothing interesting happens, your subscription might get expired before you even get the deadline known. Fortunately, if you are trying to read the stream of an expired subscription, you will get http response code 404 "Not found" - which can be handled quite easily. Thus my code is reacting to this exception and ignoring the existence of an eventual expiration timestamp.

Once you have the Id, you can start to listen, by POSTing an other request. This request will last (almost) infinitely. Well actually, it has a maximum duration of 90 minutes, but from the processing point of view, it is much the same: you have to process it while reading the chunks.

The format of the stream you get is like this:

 {
  "@odata.context":"https://outlook.office.com/api/beta/metadata#Notifications", 
  "value": [
             // messages come here
           ]
 }

This is a valid JSON object: the first three rows come as the stream is opened, the last two rows are sent when the connection timeout is reached (the server will end the connection after a maximum of 90 minutes). The content of the array arrives depending on the keealive period requested and the change events in the resource for which the notification was requested. Practically, it can happen that nothing is in the pipe for minutes. And this is a pitfall, as we will see later on.

Thus, you will get in the stream with any or other fragmentation and order keepalive and event notifications. Of course, these are also valid JSON objects, as valid elements of the values array.

You have to be prepared to handle not only connection termination by the server but also other network issues that might result in broken communication. The documentation is requesting that you should try to reestablish the stream using the same subscription Id, while it is still valid (see the paragraph above about the expiration issue). Thus this whole cycle should end only when you want it to end - you have to keep reopening the stream, renewing subscription as long as you need to, by respecting fair use policies.

Authentication

Before continuing with the actual API, I have to speak about the authentication process. Of course, we have OAuth2 at the base, but fortunately, there is a library we can use, called ADAL (Azure Active Directory Authentication Library), that manages most of the hard work related to this protocol (for the V2.0 endpoint, there is the MSAL library).

Graph API is the new, unified API for Office 365. Still, it is not yet complete. For example, streaming subscriptions are not (yet) implemented in Graph API. But, to be prepared for that, I have used and implemented the IAuthenticationProvider interface defined in the Graph API Library Core:

class AzureAuthenticationProvider: IAuthenticationProvider
{
    private static string aadInstance = ConfigurationManager.AppSettings["ida:AADInstance"];
    private static string tenant = ConfigurationManager.AppSettings["ida:Tenant"];
    private static string clientId = ConfigurationManager.AppSettings["ida:ClientId"];

    private static string authority = 
            String.Format(CultureInfo.InvariantCulture, aadInstance, tenant);
    private static string resource = "https://outlook.office.com";

    TokenCache cache = new TokenCache();
    Action<DeviceCodeResult> signInFeedback = null;

    public AzureAuthenticationProvider(string userPrincipalName, 
                                       Action<DeviceCodeResult> signInFeedback)
    {
        this.signInFeedback = signInFeedback;
        
        string cachefile = Path.Combine(System.Environment.GetFolderPath
               (Environment.SpecialFolder.LocalApplicationData), userPrincipalName);
        cache.AfterAccess += (c) => File.WriteAllBytes(cachefile, cache.Serialize());

        if (File.Exists(cachefile))
        {
            cache.Deserialize(File.ReadAllBytes(cachefile));
        }
    }

    public async Task AuthenticateRequestAsync(HttpRequestMessage request)
    {
        AuthenticationContext authContext = new AuthenticationContext(authority, true, cache);
        AuthenticationResult result = null;

        bool signInNeeded = authContext.TokenCache.Count < 1;
        try
        {
            if (!signInNeeded)
            {
                result = await authContext.AcquireTokenSilentAsync(resource, clientId);
            }
        }
        catch (Exception ex)
        {
            var adalEx = ex.InnerException as AdalException;
            if ((adalEx != null) && (adalEx.ErrorCode == "failed_to_acquire_token_silently"))
            {
                signInNeeded = true;
            }
            else
            {
                throw ex;
            }
        }

        if (signInNeeded)
        {
            DeviceCodeResult codeResult = 
                  await authContext.AcquireDeviceCodeAsync(resource, clientId);
            this.signInFeedback?.Invoke(codeResult);
            result = await authContext.AcquireTokenByDeviceCodeAsync(codeResult);
        }

        request.Headers.Authorization = 
                new AuthenticationHeaderValue("Bearer", result.AccessToken);
    }
}

This provider is using quite a simple but efficient token cache: on every change, it stores the content in a file using the UPN as the file name. Feel free to use your implementation. As stated above, the solution is using the device profile flow, which means that if the cache is empty or the silent authentication fails, the application will prompt the user to open the browser, navigate to a specific location and to enter a specific code. The process will wait until the server confirms that the code was entered and to authorize the application for the user - or it will time out eventually. Because the refresh token will practically never expire, the application will be able to silently authenticate anytime later.

Subscribing

This is the easy part:

public async Task<StreamingSubscriptionResponse> SubscribeToEventsAsync()
    {
        string requestUrl = "https://outlook.office.com/api/beta/me/subscriptions";
        var subscriptionRequestContent = 
            JsonConvert.SerializeObject(new SimpleStreamingEventSubscription());

        HttpRequestMessage request = new HttpRequestMessage(HttpMethod.Post, requestUrl);
        request.Content = new StringContent
                (subscriptionRequestContent, Encoding.UTF8, "application/json");
        await provider.AuthenticateRequestAsync(request);
        HttpResponseMessage response = await client.SendAsync(request);

        return await response.Content.ReadAsAsync<StreamingSubscriptionResponse>();
    }

Please note the spot when authorization occurs using the provider from above. The Graph API client library would perform these steps transparently, but as said before, this feature is not in the Graph API yet.

Consuming Notifications

And now, the interesting part.

public async Task ListenForSubscriptionStreamAsync(CancellationToken ct)
    {
        string requestUrl = "https://outlook.office.com/api/beta/Me/GetNotifications";

        var requestContent = new ListenRequest 
        { ConnectionTimeoutInMinutes = 60, KeepAliveNotificationIntervalInSeconds = 15 };

        var handler = new WinHttpHandler();
        handler.ReceiveDataTimeout = TimeSpan.FromSeconds
                        (requestContent.KeepAliveNotificationIntervalInSeconds + 5);
        HttpClient listeningClient = new HttpClient(handler);

As this is a long running async operation, collaborative cancellation is essential.

When the operation is starting, you can specify the timeout requested in minutes and the keepalive notification interval. If any is missing, the defaults are used. The request is not yet complete, though.

As I highlighted before, it can happen that no any byte will arrive in the socket between these keepalive notifications. The connection might be broken in the meantime and neither the HttpClient object nor the stream reader will notify it - and even cancellation won't work in such a case. Fortunately, we can use the WinHttpHandler "plugin", that has a ReceiveDataTimeout property. You should set its value a little bit higher than the keepalive interval.

In my code, each listener is using its own HttpClient instance. Until further tests are performed, I can't tell how a shared one can handle this long-running-request scenario.

        var subscription = await SubscribeToEventsAsync();

        try
        {
            while (!ct.IsCancellationRequested)
            {
                try
                {
                    requestContent.SubscriptionIds = new string[] { subscription.Id };
                    HttpRequestMessage request = 
                               new HttpRequestMessage(HttpMethod.Post, requestUrl);
                    request.Content = new ObjectContent<ListenRequest>
                                      (requestContent, new JsonMediaTypeFormatter());
                    await provider.AuthenticateRequestAsync(request);

Now we create the initial subscription. Let's suppose for now that this won't fail, but you will probably want to wait a longer bit and try again.

The request is completed with the initial subscription Id, and then authenticated. Everything is part of a large cycle, which is only interrupted by the cancellation initiated from outside.

using (var response = await listeningClient.SendAsync
      (request, HttpCompletionOption.ResponseHeadersRead, ct))
                    {
                        if (!response.IsSuccessStatusCode)
                        {
                            if (response.StatusCode != HttpStatusCode.NotFound)
                            {
                                await Task.Delay(TimeSpan.FromMinutes(1));
                            }
                            subscription = 
                                await SubscribeToEventsAsync(); // renew subscription
                            
                            continue;
                        }

Now we try to initiate the long-running-request. As mentioned before, this is the spot, when the expired request situation is handled: if "not found" is returned, we go to get a new one. In case of any other problem, we do the same, but after a little delay.

Please note the HttpCompletionOption.ResponseHeadersRead option when opening the stream: without this, the client would wait until the whole content arrived, which means, for an hour in my case. This is not what we want. To be able to process the notifications as they arrive, we have to proceed after the header arrived.

And here comes the code that processes the stream:

using (var stream = await response.Content.ReadAsStreamAsync())
using (var streamReader = new StreamReader(stream))
using (var jsonReader = new JsonTextReader(streamReader))
{
   bool inValuesArray = false;
   while (await jsonReader.ReadAsync(ct))
   {
      ct.ThrowIfCancellationRequested();
      if (!inValuesArray && jsonReader.TokenType == 
          JsonToken.PropertyName && jsonReader.Value.ToString() == "value")
      {
         inValuesArray = true;
      }
      if (inValuesArray && jsonReader.TokenType == JsonToken.StartObject)
      {
        JObject obj = await JObject.LoadAsync(jsonReader, ct);
        if (obj["@odata.type"].ToString() == 
            "#Microsoft.OutlookServices.KeepAliveNotification")
        {
          $"{DateTime.Now} keepalive".Dump();
        }
        else
        {
          // This is where you do your job...
          obj.ToString(Newtonsoft.Json.Formatting.Indented, null).Dump();
        }
      }
   }

   $"{DateTime.Now} Stream ended".Dump();
}

As mentioned before, the stream will contain a big, valid JSON object. But to be able to react on the notification, we need to process the elements of the value array - that means we will need to deserialize portions of the stream. Fortunately, the great Json.NET package contains a JsonTextReader class that is capable of processing JSON tokens as they arrive. With one problem: jsonReader.ReadAsync is blocked waiting for the next chars in the stream. If the socket is not closed but the communication is failing, we would be stuck without return... even the cancellation token passed has no effect. But lucky us, we have WinHttpHandler.ReceiveDataTimeout that won't let it wait infinitely.

The stream will contain only keepalive notifications and event change notifications. We are simply reading them into JObject instances, which can easily be converted to our custom objects (not declared in my code).

At the end of this method, you will find some catch blocks, but practically that's it.

Recap

Although this code is not complete and not optimal either, I think it is a good starting point for everybody who wants to consume Outlook streaming notification API from dotNet application.

History

8^th August, 2017: Version 1.0

Consuming Outlook Streaming Notifications in C#