Diff
checker
文本
文本
图像
文档
Excel
文件夹
Legal
Enterprise
桌面版
定价
登录
下载 Diffchecker 桌面版
比较文本
查找两个文本文件之间的差异
工具
历史
显示
视图
拆分
统一
实时编辑器
隐藏空白更改
折叠未更改行
关闭换行
比对精度
智能
语法
无
更改外观
处理
忽略
文本转换
转到第一个差异
编辑输入
Diffchecker Desktop
运行Diffchecker最安全的方式。获取Diffchecker桌面应用:您的差异永远不会离开您的电脑!
获取桌面版
Untitled diff
创建于
7年前
差异永不过期
清除
导出
分享
解释
12 删除
行
总计
删除
字符
总计
删除
要继续使用此功能,请升级到
Diff
checker
Pro
查看价格
402 行
全部复制
4 添加
行
总计
添加
字符
总计
添加
要继续使用此功能,请升级到
Diff
checker
Pro
查看价格
393 行
全部复制
// Copyright (c) Microsoft Corporation. All rights reserved.
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.
// Licensed under the MIT License.
using System;
using System;
using System.Collections.Generic;
using System.Collections.Generic;
using System.Linq;
using System.Linq;
using System.Threading;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Client;
using Microsoft.Azure.Documents.Linq;
using Microsoft.Azure.Documents.Linq;
using Newtonsoft.Json;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Newtonsoft.Json.Linq;
using Newtonsoft.Json.Serialization;
using Newtonsoft.Json.Serialization;
namespace Microsoft.Bot.Builder.Azure
namespace Microsoft.Bot.Builder.Azure
{
{
/// <summary>
/// <summary>
/// Implements an CosmosDB based storage provider for a bot.
/// Implements an CosmosDB based storage provider for a bot.
/// </summary>
/// </summary>
public class CosmosDbStorage : IStorage
public class CosmosDbStorage : IStorage
{
{
// When setting up the database, calls are made to CosmosDB. If multiple calls are made, we'll end up setting the
// When setting up the database, calls are made to CosmosDB. If multiple calls are made, we'll end up setting the
// collectionLink member variable more than once. The semaphore is for making sure the initialization of the
// collectionLink member variable more than once. The semaphore is for making sure the initialization of the
// database is done only once.
// database is done only once.
private static SemaphoreSlim _semaphore = new SemaphoreSlim(1);
private static SemaphoreSlim _semaphore = new SemaphoreSlim(1);
private readonly JsonSerializer _jsonSerializer = JsonSerializer.Create(new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
private readonly JsonSerializer _jsonSerializer = JsonSerializer.Create(new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All });
private readonly string _databaseId;
private readonly string _databaseId;
private readonly string _partitionKey;
private readonly string _partitionKey;
private readonly string _collectionId;
private readonly string _collectionId;
private readonly RequestOptions _documentCollectionCreationRequestOptions = null;
private readonly RequestOptions _documentCollectionCreationRequestOptions = null;
private readonly RequestOptions _databaseCreationRequestOptions = null;
private readonly RequestOptions _databaseCreationRequestOptions = null;
private readonly IDocumentClient _client;
private readonly IDocumentClient _client;
private string _collectionLink = null;
private string _collectionLink = null;
/// <summary>
/// <summary>
/// Initializes a new instance of the <see cref="CosmosDbStorage"/> class.
/// Initializes a new instance of the <see cref="CosmosDbStorage"/> class.
/// using the provided CosmosDB credentials, database ID, and collection ID.
/// using the provided CosmosDB credentials, database ID, and collection ID.
/// </summary>
/// </summary>
/// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param>
/// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param>
public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions)
public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions)
{
{
if (cosmosDbStorageOptions == null)
if (cosmosDbStorageOptions == null)
{
{
throw new ArgumentNullException(nameof(cosmosDbStorageOptions));
throw new ArgumentNullException(nameof(cosmosDbStorageOptions));
}
}
if (cosmosDbStorageOptions.CosmosDBEndpoint == null)
if (cosmosDbStorageOptions.CosmosDBEndpoint == null)
{
{
throw new ArgumentNullException(nameof(cosmosDbStorageOptions.CosmosDBEndpoint), "Service EndPoint for CosmosDB is required.");
throw new ArgumentNullException(nameof(cosmosDbStorageOptions.CosmosDBEndpoint), "Service EndPoint for CosmosDB is required.");
}
}
if (string.IsNullOrEmpty(cosmosDbStorageOptions.AuthKey))
if (string.IsNullOrEmpty(cosmosDbStorageOptions.AuthKey))
{
{
throw new ArgumentException("AuthKey for CosmosDB is required.", nameof(cosmosDbStorageOptions.AuthKey));
throw new ArgumentException("AuthKey for CosmosDB is required.", nameof(cosmosDbStorageOptions.AuthKey));
}
}
if (string.IsNullOrEmpty(cosmosDbStorageOptions.DatabaseId))
if (string.IsNullOrEmpty(cosmosDbStorageOptions.DatabaseId))
{
{
throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbStorageOptions.DatabaseId));
throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbStorageOptions.DatabaseId));
}
}
if (string.IsNullOrEmpty(cosmosDbStorageOptions.CollectionId))
if (string.IsNullOrEmpty(cosmosDbStorageOptions.CollectionId))
{
{
throw new ArgumentException("CollectionId is required.", nameof(cosmosDbStorageOptions.CollectionId));
throw new ArgumentException("CollectionId is required.", nameof(cosmosDbStorageOptions.CollectionId));
}
}
_databaseId = cosmosDbStorageOptions.DatabaseId;
_databaseId = cosmosDbStorageOptions.DatabaseId;
_collectionId = cosmosDbStorageOptions.CollectionId;
_collectionId = cosmosDbStorageOptions.CollectionId;
_partitionKey = cosmosDbStorageOptions.PartitionKey;
_partitionKey = cosmosDbStorageOptions.PartitionKey;
_documentCollectionCreationRequestOptions = cosmosDbStorageOptions.DocumentCollectionRequestOptions;
_documentCollectionCreationRequestOptions = cosmosDbStorageOptions.DocumentCollectionRequestOptions;
_databaseCreationRequestOptions = cosmosDbStorageOptions.DatabaseCreationRequestOptions;
_databaseCreationRequestOptions = cosmosDbStorageOptions.DatabaseCreationRequestOptions;
// Inject BotBuilder version to CosmosDB Requests
// Inject BotBuilder version to CosmosDB Requests
var version = GetType().Assembly.GetName().Version;
var version = GetType().Assembly.GetName().Version;
var connectionPolicy = new ConnectionPolicy { UserAgentSuffix = $"Microsoft-BotFramework {version}" };
var connectionPolicy = new ConnectionPolicy { UserAgentSuffix = $"Microsoft-BotFramework {version}" };
// Invoke CollectionPolicy delegate to further customize settings
// Invoke CollectionPolicy delegate to further customize settings
cosmosDbStorageOptions.ConnectionPolicyConfigurator?.Invoke(connectionPolicy);
cosmosDbStorageOptions.ConnectionPolicyConfigurator?.Invoke(connectionPolicy);
_client = new DocumentClient(cosmosDbStorageOptions.CosmosDBEndpoint, cosmosDbStorageOptions.AuthKey, connectionPolicy);
_client = new DocumentClient(cosmosDbStorageOptions.CosmosDBEndpoint, cosmosDbStorageOptions.AuthKey, connectionPolicy);
}
}
/// <summary>
/// <summary>
/// Initializes a new instance of the <see cref="CosmosDbStorage"/> class.
/// Initializes a new instance of the <see cref="CosmosDbStorage"/> class.
/// using the provided CosmosDB credentials, database ID, and collection ID.
/// using the provided CosmosDB credentials, database ID, and collection ID.
/// </summary>
/// </summary>
/// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param>
/// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param>
/// <param name="jsonSerializer">If passing in a custom JsonSerializer, we recommend the following settings:
/// <param name="jsonSerializer">If passing in a custom JsonSerializer, we recommend the following settings:
/// <para>jsonSerializer.TypeNameHandling = TypeNameHandling.All.</para>
/// <para>jsonSerializer.TypeNameHandling = TypeNameHandling.All.</para>
/// <para>jsonSerializer.NullValueHandling = NullValueHandling.Include.</para>
/// <para>jsonSerializer.NullValueHandling = NullValueHandling.Include.</para>
/// <para>jsonSerializer.ContractResolver = new DefaultContractResolver().</para>
/// <para>jsonSerializer.ContractResolver = new DefaultContractResolver().</para>
/// </param>
/// </param>
public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions, JsonSerializer jsonSerializer)
public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions, JsonSerializer jsonSerializer)
: this(cosmosDbStorageOptions)
: this(cosmosDbStorageOptions)
{
{
if (jsonSerializer == null)
if (jsonSerializer == null)
{
{
throw new ArgumentNullException(nameof(jsonSerializer));
throw new ArgumentNullException(nameof(jsonSerializer));
已保存差异
原始文本
打开文件
// Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Azure.Documents; using Microsoft.Azure.Documents.Client; using Microsoft.Azure.Documents.Linq; using Newtonsoft.Json; using Newtonsoft.Json.Linq; using Newtonsoft.Json.Serialization; namespace Microsoft.Bot.Builder.Azure { /// <summary> /// Implements an CosmosDB based storage provider for a bot. /// </summary> public class CosmosDbStorage : IStorage { // When setting up the database, calls are made to CosmosDB. If multiple calls are made, we'll end up setting the // collectionLink member variable more than once. The semaphore is for making sure the initialization of the // database is done only once. private static SemaphoreSlim _semaphore = new SemaphoreSlim(1); private readonly JsonSerializer _jsonSerializer = JsonSerializer.Create(new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }); private readonly string _databaseId; private readonly string _partitionKey; private readonly string _collectionId; private readonly RequestOptions _documentCollectionCreationRequestOptions = null; private readonly RequestOptions _databaseCreationRequestOptions = null; private readonly IDocumentClient _client; private string _collectionLink = null; /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// using the provided CosmosDB credentials, database ID, and collection ID. /// </summary> /// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param> public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions) { if (cosmosDbStorageOptions == null) { throw new ArgumentNullException(nameof(cosmosDbStorageOptions)); } if (cosmosDbStorageOptions.CosmosDBEndpoint == null) { throw new ArgumentNullException(nameof(cosmosDbStorageOptions.CosmosDBEndpoint), "Service EndPoint for CosmosDB is required."); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.AuthKey)) { throw new ArgumentException("AuthKey for CosmosDB is required.", nameof(cosmosDbStorageOptions.AuthKey)); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.DatabaseId)) { throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbStorageOptions.DatabaseId)); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.CollectionId)) { throw new ArgumentException("CollectionId is required.", nameof(cosmosDbStorageOptions.CollectionId)); } _databaseId = cosmosDbStorageOptions.DatabaseId; _collectionId = cosmosDbStorageOptions.CollectionId; _partitionKey = cosmosDbStorageOptions.PartitionKey; _documentCollectionCreationRequestOptions = cosmosDbStorageOptions.DocumentCollectionRequestOptions; _databaseCreationRequestOptions = cosmosDbStorageOptions.DatabaseCreationRequestOptions; // Inject BotBuilder version to CosmosDB Requests var version = GetType().Assembly.GetName().Version; var connectionPolicy = new ConnectionPolicy { UserAgentSuffix = $"Microsoft-BotFramework {version}" }; // Invoke CollectionPolicy delegate to further customize settings cosmosDbStorageOptions.ConnectionPolicyConfigurator?.Invoke(connectionPolicy); _client = new DocumentClient(cosmosDbStorageOptions.CosmosDBEndpoint, cosmosDbStorageOptions.AuthKey, connectionPolicy); } /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// using the provided CosmosDB credentials, database ID, and collection ID. /// </summary> /// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param> /// <param name="jsonSerializer">If passing in a custom JsonSerializer, we recommend the following settings: /// <para>jsonSerializer.TypeNameHandling = TypeNameHandling.All.</para> /// <para>jsonSerializer.NullValueHandling = NullValueHandling.Include.</para> /// <para>jsonSerializer.ContractResolver = new DefaultContractResolver().</para> /// </param> public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions, JsonSerializer jsonSerializer) : this(cosmosDbStorageOptions) { if (jsonSerializer == null) { throw new ArgumentNullException(nameof(jsonSerializer)); } _jsonSerializer = jsonSerializer; } /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// This constructor should only be used if the default behavior of the DocumentClient needs to be changed. /// The <see cref="CosmosDbStorage(CosmosDbStorageOptions)"/> constructor is preferer for most cases. /// </summary> /// <param name="documentClient">The custom implementation of IDocumentClient.</param> /// <param name="cosmosDbCustomClientOptions">Custom client configuration options.</param> public CosmosDbStorage(IDocumentClient documentClient, CosmosDbCustomClientOptions cosmosDbCustomClientOptions) { if (cosmosDbCustomClientOptions == null) { throw new ArgumentNullException(nameof(cosmosDbCustomClientOptions)); } if (string.IsNullOrEmpty(cosmosDbCustomClientOptions.DatabaseId)) { throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbCustomClientOptions.DatabaseId)); } if (string.IsNullOrEmpty(cosmosDbCustomClientOptions.CollectionId)) { throw new ArgumentException("CollectionId is required.", nameof(cosmosDbCustomClientOptions.CollectionId)); } _client = documentClient ?? throw new ArgumentNullException(nameof(documentClient), "An implementation of IDocumentClient for CosmosDB is required."); _databaseId = cosmosDbCustomClientOptions.DatabaseId; _collectionId = cosmosDbCustomClientOptions.CollectionId; _documentCollectionCreationRequestOptions = cosmosDbCustomClientOptions.DocumentCollectionRequestOptions; _databaseCreationRequestOptions = cosmosDbCustomClientOptions.DatabaseCreationRequestOptions; // Inject BotBuilder version to CosmosDB Requests var version = GetType().Assembly.GetName().Version; _client.ConnectionPolicy.UserAgentSuffix = $"Microsoft-BotFramework {version}"; } [Obsolete("Replaced by CosmosDBKeyEscape.EscapeKey.")] public static string SanitizeKey(string key) => CosmosDbKeyEscape.EscapeKey(key); /// <summary> /// Deletes storage items from storage. /// </summary> /// <param name="keys">keys of the <see cref="IStoreItem"/> objects to remove from the store.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <seealso cref="ReadAsync(string[], CancellationToken)"/> /// <seealso cref="WriteAsync(IDictionary{string, object}, CancellationToken)"/> public async Task DeleteAsync(string[] keys, CancellationToken cancellationToken) { RequestOptions options = null; if (keys == null) { throw new ArgumentNullException(nameof(keys)); } if (keys.Length == 0) { return; } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); if (!string.IsNullOrEmpty(this._partitionKey)) { options = new RequestOptions() { PartitionKey = new PartitionKey(this._partitionKey) }; } // Parallelize deletion var tasks = keys.Select(key => _client.DeleteDocumentAsync( UriFactory.CreateDocumentUri( _databaseId, _collectionId, CosmosDbKeyEscape.EscapeKey(key)), options, cancellationToken: cancellationToken)); // await to deletion tasks to complete await Task.WhenAll(tasks).ConfigureAwait(false); } /// <summary> /// Reads storage items from storage. /// </summary> /// <param name="keys">keys of the <see cref="IStoreItem"/> objects to read from the store.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <remarks>If the activities are successfully sent, the task result contains /// the items read, indexed by key.</remarks> /// <seealso cref="DeleteAsync(string[], CancellationToken)"/> /// <seealso cref="WriteAsync(IDictionary{string, object}, CancellationToken)"/> public async Task<IDictionary<string, object>> ReadAsync(string[] keys, CancellationToken cancellationToken) { FeedOptions options = null; if (keys == null) { throw new ArgumentNullException(nameof(keys)); } if (keys.Length == 0) { // No keys passed in, no result to return. return new Dictionary<string, object>(); } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); if (!string.IsNullOrEmpty(this._partitionKey)) { options = new FeedOptions() { PartitionKey = new PartitionKey(this._partitionKey) }; } var storeItems = new Dictionary<string, object>(keys.Length); var parameterSequence = string.Join(",", Enumerable.Range(0, keys.Length).Select(i => $"@id{i}")); var parameterValues = keys.Select((key, ix) => new SqlParameter($"@id{ix}", CosmosDbKeyEscape.EscapeKey(key))); var querySpec = new SqlQuerySpec { QueryText = $"SELECT c.id, c.realId, c.document, c._etag FROM c WHERE c.id in ({parameterSequence})", Parameters = new SqlParameterCollection(parameterValues), }; var query = _client.CreateDocumentQuery<DocumentStoreItem>(_collectionLink, querySpec, options).AsDocumentQuery(); while (query.HasMoreResults) { foreach (var doc in await query.ExecuteNextAsync<DocumentStoreItem>(cancellationToken).ConfigureAwait(false)) { var item = doc.Document.ToObject(typeof(object), _jsonSerializer); if (item is IStoreItem storeItem) { storeItem.ETag = doc.ETag; } // doc.Id cannot be used since it is escaped, read it from RealId property instead storeItems.Add(doc.ReadlId, item); } } return storeItems; } /// <summary> /// Writes storage items to storage. /// </summary> /// <param name="changes">The items to write to storage, indexed by key.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <seealso cref="DeleteAsync(string[], CancellationToken)"/> /// <seealso cref="ReadAsync(string[], CancellationToken)"/> public async Task WriteAsync(IDictionary<string, object> changes, CancellationToken cancellationToken) { if (changes == null) { throw new ArgumentNullException(nameof(changes)); } if (changes.Count == 0) { return; } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); foreach (var change in changes) { var json = JObject.FromObject(change.Value, _jsonSerializer); // Remove etag from JSON object that was copied from IStoreItem. // The ETag information is updated as an _etag attribute in the document metadata. json.Remove("eTag"); var documentChange = new DocumentStoreItem { Id = CosmosDbKeyEscape.EscapeKey(change.Key), ReadlId = change.Key, Document = json, }; var etag = (change.Value as IStoreItem)?.ETag; if (etag == null || etag == "*") { // if new item or * then insert or replace unconditionaly await _client.UpsertDocumentAsync( _collectionLink, documentChange, disableAutomaticIdGeneration: true, cancellationToken: cancellationToken).ConfigureAwait(false); } else if (etag.Length > 0) { // if we have an etag, do opt. concurrency replace var uri = UriFactory.CreateDocumentUri(_databaseId, _collectionId, documentChange.Id); var ac = new AccessCondition { Condition = etag, Type = AccessConditionType.IfMatch }; await _client.ReplaceDocumentAsync( uri, documentChange, new RequestOptions { AccessCondition = ac }, cancellationToken: cancellationToken).ConfigureAwait(false); } else { throw new Exception("etag empty"); } } } /// <summary> /// Creates the CosmosDB Database and populates the _collectionLink member variable. /// </summary> /// <remarks> /// This method is idempotent, and thread safe. /// </remarks> private async Task InitializeAsync() { // In the steady-state case, we'll already have a connection string to the // database setup. If so, no need to enter locks or call CosmosDB to get one. if (_collectionLink == null) { // We don't (probably) have a database link yet. Enter the lock, // then check again (aka: Double-Check Lock pattern). await _semaphore.WaitAsync().ConfigureAwait(false); try { if (_collectionLink == null) { // We don't have a database link. Create one. Note that we're inside a semaphore at this point // so other threads may be blocked on us. await _client.CreateDatabaseIfNotExistsAsync( new Database { Id = _databaseId }, _databaseCreationRequestOptions) // pass in any user set database creation flags .ConfigureAwait(false); var documentCollection = new DocumentCollection { Id = _collectionId, }; var response = await _client.CreateDocumentCollectionIfNotExistsAsync( UriFactory.CreateDatabaseUri(_databaseId), documentCollection, _documentCollectionCreationRequestOptions) // pass in any user set collection creation flags .ConfigureAwait(false); _collectionLink = response.Resource.SelfLink; } } finally { _semaphore.Release(); } } } /// <summary> /// Internal data structure for storing items in a CosmosDB Collection. /// </summary> private class DocumentStoreItem { /// <summary> /// Gets or sets the sanitized Id/Key used as PrimaryKey. /// </summary> [JsonProperty("id")] public string Id { get; set; } /// <summary> /// Gets or sets the un-sanitized Id/Key. /// </summary> /// <remarks> /// Note: There is a Typo in the property name ("ReadlId"), that can't be changed due to compatability concerns. The /// Json is correct due to the JsonProperty field, but the Typo needs to stay. /// </remarks> [JsonProperty("realId")] public string ReadlId { get; internal set; } // DO NOT FIX THE TYPO BELOW (See Remarks above). /// <summary> /// Gets or sets the persisted object. /// </summary> [JsonProperty("document")] public JObject Document { get; set; } /// <summary> /// Gets or sets the ETag information for handling optimistic concurrency updates. /// </summary> [JsonProperty("_etag")] public string ETag { get; set; } } } }
更改后文本
打开文件
// Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. using System; using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Azure.Documents; using Microsoft.Azure.Documents.Client; using Microsoft.Azure.Documents.Linq; using Newtonsoft.Json; using Newtonsoft.Json.Linq; using Newtonsoft.Json.Serialization; namespace Microsoft.Bot.Builder.Azure { /// <summary> /// Implements an CosmosDB based storage provider for a bot. /// </summary> public class CosmosDbStorage : IStorage { // When setting up the database, calls are made to CosmosDB. If multiple calls are made, we'll end up setting the // collectionLink member variable more than once. The semaphore is for making sure the initialization of the // database is done only once. private static SemaphoreSlim _semaphore = new SemaphoreSlim(1); private readonly JsonSerializer _jsonSerializer = JsonSerializer.Create(new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All }); private readonly string _databaseId; private readonly string _partitionKey; private readonly string _collectionId; private readonly RequestOptions _documentCollectionCreationRequestOptions = null; private readonly RequestOptions _databaseCreationRequestOptions = null; private readonly IDocumentClient _client; private string _collectionLink = null; /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// using the provided CosmosDB credentials, database ID, and collection ID. /// </summary> /// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param> public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions) { if (cosmosDbStorageOptions == null) { throw new ArgumentNullException(nameof(cosmosDbStorageOptions)); } if (cosmosDbStorageOptions.CosmosDBEndpoint == null) { throw new ArgumentNullException(nameof(cosmosDbStorageOptions.CosmosDBEndpoint), "Service EndPoint for CosmosDB is required."); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.AuthKey)) { throw new ArgumentException("AuthKey for CosmosDB is required.", nameof(cosmosDbStorageOptions.AuthKey)); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.DatabaseId)) { throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbStorageOptions.DatabaseId)); } if (string.IsNullOrEmpty(cosmosDbStorageOptions.CollectionId)) { throw new ArgumentException("CollectionId is required.", nameof(cosmosDbStorageOptions.CollectionId)); } _databaseId = cosmosDbStorageOptions.DatabaseId; _collectionId = cosmosDbStorageOptions.CollectionId; _partitionKey = cosmosDbStorageOptions.PartitionKey; _documentCollectionCreationRequestOptions = cosmosDbStorageOptions.DocumentCollectionRequestOptions; _databaseCreationRequestOptions = cosmosDbStorageOptions.DatabaseCreationRequestOptions; // Inject BotBuilder version to CosmosDB Requests var version = GetType().Assembly.GetName().Version; var connectionPolicy = new ConnectionPolicy { UserAgentSuffix = $"Microsoft-BotFramework {version}" }; // Invoke CollectionPolicy delegate to further customize settings cosmosDbStorageOptions.ConnectionPolicyConfigurator?.Invoke(connectionPolicy); _client = new DocumentClient(cosmosDbStorageOptions.CosmosDBEndpoint, cosmosDbStorageOptions.AuthKey, connectionPolicy); } /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// using the provided CosmosDB credentials, database ID, and collection ID. /// </summary> /// <param name="cosmosDbStorageOptions">Cosmos DB storage configuration options.</param> /// <param name="jsonSerializer">If passing in a custom JsonSerializer, we recommend the following settings: /// <para>jsonSerializer.TypeNameHandling = TypeNameHandling.All.</para> /// <para>jsonSerializer.NullValueHandling = NullValueHandling.Include.</para> /// <para>jsonSerializer.ContractResolver = new DefaultContractResolver().</para> /// </param> public CosmosDbStorage(CosmosDbStorageOptions cosmosDbStorageOptions, JsonSerializer jsonSerializer) : this(cosmosDbStorageOptions) { if (jsonSerializer == null) { throw new ArgumentNullException(nameof(jsonSerializer)); } _jsonSerializer = jsonSerializer; } /// <summary> /// Initializes a new instance of the <see cref="CosmosDbStorage"/> class. /// This constructor should only be used if the default behavior of the DocumentClient needs to be changed. /// The <see cref="CosmosDbStorage(CosmosDbStorageOptions)"/> constructor is preferer for most cases. /// </summary> /// <param name="documentClient">The custom implementation of IDocumentClient.</param> /// <param name="cosmosDbCustomClientOptions">Custom client configuration options.</param> public CosmosDbStorage(IDocumentClient documentClient, CosmosDbCustomClientOptions cosmosDbCustomClientOptions) { if (cosmosDbCustomClientOptions == null) { throw new ArgumentNullException(nameof(cosmosDbCustomClientOptions)); } if (string.IsNullOrEmpty(cosmosDbCustomClientOptions.DatabaseId)) { throw new ArgumentException("DatabaseId is required.", nameof(cosmosDbCustomClientOptions.DatabaseId)); } if (string.IsNullOrEmpty(cosmosDbCustomClientOptions.CollectionId)) { throw new ArgumentException("CollectionId is required.", nameof(cosmosDbCustomClientOptions.CollectionId)); } _client = documentClient ?? throw new ArgumentNullException(nameof(documentClient), "An implementation of IDocumentClient for CosmosDB is required."); _databaseId = cosmosDbCustomClientOptions.DatabaseId; _collectionId = cosmosDbCustomClientOptions.CollectionId; _documentCollectionCreationRequestOptions = cosmosDbCustomClientOptions.DocumentCollectionRequestOptions; _databaseCreationRequestOptions = cosmosDbCustomClientOptions.DatabaseCreationRequestOptions; // Inject BotBuilder version to CosmosDB Requests var version = GetType().Assembly.GetName().Version; _client.ConnectionPolicy.UserAgentSuffix = $"Microsoft-BotFramework {version}"; } [Obsolete("Replaced by CosmosDBKeyEscape.EscapeKey.")] public static string SanitizeKey(string key) => CosmosDbKeyEscape.EscapeKey(key); /// <summary> /// Deletes storage items from storage. /// </summary> /// <param name="keys">keys of the <see cref="IStoreItem"/> objects to remove from the store.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <seealso cref="ReadAsync(string[], CancellationToken)"/> /// <seealso cref="WriteAsync(IDictionary{string, object}, CancellationToken)"/> public async Task DeleteAsync(string[] keys, CancellationToken cancellationToken) { RequestOptions options = null; if (keys == null) { throw new ArgumentNullException(nameof(keys)); } if (keys.Length == 0) { return; } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); // Parallelize deletion var tasks = keys.Select(key => _client.DeleteDocumentAsync( UriFactory.CreateDocumentUri( _databaseId, _collectionId, CosmosDbKeyEscape.EscapeKey(key)), new RequestOptions() { PartitionKey = new PartitionKey(key) }, cancellationToken: cancellationToken)); // await to deletion tasks to complete await Task.WhenAll(tasks).ConfigureAwait(false); } /// <summary> /// Reads storage items from storage. /// </summary> /// <param name="keys">keys of the <see cref="IStoreItem"/> objects to read from the store.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <remarks>If the activities are successfully sent, the task result contains /// the items read, indexed by key.</remarks> /// <seealso cref="DeleteAsync(string[], CancellationToken)"/> /// <seealso cref="WriteAsync(IDictionary{string, object}, CancellationToken)"/> public async Task<IDictionary<string, object>> ReadAsync(string[] keys, CancellationToken cancellationToken) { FeedOptions options = null; if (keys == null) { throw new ArgumentNullException(nameof(keys)); } if (keys.Length == 0) { // No keys passed in, no result to return. return new Dictionary<string, object>(); } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); var storeItems = new Dictionary<string, object>(keys.Length); var parameterSequence = string.Join(",", Enumerable.Range(0, keys.Length).Select(i => $"@id{i}")); var parameterValues = keys.Select((key, ix) => new SqlParameter($"@id{ix}", CosmosDbKeyEscape.EscapeKey(key))); var querySpec = new SqlQuerySpec { QueryText = $"SELECT c.id, c.realId, c.document, c._etag FROM c WHERE c.id in ({parameterSequence})", Parameters = new SqlParameterCollection(parameterValues), }; var query = _client.CreateDocumentQuery<DocumentStoreItem>(_collectionLink, querySpec, new FeedOptions() { PartitionKey = new PartitionKey(keys[0]) }).AsDocumentQuery(); while (query.HasMoreResults) { foreach (var doc in await query.ExecuteNextAsync<DocumentStoreItem>(cancellationToken).ConfigureAwait(false)) { var item = doc.Document.ToObject(typeof(object), _jsonSerializer); if (item is IStoreItem storeItem) { storeItem.ETag = doc.ETag; } // doc.Id cannot be used since it is escaped, read it from RealId property instead storeItems.Add(doc.ReadlId, item); } } return storeItems; } /// <summary> /// Writes storage items to storage. /// </summary> /// <param name="changes">The items to write to storage, indexed by key.</param> /// <param name="cancellationToken">A cancellation token that can be used by other objects /// or threads to receive notice of cancellation.</param> /// <returns>A task that represents the work queued to execute.</returns> /// <seealso cref="DeleteAsync(string[], CancellationToken)"/> /// <seealso cref="ReadAsync(string[], CancellationToken)"/> public async Task WriteAsync(IDictionary<string, object> changes, CancellationToken cancellationToken) { if (changes == null) { throw new ArgumentNullException(nameof(changes)); } if (changes.Count == 0) { return; } // Ensure Initialization has been run await InitializeAsync().ConfigureAwait(false); foreach (var change in changes) { var json = JObject.FromObject(change.Value, _jsonSerializer); // Remove etag from JSON object that was copied from IStoreItem. // The ETag information is updated as an _etag attribute in the document metadata. json.Remove("eTag"); var documentChange = new DocumentStoreItem { Id = CosmosDbKeyEscape.EscapeKey(change.Key), ReadlId = change.Key, Document = json, }; var etag = (change.Value as IStoreItem)?.ETag; if (etag == null || etag == "*") { // if new item or * then insert or replace unconditionaly await _client.UpsertDocumentAsync( _collectionLink, documentChange, disableAutomaticIdGeneration: true, cancellationToken: cancellationToken).ConfigureAwait(false); } else if (etag.Length > 0) { // if we have an etag, do opt. concurrency replace var uri = UriFactory.CreateDocumentUri(_databaseId, _collectionId, documentChange.Id); var ac = new AccessCondition { Condition = etag, Type = AccessConditionType.IfMatch }; await _client.ReplaceDocumentAsync( uri, documentChange, new RequestOptions { AccessCondition = ac }, cancellationToken: cancellationToken).ConfigureAwait(false); } else { throw new Exception("etag empty"); } } } /// <summary> /// Creates the CosmosDB Database and populates the _collectionLink member variable. /// </summary> /// <remarks> /// This method is idempotent, and thread safe. /// </remarks> private async Task InitializeAsync() { // In the steady-state case, we'll already have a connection string to the // database setup. If so, no need to enter locks or call CosmosDB to get one. if (_collectionLink == null) { // We don't (probably) have a database link yet. Enter the lock, // then check again (aka: Double-Check Lock pattern). await _semaphore.WaitAsync().ConfigureAwait(false); try { if (_collectionLink == null) { // We don't have a database link. Create one. Note that we're inside a semaphore at this point // so other threads may be blocked on us. await _client.CreateDatabaseIfNotExistsAsync( new Database { Id = _databaseId }, _databaseCreationRequestOptions) // pass in any user set database creation flags .ConfigureAwait(false); var documentCollection = new DocumentCollection { Id = _collectionId, }; var response = await _client.CreateDocumentCollectionIfNotExistsAsync( UriFactory.CreateDatabaseUri(_databaseId), documentCollection, _documentCollectionCreationRequestOptions) // pass in any user set collection creation flags .ConfigureAwait(false); _collectionLink = response.Resource.SelfLink; } } finally { _semaphore.Release(); } } } /// <summary> /// Internal data structure for storing items in a CosmosDB Collection. /// </summary> private class DocumentStoreItem { /// <summary> /// Gets or sets the sanitized Id/Key used as PrimaryKey. /// </summary> [JsonProperty("id")] public string Id { get; set; } /// <summary> /// Gets or sets the un-sanitized Id/Key. /// </summary> /// <remarks> /// Note: There is a Typo in the property name ("ReadlId"), that can't be changed due to compatability concerns. The /// Json is correct due to the JsonProperty field, but the Typo needs to stay. /// </remarks> [JsonProperty("realId")] public string ReadlId { get; internal set; } // DO NOT FIX THE TYPO BELOW (See Remarks above). /// <summary> /// Gets or sets the persisted object. /// </summary> [JsonProperty("document")] public JObject Document { get; set; } /// <summary> /// Gets or sets the ETag information for handling optimistic concurrency updates. /// </summary> [JsonProperty("_etag")] public string ETag { get; set; } } } }
查找差异