forked from ravendb/ravendb
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
RavenDB-21956 azure queue storage wip
RavenDB-21956 azure queue storage etl - wip
- Loading branch information
1 parent
df870d3
commit 2ed871d
Showing
13 changed files
with
535 additions
and
7 deletions.
There are no files selected for viewing
63 changes: 63 additions & 0 deletions
63
src/Raven.Client/Documents/Operations/ETL/Queue/AzureQueueStorageConnectionSettings.cs
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,63 @@ | ||
| using Sparrow.Json.Parsing; | ||
|
|
||
| namespace Raven.Client.Documents.Operations.ETL.Queue; | ||
|
|
||
| public sealed class AzureQueueStorageConnectionSettings | ||
| { | ||
| // TODO djordje: what should be input, minutes? | ||
| public int TimeToLive { get; set; } | ||
|
|
||
| public int VisibilityTimeout { get; set; } | ||
|
|
||
| public Authentication Authentication; | ||
|
|
||
| public DynamicJsonValue ToJson() | ||
| { | ||
| var json = new DynamicJsonValue | ||
| { | ||
| [nameof(TimeToLive)] = TimeToLive, | ||
| [nameof(VisibilityTimeout)] = VisibilityTimeout, | ||
| [nameof(Authentication)] = Authentication == null | ||
| ? null | ||
| : new DynamicJsonValue | ||
| { | ||
| [nameof(Authentication.EntraId)] = | ||
| Authentication.EntraId == null | ||
| ? null | ||
| : new DynamicJsonValue | ||
| { | ||
| [nameof(Authentication.EntraId.StorageAccountName)] = | ||
| Authentication?.EntraId?.StorageAccountName, | ||
| [nameof(Authentication.EntraId.TenantId)] = | ||
| Authentication?.EntraId?.TenantId, | ||
| [nameof(Authentication.EntraId.ClientId)] = | ||
| Authentication?.EntraId?.ClientId, | ||
| [nameof(Authentication.EntraId.ClientSecret)] = | ||
| Authentication?.EntraId?.ClientSecret | ||
| }, | ||
| [nameof(Authentication.ConnectionString)] = Authentication.ConnectionString == null | ||
| ? null | ||
| : new DynamicJsonValue | ||
| { | ||
| [nameof(Authentication.ConnectionString)] = Authentication?.ConnectionString | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| return json; | ||
| } | ||
| } | ||
|
|
||
| public sealed class Authentication | ||
| { | ||
| public EntraId EntraId { get; set; } | ||
| public string ConnectionString { get; set; } | ||
| } | ||
|
|
||
| public sealed class EntraId | ||
| { | ||
| public string StorageAccountName { get; set; } | ||
| public string TenantId { get; set; } | ||
| public string ClientId { get; set; } | ||
| public string ClientSecret { 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 |
|---|---|---|
|
|
@@ -4,5 +4,6 @@ public enum QueueBrokerType | |
| { | ||
| None, | ||
| Kafka, | ||
| RabbitMq | ||
| RabbitMq, | ||
| AzureQueueStorage | ||
| } | ||
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
106 changes: 106 additions & 0 deletions
106
...r/Documents/ETL/Providers/Queue/AzureQueueStorage/AzureQueueStorageDocumentTransformer.cs
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,106 @@ | ||
| using Jint; | ||
| using Jint.Native; | ||
| using Jint.Native.Object; | ||
| using Jint.Runtime.Interop; | ||
| using Raven.Client.Documents.Operations.ETL; | ||
| using Raven.Client.Documents.Operations.ETL.Queue; | ||
| using Raven.Server.Documents.Patch; | ||
| using Raven.Server.ServerWide.Context; | ||
|
|
||
| namespace Raven.Server.Documents.ETL.Providers.Queue.AzureQueueStorage; | ||
|
|
||
| public sealed class AzureQueueStorageDocumentTransformer<T> : QueueDocumentTransformer<T, AzureQueueStorageItem> | ||
| where T : QueueItem | ||
| { | ||
| public AzureQueueStorageDocumentTransformer(Transformation transformation, DocumentDatabase database, | ||
| DocumentsOperationContext context, QueueEtlConfiguration config) : base(transformation, database, context, | ||
| config) | ||
| { | ||
| } | ||
|
|
||
| protected override void LoadToFunction(string queueName, ScriptRunnerResult document) | ||
| { | ||
| LoadToFunction(queueName, document, null); | ||
| } | ||
|
|
||
| private void LoadToFunction(string queueName, ScriptRunnerResult document, CloudEventAttributes attributes) | ||
| { | ||
| if (queueName == null) | ||
| ThrowLoadParameterIsMandatory(nameof(queueName)); | ||
|
|
||
| var result = document.TranslateToObject(Context); | ||
|
|
||
| var topic = GetOrAdd(queueName); | ||
|
|
||
| topic.Items.Add(new AzureQueueStorageItem(Current) { TransformationResult = result, Attributes = attributes }); | ||
| } | ||
|
|
||
| public override void Initialize(bool debugMode) | ||
| { | ||
| base.Initialize(debugMode); | ||
|
|
||
| DocumentScript.ScriptEngine.SetValue(Transformation.LoadTo, | ||
| new ClrFunction(DocumentScript.ScriptEngine, Transformation.LoadTo, | ||
| LoadToFunctionTranslatorWithAttributes)); | ||
|
|
||
| foreach (var queueName in LoadToDestinations) | ||
| { | ||
| var name = Transformation.LoadTo + queueName; | ||
|
|
||
| DocumentScript.ScriptEngine.SetValue(name, new ClrFunction(DocumentScript.ScriptEngine, name, | ||
| (self, args) => LoadToFunctionTranslatorWithAttributes(queueName, args))); | ||
| } | ||
| } | ||
|
|
||
| private JsValue LoadToFunctionTranslatorWithAttributes(JsValue self, JsValue[] args) | ||
| { | ||
| var methodSignature = "loadTo(name, obj, attributes)"; | ||
|
|
||
| if (args.Length != 2 && args.Length != 3) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} must be called with 2 or 3 parameters"); | ||
|
|
||
| if (args[0].IsString() == false) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} first argument must be a string"); | ||
|
|
||
| if (args[1].IsObject() == false) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} second argument must be an object"); | ||
|
|
||
| if (args.Length == 3 && args[2].IsObject() == false) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} third argument must be an object"); | ||
|
|
||
| return LoadToFunctionTranslatorWithAttributesInternal(args[0].AsString(), args[1].AsObject(), | ||
| args.Length == 3 ? args[2].AsObject() : null); | ||
| } | ||
|
|
||
| private JsValue LoadToFunctionTranslatorWithAttributes(string name, JsValue[] args) | ||
| { | ||
| var methodSignature = $"loadTo{name}(obj, attributes)"; | ||
|
|
||
| if (args.Length != 1 && args.Length != 2) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} must be called with with 1 or 2 parameters"); | ||
|
|
||
| if (args[0].IsObject() == false) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} argument 'obj' must be an object"); | ||
|
|
||
| if (args.Length == 2 && args[1].IsObject() == false) | ||
| ThrowInvalidScriptMethodCall($"{methodSignature} argument 'attributes' must be an object"); | ||
|
|
||
| return LoadToFunctionTranslatorWithAttributesInternal(name, args[0].AsObject(), | ||
| args.Length == 2 ? args[1].AsObject() : null); | ||
| } | ||
|
|
||
| private JsValue LoadToFunctionTranslatorWithAttributesInternal(string name, ObjectInstance obj, | ||
| ObjectInstance attributes) | ||
| { | ||
| var result = new ScriptRunnerResult(DocumentScript, obj); | ||
|
|
||
| CloudEventAttributes cloudEventAttributes = null; | ||
|
|
||
| if (attributes != null) | ||
| cloudEventAttributes = GetCloudEventAttributes(attributes); | ||
|
|
||
| LoadToFunction(name, result, cloudEventAttributes); | ||
|
|
||
| return result.Instance; | ||
| } | ||
| } |
Oops, something went wrong.