Updates related to XComponent V 5.1 (#83)

Migrating to string state machine ids in order to support xcomponent versions > 5.0 and adding support for chunks in snapshots

Author: Anouar Hassine

ReactiveXComponent/Common/ChunkCountdown.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Threading;
+
+namespace ReactiveXComponent.Common
+{
+    public class ChunkCountdown : IDisposable
+    {
+        private readonly ManualResetEvent _completionResetEvent = new ManualResetEvent(false);
+
+        private const int ValueNotSet = int.MinValue;
+        private int _countdown = ValueNotSet;
+
+        public void SetValueIfNotInitialized(int countdown)
+        {
+            Interlocked.CompareExchange(ref _countdown, countdown, ValueNotSet);
+        }
+
+        public void Decrement()
+        {
+            bool isCompleted = Interlocked.Decrement(ref _countdown) <= 0;
+            if (isCompleted)
+            {
+                _completionResetEvent.Set();
+            }
+        }
+
+        public WaitHandle CompletionResetEvent => _completionResetEvent;
+
+        public int Countdown => _countdown;
+
+        public void Dispose()
+        {
+            _completionResetEvent?.Dispose();
+        }
+    }
+}

ReactiveXComponent/Common/ChunkedSnapshotEvent.cs

@@ -0,0 +1,11 @@
+using System;
+
+namespace ReactiveXComponent.Common
+{
+    public class ChunkedSnapshotEvent
+    {
+        public string RequestId { get; set; }
+
+        public SnapshotResponseChunk SnapshotResponseChunk { get; set; }
+    }
+}

ReactiveXComponent/Common/GZipSnapshotItemArrayJsonConverter.cs

@@ -0,0 +1,62 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.IO.Compression;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+using Newtonsoft.Json;
+
+namespace ReactiveXComponent.Common
+{
+    public class GZipSnapshotItemArrayJsonConverter<T> : JsonConverter
+    {
+        public override void WriteJson(JsonWriter writer, object value, JsonSerializer serializer)
+        {
+            var stringWriter = new StringWriter(new StringBuilder(256), CultureInfo.InvariantCulture);
+            using (var jsonTextWriter = new JsonTextWriter(stringWriter))
+            {
+                jsonTextWriter.Formatting = serializer.Formatting;
+                serializer.Serialize(jsonTextWriter, value);
+                var input = Encoding.UTF8.GetBytes(stringWriter.ToString());
+                using (var ms = new MemoryStream())
+                {
+                    using (var compressedStream = new GZipStream(ms, CompressionMode.Compress))
+                    {
+                        compressedStream.Write(input, 0, input.Length);
+                        compressedStream.Close();
+                    }
+                    var compressedBase64 = Convert.ToBase64String(ms.ToArray());
+                    serializer.Serialize(writer, compressedBase64, typeof(string));
+                }
+            }
+        }
+
+        public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
+        {
+            var input = reader.Value as string;
+            if (input != null)
+            {
+                var compressedInput = Convert.FromBase64String(input);
+                using (var ms = new MemoryStream(compressedInput))
+                {
+                    using (var decompressedStream = new GZipStream(ms, CompressionMode.Decompress))
+                    {
+                        using (var streamReader = new StreamReader(decompressedStream))
+                        {
+                            return serializer.Deserialize(streamReader, typeof(T));
+                        }
+                    }
+                }
+            }
+
+            return null;
+        }
+
+        public override bool CanConvert(Type objectType)
+        {
+            return typeof(T) == objectType;
+        }
+    }
+}

ReactiveXComponent/Common/SnapshotEvent.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+
+namespace ReactiveXComponent.Common
+{
+    public class SnapshotEvent
+    {
+        public string RequestId { get; set; }
+
+        public SnapshotResponse SnapshotResponse { get; set; }
+    }
+}

ReactiveXComponent/Common/SnapshotItem.cs

@@ -0,0 +1,36 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace ReactiveXComponent.Common
+{
+    public class SnapshotItem
+    {
+        public const int DefaultIntValue = -1;
+        public const string DefaultStringValue = "";
+
+        public string StateMachineId { get; set; }
+
+        public int WorkerId { get; set; }
+
+        public int StateMachineCode { get; set; }
+
+        public int StateCode { get; set; }
+
+        public int ComponentCode { get; set; }
+
+        public object PublicMember { get; set; }
+
+        public SnapshotItem()
+        {
+            StateMachineId = DefaultStringValue;
+            WorkerId = DefaultIntValue;
+            StateMachineCode = DefaultIntValue;
+            StateCode = DefaultIntValue;
+            ComponentCode = DefaultIntValue;
+            PublicMember = null;
+        }
+    }
+}

ReactiveXComponent/Common/SnapshotItems.cs

@@ -13,5 +13,6 @@ public class SnapshotMessage
         public TimeSpan? Timeout { get; set; } = DefaultTimeout;
         public List<string> CallerPrivateTopic { get; set; }
         public string ReplyTopic { get; set; }
+        public int? ChunkSize { get; set; }
     }
 }

ReactiveXComponent/Common/SnapshotMessage.cs

@@ -0,0 +1,19 @@
+
+using System;
+using System.Collections.Generic;
+using Newtonsoft.Json;
+
+namespace ReactiveXComponent.Common
+{
+    [Serializable]
+    public class SnapshotResponse
+    {
+        [JsonConverter(typeof(GZipSnapshotItemArrayJsonConverter<List<SnapshotItem>>))]
+        public List<SnapshotItem> Items { get; set; }
+
+        public SnapshotResponse()
+        {
+            Items = new List<SnapshotItem>();
+        }
+    }
+}

ReactiveXComponent/Common/SnapshotResponse.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Collections.Generic;
+
+namespace ReactiveXComponent.Common
+{
+    public class SnapshotResponseChunk
+    {
+        public SnapshotResponseChunk()
+        {
+        }
+
+        public SnapshotResponseChunk(string runtimeId, List<string> knownRuntimeIds)
+        {
+            RuntimeId = runtimeId;
+            KnownRuntimeIds = knownRuntimeIds;
+        }
+
+        public int ChunkCount { get; set; }
+
+        public int ChunkId { get; set; }
+
+        public string RuntimeId { get; set; }
+
+        public List<string> KnownRuntimeIds { get; set; }
+
+        public SnapshotResponse Response { get; set; }
+    }
+}

ReactiveXComponent/Common/SnapshotResponseChunk.cs

@@ -5,7 +5,7 @@ namespace ReactiveXComponent.Common
     [Serializable]
     public class StateMachineInstance
     {
-        public long StateMachineId { get; set; }
+        public string StateMachineId { get; set; }
 
         public int AgentId { get; set; }