From 6694fc7a165b2aafb0016f507613fa6c026a3151 Mon Sep 17 00:00:00 2001 From: Pete Fullergreen Date: Thu, 6 Jul 2023 14:55:11 +0100 Subject: [PATCH] Adding a Json provider in the same vein as the Xml provider. --- .../Core/IJsonNodeDefinition.cs | 12 ++ .../Core/JsonFileDefinition.cs | 35 ++++ .../Core/JsonNodeDefinition.cs | 44 +++++ src/Paillave.Etl.Json/Core/JsonNodeParsed.cs | 15 ++ .../Core/JsonObjectReader.cs | 177 ++++++++++++++++++ .../Core/Mapping/IJsonFieldMapper.cs | 10 + .../Core/Mapping/JsonFieldDefinition.cs | 46 +++++ .../Core/Mapping/JsonMappingProcessor.cs | 17 ++ .../Core/Mapping/Visitors/DummyFieldMapper.cs | 34 ++++ .../Mapping/Visitors/JsonMapperVisitor.cs | 17 ++ .../Visitors/JsonMappingSetterVisitor.cs | 17 ++ .../Mapping/Visitors/NewInstanceVisitor.cs | 56 ++++++ src/Paillave.Etl.Json/JsonFile.Stream.Ex.cs | 43 +++++ .../JsonFileValuesProvider.cs | 26 +++ .../JsonNodeOfTypeStreamNode.cs | 46 +++++ .../Paillave.Etl.Json.csproj | 27 +++ src/Paillave.Etl.sln | 82 +++++--- 17 files changed, 679 insertions(+), 25 deletions(-) create mode 100644 src/Paillave.Etl.Json/Core/IJsonNodeDefinition.cs create mode 100644 src/Paillave.Etl.Json/Core/JsonFileDefinition.cs create mode 100644 src/Paillave.Etl.Json/Core/JsonNodeDefinition.cs create mode 100644 src/Paillave.Etl.Json/Core/JsonNodeParsed.cs create mode 100644 src/Paillave.Etl.Json/Core/JsonObjectReader.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/IJsonFieldMapper.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/JsonFieldDefinition.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/JsonMappingProcessor.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/Visitors/DummyFieldMapper.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMapperVisitor.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMappingSetterVisitor.cs create mode 100644 src/Paillave.Etl.Json/Core/Mapping/Visitors/NewInstanceVisitor.cs create mode 100644 src/Paillave.Etl.Json/JsonFile.Stream.Ex.cs create mode 100644 src/Paillave.Etl.Json/JsonFileValuesProvider.cs create mode 100644 src/Paillave.Etl.Json/JsonNodeOfTypeStreamNode.cs create mode 100644 src/Paillave.Etl.Json/Paillave.Etl.Json.csproj diff --git a/src/Paillave.Etl.Json/Core/IJsonNodeDefinition.cs b/src/Paillave.Etl.Json/Core/IJsonNodeDefinition.cs new file mode 100644 index 00000000..274e1e94 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/IJsonNodeDefinition.cs @@ -0,0 +1,12 @@ +using Paillave.Etl.Json.Core.Mapping; + +namespace Paillave.Etl.Json.Core +{ + public interface IJsonNodeDefinition + { + string Name { get; } + string NodePath { get; } + Type Type { get; } + IList GetJsonFieldDefinitions(); + } +} diff --git a/src/Paillave.Etl.Json/Core/JsonFileDefinition.cs b/src/Paillave.Etl.Json/Core/JsonFileDefinition.cs new file mode 100644 index 00000000..ea698cb3 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/JsonFileDefinition.cs @@ -0,0 +1,35 @@ +using System.Linq.Expressions; +using Paillave.Etl.Json.Core.Mapping; + +namespace Paillave.Etl.Json.Core +{ + public class JsonFileDefinition + { + public Dictionary PrefixToUriNameSpacesDictionary { get; } = new Dictionary(); + internal List JsonNodeDefinitions { get; } = new List(); + + public JsonFileDefinition AddNameSpace(string prefix, string uri) + { + this.PrefixToUriNameSpacesDictionary[prefix] = uri; + return this; + } + + public JsonFileDefinition AddNameSpaces(IDictionary _prefixToUriNameSpacesDictionary) + { + foreach (var item in _prefixToUriNameSpacesDictionary) + this.PrefixToUriNameSpacesDictionary[item.Key] = item.Value; + return this; + } + + public JsonFileDefinition AddNodeDefinition(IJsonNodeDefinition xmlNodeDefinition) + { + this.JsonNodeDefinitions.Add(xmlNodeDefinition); + return this; + } + public JsonFileDefinition AddNodeDefinition(string name, string nodeXPath, Expression> expression) + { + this.JsonNodeDefinitions.Add(JsonNodeDefinition.Create(name, nodeXPath, expression)); + return this; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/JsonNodeDefinition.cs b/src/Paillave.Etl.Json/Core/JsonNodeDefinition.cs new file mode 100644 index 00000000..9d1798d1 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/JsonNodeDefinition.cs @@ -0,0 +1,44 @@ +using Paillave.Etl.Json.Core.Mapping; +using Paillave.Etl.Json.Core.Mapping.Visitors; +using System.Linq.Expressions; + +namespace Paillave.Etl.Json.Core +{ + public static class JsonNodeDefinition + { + public static JsonNodeDefinition Create(string name, string nodePath, Expression> expression) + => new JsonNodeDefinition(name, nodePath).WithMap(expression); + } + public class JsonNodeDefinition : IJsonNodeDefinition + { + public string Name { get; set; } + public IList _jsonFieldDefinitions = new List(); + + public IList GetJsonFieldDefinitions() => _jsonFieldDefinitions.ToList(); + public string NodePath { get; private set; } + + public Type Type { get; } = typeof(T); + + public JsonNodeDefinition(string name, string nodePath) + { + this.Name = name; + this.NodePath = nodePath; + } + public JsonNodeDefinition WithMap(Expression> expression) + { + JsonMapperVisitor vis = new JsonMapperVisitor(); + vis.Visit(expression); + foreach (var item in vis.MappingSetters) + this.SetFieldDefinition(item); + return this; + } + private void SetFieldDefinition(JsonFieldDefinition jsonFieldDefinition) + { + var existingFieldDefinition = _jsonFieldDefinitions.FirstOrDefault(i => i.TargetPropertyInfo.Name == jsonFieldDefinition.TargetPropertyInfo.Name); + if (existingFieldDefinition == null) + _jsonFieldDefinitions.Add(jsonFieldDefinition); + else + if (jsonFieldDefinition.NodePath != null) existingFieldDefinition.NodePath = jsonFieldDefinition.NodePath; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/JsonNodeParsed.cs b/src/Paillave.Etl.Json/Core/JsonNodeParsed.cs new file mode 100644 index 00000000..58061243 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/JsonNodeParsed.cs @@ -0,0 +1,15 @@ +namespace Paillave.Etl.Json.Core +{ + public class JsonNodeParsed + { + public string SourceName { get; internal set; } + public string NodeDefinitionName { get; internal set; } + public string NodePath { get; internal set; } + public Type Type { get; internal set; } + public object Value { get; internal set; } + public T GetValue() => (T)Value; + // public object[] ParentValues { get; internal set; } + // public T GetValue(int level = 0) => (T)(level == 0 ? Value : ParentValues[level - 1]); + public HashSet CorrelationKeys { get; set; } = new HashSet(); + } +} diff --git a/src/Paillave.Etl.Json/Core/JsonObjectReader.cs b/src/Paillave.Etl.Json/Core/JsonObjectReader.cs new file mode 100644 index 00000000..bdcfa6fa --- /dev/null +++ b/src/Paillave.Etl.Json/Core/JsonObjectReader.cs @@ -0,0 +1,177 @@ +using Paillave.Etl.Core; +using Paillave.Etl.Json.Core.Mapping; +using System.Buffers; +using System.Runtime.Serialization.Json; +using System.Text.Json; +using System.Xml; + +namespace Paillave.Etl.Json.Core +{ + public class JsonObjectReader + { + private class JsonReadField + { + public JsonFieldDefinition Definition { get; set; } + public IJsonNodeDefinition NodeDefinition { get; set; } + public int Depth { get; set; } + public object Value { get; set; } + } + + private HashSet _jsonFieldsDefinitionSearch; + private HashSet _jsonNodesDefinitionSearch; + + private readonly List _inScopeReadFields = new List(); + private readonly JsonFileDefinition _jsonFileDefinition; + + public JsonObjectReader(JsonFileDefinition jsonFileDefinition) + { + _jsonFileDefinition = jsonFileDefinition; + _jsonNodesDefinitionSearch = new HashSet(jsonFileDefinition.JsonNodeDefinitions.Select(i => i.NodePath).Distinct()); + _jsonFieldsDefinitionSearch = new HashSet(jsonFileDefinition.JsonNodeDefinitions.SelectMany(nd => nd.GetJsonFieldDefinitions().Select(fd => fd.NodePath)).Distinct()); + } + private bool JsonReadFieldShouldBeCleanedUp(JsonReadField jsonReadField, int depth) + { + var depthScope = jsonReadField.Definition.DepthScope; + int depthLimit; + if (depthScope > 0) + depthLimit = depthScope; + else + depthLimit = jsonReadField.Depth + depthScope; + return depth < depthLimit; + } + private void ProcessEndOfAnyNode(Stack nodes) + { + foreach (var item in _inScopeReadFields.Where(i => JsonReadFieldShouldBeCleanedUp(i, nodes.Count - 1)).ToList()) + _inScopeReadFields.Remove(item); + } + private void ProcessAttributeValue(string key, Stack nodes, string stringContent) + { + // string key = $"/{string.Join("/", nodes.Reverse())}"; + if (!_jsonFieldsDefinitionSearch.Contains(key)) return; + var fds = _jsonFileDefinition.JsonNodeDefinitions.SelectMany(nd => nd.GetJsonFieldDefinitions().Select(fd => new { Fd = fd, Nd = nd })).Where(i => i.Fd.NodePath == key).ToList(); + if (string.IsNullOrWhiteSpace(stringContent)) + { + foreach (var fd in fds) + { + _inScopeReadFields.Add(new JsonReadField + { + Depth = nodes.Count - 1, + Definition = fd.Fd, + NodeDefinition = fd.Nd, + Value = null + }); + } + } + else + { + foreach (var fd in fds) + { + _inScopeReadFields.Add(new JsonReadField + { + Depth = nodes.Count - 1, + Definition = fd.Fd, + NodeDefinition = fd.Nd, + Value = fd.Fd.Convert(stringContent) + }); + } + } + } + private string ComputeKey(Stack nodes) => $"/{string.Join("/", nodes.Select(i => i.Name).Reverse())}"; + private void ProcessEndOfNode(Stack nodes, string text, Action pushResult, string sourceName) + { + string key = ComputeKey(nodes); + if (_jsonFieldsDefinitionSearch.Contains(key)) + { + ProcessAttributeValue(key, nodes, text); + } + else if (_jsonNodesDefinitionSearch.Contains(key)) + { + var (value, nd) = CreateValue(sourceName, key); + pushResult(new JsonNodeParsed + { + NodeDefinitionName = nd.Name, + SourceName = sourceName, + NodePath = nd.NodePath, + Type = nd.Type, + Value = value, + CorrelationKeys = nodes.Select(i => i.Guid).Where(i => i.HasValue).Select(i => i.Value).ToHashSet() + }); + } + ProcessEndOfAnyNode(nodes); + } + + private (object value, IJsonNodeDefinition nd) CreateValue(string sourceName, string key) + { + var nd = _jsonFileDefinition.JsonNodeDefinitions.FirstOrDefault(i => i.NodePath == key); + var objectBuilder = new ObjectBuilder(nd.Type); + foreach (var inScopeReadField in _inScopeReadFields.Where(rf => rf.NodeDefinition.NodePath == key)) + objectBuilder.Values[inScopeReadField.Definition.TargetPropertyInfo.Name] = inScopeReadField.Value; + foreach (var propName in nd.GetJsonFieldDefinitions().Where(i => i.ForRowGuid).Select(i => i.TargetPropertyInfo.Name).ToList()) + objectBuilder.Values[propName] = Guid.NewGuid(); + foreach (var propName in nd.GetJsonFieldDefinitions().Where(i => i.ForSourceName).Select(i => i.TargetPropertyInfo.Name).ToList()) + objectBuilder.Values[propName] = sourceName; + return (objectBuilder.CreateInstance(), nd); + } + + public void Read(Stream fileStream, string sourceName, Action pushResult, CancellationToken cancellationToken) + { + var readerSettings = new JsonReaderOptions(); + readerSettings.CommentHandling = JsonCommentHandling.Skip; + + using var streamReader = new StreamReader(fileStream); + var jsonReader = new Utf8JsonReader(new ReadOnlySequence(streamReader.ReadToEnd().Select(Convert.ToByte).ToArray()), readerSettings); + + var nodes = new Stack(); + string lastPropertyName = null; + string lastArrayPropertyName = null; + while (jsonReader.Read()) + { + if (cancellationToken.IsCancellationRequested) break; + switch (jsonReader.TokenType) + { + case JsonTokenType.PropertyName: + lastPropertyName = jsonReader.GetString(); + break; + + case JsonTokenType.EndObject: + if (nodes.Any()) + { + ProcessEndOfNode(nodes, null, pushResult, sourceName); + nodes.Pop(); + } + break; + + case JsonTokenType.StartArray: + lastArrayPropertyName = lastPropertyName; + break; + + case JsonTokenType.EndArray: + lastArrayPropertyName = null; + break; + + case JsonTokenType.StartObject: + if (!string.IsNullOrEmpty(lastPropertyName)) + { + nodes.Push(new NodeLevel { Name = lastArrayPropertyName ?? lastPropertyName, Guid = Guid.NewGuid() }); + } + break; + + // read values into last-value var + case JsonTokenType.String: + case JsonTokenType.Number: + case JsonTokenType.True: + case JsonTokenType.False: + nodes.Push(new NodeLevel { Name = lastPropertyName, Guid = Guid.NewGuid() }); + ProcessEndOfNode(nodes, jsonReader.GetString(), pushResult, sourceName); + nodes.Pop(); + break; + } + } + } + private struct NodeLevel + { + public string Name { get; set; } + public Guid? Guid { get; set; } + } + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/IJsonFieldMapper.cs b/src/Paillave.Etl.Json/Core/Mapping/IJsonFieldMapper.cs new file mode 100644 index 00000000..38975449 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/IJsonFieldMapper.cs @@ -0,0 +1,10 @@ +namespace Paillave.Etl.Json.Core.Mapping +{ + public interface IJsonFieldMapper + { + T ToPathQuery(string xPathQuery); + T ToPathQuery(string xPathQuery, int depthScope); + string ToSourceName(); + Guid ToRowGuid(); + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/JsonFieldDefinition.cs b/src/Paillave.Etl.Json/Core/Mapping/JsonFieldDefinition.cs new file mode 100644 index 00000000..efa3fcdd --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/JsonFieldDefinition.cs @@ -0,0 +1,46 @@ +using System.Reflection; +using System.Xml; + +namespace Paillave.Etl.Json.Core.Mapping +{ + public class JsonFieldDefinition + { + static JsonFieldDefinition() + { + _typeConverters = typeof(XmlConvert) + .GetMethods(BindingFlags.Static | BindingFlags.Public) + .Where(i => i.Name.StartsWith("To") && i.GetParameters().Count() == 1 && i.GetParameters()[0].ParameterType == typeof(string)) + .ToDictionary(i => i.ReturnType); + } + private MethodInfo _convertMethod = null; + private static readonly Dictionary _typeConverters; + public int DepthScope { get; internal set; } = 0; + public string NodePath { get; internal set; } = null; + public bool ForSourceName { get; internal set; } = false; + public bool ForRowGuid { get; internal set; } = false; + private PropertyInfo _targetPropertyInfo = null; + public PropertyInfo TargetPropertyInfo + { + get => _targetPropertyInfo; + internal set + { + _targetPropertyInfo = value; + var underlyingType = Nullable.GetUnderlyingType(_targetPropertyInfo.PropertyType); + IsNullableProperty = underlyingType != null; + if (!IsNullableProperty) + underlyingType = _targetPropertyInfo.PropertyType; + if (underlyingType == typeof(string)) + _convertMethod = null; + else + _convertMethod = _typeConverters[underlyingType]; + } + } + public bool IsNullableProperty { get; private set; } + public object Convert(string stringValue) + { + if (_convertMethod != null) + return _convertMethod.Invoke(null, new object[] { stringValue }); + else return stringValue; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/JsonMappingProcessor.cs b/src/Paillave.Etl.Json/Core/Mapping/JsonMappingProcessor.cs new file mode 100644 index 00000000..fee86638 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/JsonMappingProcessor.cs @@ -0,0 +1,17 @@ +using Paillave.Etl.Json.Core.Mapping.Visitors; +using System.Linq.Expressions; + +namespace Paillave.Etl.Json.Core.Mapping +{ + public class JsonMappingProcessor + { + public readonly List _mappingSetters; + + public JsonMappingProcessor(Expression> expression) + { + JsonMapperVisitor vis = new JsonMapperVisitor(); + vis.Visit(expression); + this._mappingSetters = vis.MappingSetters; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/Visitors/DummyFieldMapper.cs b/src/Paillave.Etl.Json/Core/Mapping/Visitors/DummyFieldMapper.cs new file mode 100644 index 00000000..ca7ad965 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/Visitors/DummyFieldMapper.cs @@ -0,0 +1,34 @@ +using System; + +namespace Paillave.Etl.Json.Core.Mapping.Visitors +{ + public class DummyFieldMapper : IJsonFieldMapper + { + public JsonFieldDefinition MappingSetter { get; } = new JsonFieldDefinition(); + + public string ToSourceName() + { + this.MappingSetter.ForSourceName = true; + return default; + } + + public Guid ToRowGuid() + { + this.MappingSetter.ForRowGuid = true; + return default; + } + + public T ToPathQuery(string xPathQuery) + { + this.MappingSetter.NodePath = xPathQuery; + return default; + } + + public T ToPathQuery(string xPathQuery, int depthScope) + { + this.MappingSetter.NodePath = xPathQuery; + this.MappingSetter.DepthScope = depthScope; + return default; + } + } +} \ No newline at end of file diff --git a/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMapperVisitor.cs b/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMapperVisitor.cs new file mode 100644 index 00000000..65a50236 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMapperVisitor.cs @@ -0,0 +1,17 @@ +using System.Linq.Expressions; + +namespace Paillave.Etl.Json.Core.Mapping.Visitors +{ + public class JsonMapperVisitor : ExpressionVisitor + { + public List MappingSetters { get; private set; } + protected override Expression VisitLambda(Expression node) + { + NewInstanceVisitor vis = new NewInstanceVisitor(); + vis.Visit(node.Body); + + this.MappingSetters = vis.MappingSetters; + return null; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMappingSetterVisitor.cs b/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMappingSetterVisitor.cs new file mode 100644 index 00000000..05b12125 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/Visitors/JsonMappingSetterVisitor.cs @@ -0,0 +1,17 @@ +using System.Linq.Expressions; + +namespace Paillave.Etl.Json.Core.Mapping.Visitors +{ + public class JsonMappingSetterVisitor : ExpressionVisitor + { + public JsonFieldDefinition MappingSetter = null; + protected override Expression VisitMethodCall(MethodCallExpression node) + { + DummyFieldMapper dummyFieldMapper = new DummyFieldMapper(); + var methodInfo = node.Method; + methodInfo.Invoke(dummyFieldMapper, node.Arguments.Cast().Select(i => i.Value).ToArray()); + this.MappingSetter = dummyFieldMapper.MappingSetter; + return null; + } + } +} diff --git a/src/Paillave.Etl.Json/Core/Mapping/Visitors/NewInstanceVisitor.cs b/src/Paillave.Etl.Json/Core/Mapping/Visitors/NewInstanceVisitor.cs new file mode 100644 index 00000000..024b3063 --- /dev/null +++ b/src/Paillave.Etl.Json/Core/Mapping/Visitors/NewInstanceVisitor.cs @@ -0,0 +1,56 @@ +using System.Collections.Generic; +using System.Linq; +using System.Linq.Expressions; +using System.Reflection; + +namespace Paillave.Etl.Json.Core.Mapping.Visitors +{ + public class NewInstanceVisitor : ExpressionVisitor + { + public List MappingSetters { get; private set; } = new List(); + + protected override Expression VisitNew(NewExpression node) + { + if (node.Members != null) + this.MappingSetters.AddRange(node.Members.Zip(node.Arguments, (m, a) => GetMappingSetterDefinition(a, m as PropertyInfo)).ToList()); + return base.VisitNew(node); + } + protected override MemberAssignment VisitMemberAssignment(MemberAssignment node) + { + var vis = new UnaryMapping(); + vis.Visit(node.Expression); + if (vis.MappingSetter != null) + { + vis.MappingSetter.TargetPropertyInfo = node.Member as PropertyInfo; + MappingSetters.Add(vis.MappingSetter); + } + else + { + var vis2 = new JsonMappingSetterVisitor(); + vis2.Visit(node.Expression); + vis2.MappingSetter.TargetPropertyInfo = node.Member as PropertyInfo; + MappingSetters.Add(vis2.MappingSetter); + } + return base.VisitMemberAssignment(node); + } + private JsonFieldDefinition GetMappingSetterDefinition(Expression argument, PropertyInfo propertyInfo) + { + JsonMappingSetterVisitor vis = new JsonMappingSetterVisitor(); + vis.Visit(argument); + JsonFieldDefinition mappingSetter = vis.MappingSetter; + mappingSetter.TargetPropertyInfo = propertyInfo; + return mappingSetter; + } + private class UnaryMapping : ExpressionVisitor + { + public JsonFieldDefinition MappingSetter = null; + protected override Expression VisitUnary(UnaryExpression node) + { + var vis = new JsonMappingSetterVisitor(); + vis.Visit(node.Operand); + this.MappingSetter = vis.MappingSetter; + return null; + } + } + } +} diff --git a/src/Paillave.Etl.Json/JsonFile.Stream.Ex.cs b/src/Paillave.Etl.Json/JsonFile.Stream.Ex.cs new file mode 100644 index 00000000..5efcd0f1 --- /dev/null +++ b/src/Paillave.Etl.Json/JsonFile.Stream.Ex.cs @@ -0,0 +1,43 @@ +using Paillave.Etl.Core; +using Paillave.Etl.Json.Core; + +namespace Paillave.Etl.Json +{ + public static class JsonFileEx + { + public static IStream CrossApplyJsonFile(this IStream stream, string name, Func map, bool noParallelisation = false, bool useStreamCopy = true) + { + var valuesProvider = new JsonFileValuesProvider(new JsonFileValuesProviderArgs + { + JsonFileDefinition = map(new JsonFileDefinition()), + UseStreamCopy = useStreamCopy + }); + return stream.CrossApply(name, valuesProvider, noParallelisation); + } + public static IStream CrossApplyJsonFile(this IStream stream, string name, JsonFileDefinition jsonFileDefinition, bool noParallelisation = false, bool useStreamCopy = true) + { + var valuesProvider = new JsonFileValuesProvider(new JsonFileValuesProviderArgs + { + JsonFileDefinition = jsonFileDefinition, + UseStreamCopy = useStreamCopy + }); + return stream.CrossApply(name, valuesProvider, noParallelisation); + } + public static IStream JsonNodeOfType(this IStream stream, string name, string nodeDefinitionName = null) + { + return new JsonNodeOfTypeStreamNode(name, new JsonNodeOfTypeFileArgs + { + MainStream = stream, + NodeDefinitionName = nodeDefinitionName + }).Output; + } + public static IStream> JsonNodeOfTypeCorrelated(this IStream stream, string name, string nodeDefinitionName = null) + { + return new JsonNodeOfTypeCorrelatedStreamNode(name, new JsonNodeOfTypeFileArgs + { + MainStream = stream, + NodeDefinitionName = nodeDefinitionName + }).Output; + } + } +} diff --git a/src/Paillave.Etl.Json/JsonFileValuesProvider.cs b/src/Paillave.Etl.Json/JsonFileValuesProvider.cs new file mode 100644 index 00000000..6d86d843 --- /dev/null +++ b/src/Paillave.Etl.Json/JsonFileValuesProvider.cs @@ -0,0 +1,26 @@ +using Paillave.Etl.Core; +using Paillave.Etl.Json.Core; + +namespace Paillave.Etl.Json +{ + public class JsonFileValuesProviderArgs + { + public JsonFileDefinition JsonFileDefinition { get; set; } + public bool UseStreamCopy { get; set; } = true; + } + public class JsonFileValuesProvider : ValuesProviderBase + { + private JsonFileValuesProviderArgs _args; + public JsonFileValuesProvider(JsonFileValuesProviderArgs args) => _args = args; + public override ProcessImpact PerformanceImpact => ProcessImpact.Heavy; + + public override ProcessImpact MemoryFootPrint => ProcessImpact.Light; + + public override void PushValues(IFileValue input, Action push, CancellationToken cancellationToken, IDependencyResolver resolver, IInvoker invoker) + { + using var stream = input.GetContent(); + var jsonObjectReader = new JsonObjectReader(_args.JsonFileDefinition); + jsonObjectReader.Read(stream, input.Name, push, cancellationToken); + } + } +} \ No newline at end of file diff --git a/src/Paillave.Etl.Json/JsonNodeOfTypeStreamNode.cs b/src/Paillave.Etl.Json/JsonNodeOfTypeStreamNode.cs new file mode 100644 index 00000000..6594f53e --- /dev/null +++ b/src/Paillave.Etl.Json/JsonNodeOfTypeStreamNode.cs @@ -0,0 +1,46 @@ +using Paillave.Etl.Core; +using Paillave.Etl.Reactive.Operators; +using Paillave.Etl.Json.Core; + +namespace Paillave.Etl.Json +{ + public class JsonNodeOfTypeFileArgs + { + public IStream MainStream { get; set; } + public string NodeDefinitionName { get; set; } + } + public class JsonNodeOfTypeStreamNode : StreamNodeBase, JsonNodeOfTypeFileArgs> + { + public override ProcessImpact PerformanceImpact => ProcessImpact.Light; + + public override ProcessImpact MemoryFootPrint => ProcessImpact.Light; + public JsonNodeOfTypeStreamNode(string name, JsonNodeOfTypeFileArgs args) : base(name, args) { } + protected override IStream CreateOutputStream(JsonNodeOfTypeFileArgs args) + { + var type = typeof(TOut); + var obs = args.MainStream.Observable.Filter(i => i.Type == type); + if (args.NodeDefinitionName != null) + obs = obs.Filter(i => i.NodeDefinitionName == args.NodeDefinitionName); + return CreateUnsortedStream(obs.Map(i => (TOut)i.Value)); + } + } + public class JsonNodeOfTypeCorrelatedStreamNode : StreamNodeBase, IStream>, JsonNodeOfTypeFileArgs> + { + public override ProcessImpact PerformanceImpact => ProcessImpact.Light; + + public override ProcessImpact MemoryFootPrint => ProcessImpact.Light; + public JsonNodeOfTypeCorrelatedStreamNode(string name, JsonNodeOfTypeFileArgs args) : base(name, args) { } + protected override IStream> CreateOutputStream(JsonNodeOfTypeFileArgs args) + { + var type = typeof(TOut); + var obs = args.MainStream.Observable.Filter(i => i.Type == type); + if (args.NodeDefinitionName != null) + obs = obs.Filter(i => i.NodeDefinitionName == args.NodeDefinitionName); + return CreateUnsortedStream(obs.Map(i => new Correlated + { + CorrelationKeys = i.CorrelationKeys, + Row = (TOut)i.Value + })); + } + } +} diff --git a/src/Paillave.Etl.Json/Paillave.Etl.Json.csproj b/src/Paillave.Etl.Json/Paillave.Etl.Json.csproj new file mode 100644 index 00000000..d7bbccf7 --- /dev/null +++ b/src/Paillave.Etl.Json/Paillave.Etl.Json.csproj @@ -0,0 +1,27 @@ + + + + net6.0 + enable + disable + + Paillave.EtlNet.Json + 2.0.47 + Stéphane Royer, Pete Fullergreen + + true + MIT + https://paillave.github.io/Etl.Net/ + ETL .net core SSIS reactive text file json + ETL.net json files extensions + Extensions for Etl.Net to read JSON files + latest + NugetIcon.png + + + + + + + + diff --git a/src/Paillave.Etl.sln b/src/Paillave.Etl.sln index 39a8fb5f..01968981 100644 --- a/src/Paillave.Etl.sln +++ b/src/Paillave.Etl.sln @@ -1,53 +1,58 @@ -Microsoft Visual Studio Solution File, Format Version 12.00 -# Visual Studio 15 -VisualStudioVersion = 15.0.26124.0 + +Microsoft Visual Studio Solution File, Format Version 12.00 +# Visual Studio Version 17 +VisualStudioVersion = 17.6.33815.320 MinimumVisualStudioVersion = 15.0.26124.0 -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl", "Paillave.Etl\Paillave.Etl.csproj", "{3D8C1E7B-CE02-4120-8F9E-FF3869645E27}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl", "Paillave.Etl\Paillave.Etl.csproj", "{3D8C1E7B-CE02-4120-8F9E-FF3869645E27}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.EntityFrameworkCore", "Paillave.Etl.EntityFrameworkCore\Paillave.Etl.EntityFrameworkCore.csproj", "{46CEEF00-633E-41B3-A473-766917BEF723}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.EntityFrameworkCore", "Paillave.Etl.EntityFrameworkCore\Paillave.Etl.EntityFrameworkCore.csproj", "{46CEEF00-633E-41B3-A473-766917BEF723}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.TextFile", "Paillave.Etl.TextFile\Paillave.Etl.TextFile.csproj", "{98C38DC0-61BD-4D9D-98FF-4E5C30E0B05F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.TextFile", "Paillave.Etl.TextFile\Paillave.Etl.TextFile.csproj", "{98C38DC0-61BD-4D9D-98FF-4E5C30E0B05F}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.XmlFile", "Paillave.Etl.XmlFile\Paillave.Etl.XmlFile.csproj", "{1DE438A4-1ECF-4C7E-8ACD-F77A593E8832}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.XmlFile", "Paillave.Etl.XmlFile\Paillave.Etl.XmlFile.csproj", "{1DE438A4-1ECF-4C7E-8ACD-F77A593E8832}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.ExcelFile", "Paillave.Etl.ExcelFile\Paillave.Etl.ExcelFile.csproj", "{10931DC9-D068-4286-9577-E4D6CCECAC8F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.ExcelFile", "Paillave.Etl.ExcelFile\Paillave.Etl.ExcelFile.csproj", "{10931DC9-D068-4286-9577-E4D6CCECAC8F}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.EntityFrameworkCoreExtension", "Paillave.EntityFrameworkCoreExtension\Paillave.EntityFrameworkCoreExtension.csproj", "{B1D7BA52-0A01-418E-8EB2-FC51E8BEA558}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.EntityFrameworkCoreExtension", "Paillave.EntityFrameworkCoreExtension\Paillave.EntityFrameworkCoreExtension.csproj", "{B1D7BA52-0A01-418E-8EB2-FC51E8BEA558}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Ftp", "Paillave.Etl.Ftp\Paillave.Etl.Ftp.csproj", "{A0E143D9-0468-4DB9-929B-97D00001E779}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Ftp", "Paillave.Etl.Ftp\Paillave.Etl.Ftp.csproj", "{A0E143D9-0468-4DB9-929B-97D00001E779}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Sftp", "Paillave.Etl.Sftp\Paillave.Etl.Sftp.csproj", "{ED8C367D-AEFD-4AF7-9B27-157EDCF9E87A}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Sftp", "Paillave.Etl.Sftp\Paillave.Etl.Sftp.csproj", "{ED8C367D-AEFD-4AF7-9B27-157EDCF9E87A}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.FileSystem", "Paillave.Etl.FileSystem\Paillave.Etl.FileSystem.csproj", "{5F41E12B-8B32-447C-98E6-8382020EBFCD}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.FileSystem", "Paillave.Etl.FileSystem\Paillave.Etl.FileSystem.csproj", "{5F41E12B-8B32-447C-98E6-8382020EBFCD}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.FromConfigurationConnectors", "Paillave.Etl.FromConfigurationConnectors\Paillave.Etl.FromConfigurationConnectors.csproj", "{F96D7C02-A0DD-4FAF-8A1C-98CA550E1841}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.FromConfigurationConnectors", "Paillave.Etl.FromConfigurationConnectors\Paillave.Etl.FromConfigurationConnectors.csproj", "{F96D7C02-A0DD-4FAF-8A1C-98CA550E1841}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Mail", "Paillave.Etl.Mail\Paillave.Etl.Mail.csproj", "{43AD98CF-4220-4A88-953C-D3322B93599B}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Mail", "Paillave.Etl.Mail\Paillave.Etl.Mail.csproj", "{43AD98CF-4220-4A88-953C-D3322B93599B}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Autofac", "Paillave.Etl.Autofac\Paillave.Etl.Autofac.csproj", "{8C5BB7A4-025C-4CA0-9A94-9DAA5F35443C}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Autofac", "Paillave.Etl.Autofac\Paillave.Etl.Autofac.csproj", "{8C5BB7A4-025C-4CA0-9A94-9DAA5F35443C}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Samples", "Tutorials\Paillave.Etl.Samples\Paillave.Etl.Samples.csproj", "{D4F4EF5F-A760-43CF-BDA2-673266AEA225}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Samples", "Tutorials\Paillave.Etl.Samples\Paillave.Etl.Samples.csproj", "{D4F4EF5F-A760-43CF-BDA2-673266AEA225}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.ExecutionToolkit", "Paillave.Etl.ExecutionToolkit\Paillave.Etl.ExecutionToolkit.csproj", "{43CCC8DC-25DF-4D01-83B5-7B4E697CDB8D}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.ExecutionToolkit", "Paillave.Etl.ExecutionToolkit\Paillave.Etl.ExecutionToolkit.csproj", "{43CCC8DC-25DF-4D01-83B5-7B4E697CDB8D}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Zip", "Paillave.Etl.Zip\Paillave.Etl.Zip.csproj", "{8F7C3193-EAD6-4FD1-B9DF-07DF19E5A5CD}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Zip", "Paillave.Etl.Zip\Paillave.Etl.Zip.csproj", "{8F7C3193-EAD6-4FD1-B9DF-07DF19E5A5CD}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.SqlServer", "Paillave.Etl.SqlServer\Paillave.Etl.SqlServer.csproj", "{1FB73BF3-529E-4CC8-A1C7-DA3D422BC656}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.SqlServer", "Paillave.Etl.SqlServer\Paillave.Etl.SqlServer.csproj", "{1FB73BF3-529E-4CC8-A1C7-DA3D422BC656}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Dropbox", "Paillave.Etl.Dropbox\Paillave.Etl.Dropbox.csproj", "{0BE639A1-6CE1-4BDF-A26B-BEB130F44C11}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Dropbox", "Paillave.Etl.Dropbox\Paillave.Etl.Dropbox.csproj", "{0BE639A1-6CE1-4BDF-A26B-BEB130F44C11}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.SimpleTutorial", "Tutorials\SimpleTutorial\SimpleTutorial.csproj", "{59203BB5-60EE-41E3-B166-E54E584C1807}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "SimpleTutorial", "Tutorials\SimpleTutorial\SimpleTutorial.csproj", "{59203BB5-60EE-41E3-B166-E54E584C1807}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.BlogTutorial", "Tutorials\BlogTutorial\BlogTutorial.csproj", "{191DFF01-0BA8-4345-AC4E-C94E8245F007}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "BlogTutorial", "Tutorials\BlogTutorial\BlogTutorial.csproj", "{191DFF01-0BA8-4345-AC4E-C94E8245F007}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Bloomberg", "Paillave.Etl.Bloomberg\Paillave.Etl.Bloomberg.csproj", "{4DBE1730-6E96-43FF-A36E-FBBB16617775}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Bloomberg", "Paillave.Etl.Bloomberg\Paillave.Etl.Bloomberg.csproj", "{4DBE1730-6E96-43FF-A36E-FBBB16617775}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Pdf", "Paillave.Pdf\Paillave.Pdf.csproj", "{8FD75A3B-6726-4F19-A82E-D67DE6539901}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Pdf", "Paillave.Pdf\Paillave.Pdf.csproj", "{8FD75A3B-6726-4F19-A82E-D67DE6539901}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Pdf", "Paillave.Etl.Pdf\Paillave.Etl.Pdf.csproj", "{471FE8AD-1A89-41E4-B39D-773A82FBAA0A}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.Pdf", "Paillave.Etl.Pdf\Paillave.Etl.Pdf.csproj", "{471FE8AD-1A89-41E4-B39D-773A82FBAA0A}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Scheduler", "Paillave.Etl.Scheduler\Paillave.Etl.Scheduler.csproj", "{38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}" EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Paillave.Etl.GraphApi", "Paillave.Etl.GraphApi\Paillave.Etl.GraphApi.csproj", "{54B581C3-AE38-4971-B957-33CE151AB9EA}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paillave.Etl.Json", "Paillave.Etl.Json\Paillave.Etl.Json.csproj", "{FEEAA450-AD8E-462F-A010-4052787E3411}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -322,6 +327,7 @@ Global {471FE8AD-1A89-41E4-B39D-773A82FBAA0A}.Release|x64.Build.0 = Release|Any CPU {471FE8AD-1A89-41E4-B39D-773A82FBAA0A}.Release|x86.ActiveCfg = Release|Any CPU {471FE8AD-1A89-41E4-B39D-773A82FBAA0A}.Release|x86.Build.0 = Release|Any CPU +<<<<<<< HEAD {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Debug|Any CPU.Build.0 = Debug|Any CPU {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Debug|x64.ActiveCfg = Debug|Any CPU @@ -334,6 +340,32 @@ Global {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Release|x64.Build.0 = Release|Any CPU {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Release|x86.ActiveCfg = Release|Any CPU {38195BEF-9DE2-4F8F-BDDA-ACA7699C4AD4}.Release|x86.Build.0 = Release|Any CPU +======= + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|Any CPU.Build.0 = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|x64.ActiveCfg = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|x64.Build.0 = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|x86.ActiveCfg = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Debug|x86.Build.0 = Debug|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|Any CPU.ActiveCfg = Release|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|Any CPU.Build.0 = Release|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|x64.ActiveCfg = Release|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|x64.Build.0 = Release|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|x86.ActiveCfg = Release|Any CPU + {54B581C3-AE38-4971-B957-33CE151AB9EA}.Release|x86.Build.0 = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|x64.ActiveCfg = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|x64.Build.0 = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|x86.ActiveCfg = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Debug|x86.Build.0 = Debug|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|Any CPU.Build.0 = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|x64.ActiveCfg = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|x64.Build.0 = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|x86.ActiveCfg = Release|Any CPU + {FEEAA450-AD8E-462F-A010-4052787E3411}.Release|x86.Build.0 = Release|Any CPU +>>>>>>> 18e334f (Adding a Json provider in the same vein as the Xml provider.) EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE