方法: 登録を一括でエクスポートおよび変更する

[アーティクル]
11/09/2015

シナリオによっては、通知ハブで多数の登録情報を作成または変更する必要があります。このようなシナリオとして、バッチ計算の後でタグを更新する場合や、Notification Hubs を使用できるように既存のプッシュ実装を移行する場合などがあります。

このトピックでは、Notification Hubs のバッチサポートを使用して、個々の通知ハブで大量の操作を実行する方法や、すべての登録をエクスポートする方法について説明します。

大まかな流れ

バッチサポートは、数百万件の登録情報が関係する長時間実行ジョブをサポートできるように設計されています。このスケールを実現するために、バッチ処理のサポートではAzure Storageを使用してジョブの詳細と出力を格納します。一括更新操作の場合、ユーザーは、登録更新操作のリストを内容とするファイルを BLOB コンテナーに作成する必要があります。ジョブの開始時に、ユーザーは入力 BLOB の URL と、出力ディレクトリ (これも BLOB コンテナー内に含まれる) の URL を提供します。ジョブの開始後は、開始時に指定した URL にクエリを実行して状態を確認できます。個々のジョブでは、特定の操作 (作成、更新、削除) しか実行できないことに注意してください。エクスポート操作も同じように実行されます。

[インポート]

セットアップ

このセクションでは、以下が確保されていることを前提にしています。

プロビジョニングされた通知ハブ。
Azure Storage BLOB コンテナー。
NuGet パッケージのAzure StorageとAzure Service Busへの参照。

入力ファイルを作成して BLOB に格納する

入力ファイルには、XML 形式でシリアル化された登録情報のリストが 1 行につき 1 件ずつ含まれています。次のコード例は、Azure SDK を使用して登録情報をシリアル化し、BLOB コンテナーにアップロードする方法を示しています。

private static void SerializeToBlob(CloudBlobContainer container, RegistrationDescription[] descriptions)
{
    StringBuilder builder = new StringBuilder();
    foreach (var registrationDescription in descriptions)
    {
        builder.AppendLine(RegistrationDescription.Serialize());
    }

    var inputBlob = container.GetBlockBlobReference(INPUT_FILE_NAME);
    using (MemoryStream stream = new MemoryStream(Encoding.UTF8.GetBytes(builder.ToString())))
    {
        inputBlob.UploadFromStream(stream);
    }
}

重要

上記のコードでは、登録情報がメモリ内でシリアル化され、ストリーム全体が BLOB にアップロードされます。数メガバイトを超えるファイルをアップロードした場合は、これらの手順の実行方法に関する Azure BLOB ガイダンスを参照してください。たとえば、ブロック BLOB などです。

URL トークンを作成する

入力ファイルをアップロードした後に、入力ファイルと出力ディレクトリの URL をそれぞれ生成し、対象の通知ハブに設定する必要があります。入力と出力で異なる BLOB コンテナーを使用できることに注意してください。

static Uri GetOutputDirectoryUrl(CloudBlobContainer container)
{
    SharedAccessBlobPolicy sasConstraints = new SharedAccessBlobPolicy
    {
        SharedAccessExpiryTime = DateTime.UtcNow.AddDays(1),
        Permissions = SharedAccessBlobPermissions.Write | SharedAccessBlobPermissions.List | SharedAccessBlobPermissions.Read
    };

    string sasContainerToken = container.GetSharedAccessSignature(sasConstraints);
    return new Uri(container.Uri + sasContainerToken);
}

static Uri GetInputFileUrl(CloudBlobContainer container, string filePath)
{
    SharedAccessBlobPolicy sasConstraints = new SharedAccessBlobPolicy
    {
        SharedAccessExpiryTime = DateTime.UtcNow.AddDays(1),
        Permissions = SharedAccessBlobPermissions.Read
    };
    string sasToken = container.GetBlockBlobReference(filePath).GetSharedAccessSignature(sasConstraints);
    return new Uri(container.Uri + "/" + filePath + sasToken);
}

ジョブを送信する

これで、入力と出力の 2 つの URL を使用してバッチジョブを開始できるようになりました。

NotificationHubClient client = NotificationHubClient.CreateClientFromConnectionString(CONNECTION_STRING, HUB_NAME);
var createTask = client.SubmitNotificationHubJobAsync(
new NotificationHubJob {
        JobType = NotificationHubJobType.ImportCreateRegistrations,
        OutputContainerUri = outputContainerSasUri,
        ImportFileUri = inputFileSasUri
    }
);
createTask.Wait();

var job = createTask.Result;
long i = 10;
while (i > 0 && job.Status != NotificationHubJobStatus.Completed)
{
    var getJobTask = client.GetNotificationHubJobAsync(job.JobId);
    getJobTask.Wait();
    job = getJobTask.Result;
    Thread.Sleep(1000);
    i--;
}

この例では、入力 URL と出力 URL に加えて、NotificationHubJob オブジェクトを作成します。このオブジェクトには、以下のいずれかが JobType オブジェクトとして含まれます。

ImportCreateRegistrations
ImportUpdateRegistrations
ImportDeleteRegistrations

この呼び出しが完了すると、通知ハブによってジョブが続行され、そのステータスを GetNotificationHubJobAsync の呼び出しによって確認できます。

ジョブが完了したら、出力ディレクトリ内の次のファイルを調べて結果を確認できます。

/<hub>/<jobid>/Failed.txt
/<hub>/<jobid>/Output.txt

これらのファイルには、バッチで成功および失敗した操作のリストが含まれています。ファイル形式は .cvs で、各行に元の入力ファイルの行番号と操作の出力 (通常は作成または更新された登録についての記述) が記されます。

[エクスポート]

登録情報のエクスポートは、インポートと似ていますが、次の点が異なります。

出力 URL のみが必要です。
ExportRegistrations 型の NotificationHubJob を作成する必要があります。

完全なサンプルコード

以下のコードは、複数の登録を 1 つの通知ハブにインポートする完全な作業サンプルです。

using Microsoft.ServiceBus.Notifications;
using Microsoft.WindowsAzure.Storage;
using Microsoft.WindowsAzure.Storage.Blob;
using System;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Xml;

namespace ConsoleApplication1
{
    class Program
    {
        private static string CONNECTION_STRING = "---";
        private static string HUB_NAME = "---";
        private static string INPUT_FILE_NAME = "CreateFile.txt";
        private static string STORAGE_ACCOUNT = "---";
        private static string STORAGE_PASSWORD = "---";
        private static StorageUri STORAGE_ENDPOINT = new StorageUri(new Uri("---"));

        static void Main(string[] args)
        {
            var descriptions = new[]
            {
                new MpnsRegistrationDescription(@"http://dm2.notify.live.net/throttledthirdparty/01.00/12G9Ed13dLb5RbCii5fWzpFpAgAAAAADAQAAAAQUZm52OkJCMjg1QTg1QkZDMkUxREQFBlVTTkMwMQ"),
                new MpnsRegistrationDescription(@"http://dm2.notify.live.net/throttledthirdparty/01.00/12G9Ed13dLb5RbCii5fWzpFpAgAAAAADAQAAAAQUZm52OkJCMjg1QTg1QkZDMjUxREQFBlVTTkMwMQ"),
                new MpnsRegistrationDescription(@"http://dm2.notify.live.net/throttledthirdparty/01.00/12G9Ed13dLb5RbCii5fWzpFpAgAAAAADAQAAAAQUZm52OkJCMjg1QTg1QkZDMhUxREQFBlVTTkMwMQ"),
                new MpnsRegistrationDescription(@"http://dm2.notify.live.net/throttledthirdparty/01.00/12G9Ed13dLb5RbCii5fWzpFpAgAAAAADAQAAAAQUZm52OkJCMjg1QTg1QkZDMdUxREQFBlVTTkMwMQ"),
            };

            //write to blob store to create an input file
            var blobClient = new CloudBlobClient(STORAGE_ENDPOINT, new Microsoft.WindowsAzure.Storage.Auth.StorageCredentials(STORAGE_ACCOUNT, STORAGE_PASSWORD));
            var container = blobClient.GetContainerReference("testjobs");
            container.CreateIfNotExists();

            SerializeToBlob(container, descriptions);

            // TODO then create Sas
            var outputContainerSasUri = GetOutputDirectoryUrl(container);
            var inputFileSasUri = GetInputFileUrl(container, INPUT_FILE_NAME);


            //Lets import this file
            NotificationHubClient client = NotificationHubClient.CreateClientFromConnectionString(CONNECTION_STRING, HUB_NAME);
            var createTask = client.SubmitNotificationHubJobAsync(
                new NotificationHubJob {
                    JobType = NotificationHubJobType.ImportCreateRegistrations,
                    OutputContainerUri = outputContainerSasUri,
                    ImportFileUri = inputFileSasUri
                }
            );
            createTask.Wait();

            var job = createTask.Result;
            long i = 10;
            while (i > 0 && job.Status != NotificationHubJobStatus.Completed)
            {
                var getJobTask = client.GetNotificationHubJobAsync(job.JobId);
                getJobTask.Wait();
                job = getJobTask.Result;
                Thread.Sleep(1000);
                i--;
            }
        }

        private static void SerializeToBlob(CloudBlobContainer container, RegistrationDescription[] descriptions)
        {
            StringBuilder builder = new StringBuilder();
            foreach (var registrationDescription in descriptions)
            {
                builder.AppendLine(RegistrationDescription.Serialize());
            }

            var inputBlob = container.GetBlockBlobReference(INPUT_FILE_NAME);
            using (MemoryStream stream = new MemoryStream(Encoding.UTF8.GetBytes(builder.ToString())))
            {
                inputBlob.UploadFromStream(stream);
            }
        }

        static Uri GetOutputDirectoryUrl(CloudBlobContainer container)
        {
            //Set the expiry time and permissions for the container.
            //In this case no start time is specified, so the shared access signature becomes valid immediately.
            SharedAccessBlobPolicy sasConstraints = new SharedAccessBlobPolicy
            {
                SharedAccessExpiryTime = DateTime.UtcNow.AddHours(4),
                Permissions = SharedAccessBlobPermissions.Write | SharedAccessBlobPermissions.List | SharedAccessBlobPermissions.Read
            };

            //Generate the shared access signature on the container, setting the constraints directly on the signature.
            string sasContainerToken = container.GetSharedAccessSignature(sasConstraints);

            //Return the URI string for the container, including the SAS token.
            return new Uri(container.Uri + sasContainerToken);
        }

        static Uri GetInputFileUrl(CloudBlobContainer container, string filePath)
        {
            //Set the expiry time and permissions for the container.
            //In this case no start time is specified, so the shared access signature becomes valid immediately.
            SharedAccessBlobPolicy sasConstraints = new SharedAccessBlobPolicy
            {
                SharedAccessExpiryTime = DateTime.UtcNow.AddHours(4),
                Permissions = SharedAccessBlobPermissions.Read
            };

            //Generate the shared access signature on the container, setting the constraints directly on the signature.
            string sasToken = container.GetBlockBlobReference(filePath).GetSharedAccessSignature(sasConstraints);

            //Return the URI string for the container, including the SAS token.
            return new Uri(container.Uri + "/" + filePath + sasToken);
        }

        static string GetJobPath(string namespaceName, string notificationHubPath, string jobId)
        {
            return string.Format(CultureInfo.InvariantCulture, @"{0}//{1}/{2}/", namespaceName, notificationHubPath, jobId);
        }
    }
}

Share via

方法: 登録を一括でエクスポートおよび変更する

大まかな流れ

[インポート]

セットアップ

入力ファイルを作成して BLOB に格納する

URL トークンを作成する

ジョブを送信する

[エクスポート]

完全なサンプルコード

その他のリソース

Share via

方法: 登録を一括でエクスポートおよび変更する

大まかな流れ

[インポート]

セットアップ

入力ファイルを作成して BLOB に格納する

URL トークンを作成する

ジョブを送信する

[エクスポート]

完全なサンプル コード

その他のリソース

完全なサンプルコード