Adding filter to GetSnapshot

anouar.hassine · anouar.hassine · commit f538aeedf02d · 2019-11-27T16:44:29.000+01:00
diff --git a/ReactiveXComponent/Connection/IXCPublisher.cs b/ReactiveXComponent/Connection/IXCPublisher.cs
@@ -10,7 +10,7 @@ public interface IXCPublisher : IDisposable
         void SendEvent(string stateMachine, object message, string messageType, Visibility visibility = Visibility.Public);
         void SendEvent(string stateMachine, object message, Visibility visibility = Visibility.Public);
         void SendEvent(StateMachineRefHeader stateMachineRefHeader, object message, Visibility visibility = Visibility.Public);
-        List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000);
-        Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000);
+        List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000);
+        Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000);
     }
 }
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqPublisher.cs b/ReactiveXComponent/RabbitMq/RabbitMqPublisher.cs
@@ -70,14 +70,14 @@ public void SendEvent(StateMachineRefHeader stateMachineRefHeader, object messag
             Send(message, routingKey, prop);
         }
 
-        public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize, int timeout = 10000)
+        public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return _rabbitMqSnapshotManager.GetSnapshot(stateMachine, chunkSize, timeout);
+            return _rabbitMqSnapshotManager.GetSnapshot(stateMachine, filter, chunkSize, timeout);
         }
 
-        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize, int timeout = 10000)
+        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return _rabbitMqSnapshotManager.GetSnapshotAsync(stateMachine, chunkSize, timeout);
+            return _rabbitMqSnapshotManager.GetSnapshotAsync(stateMachine, filter, chunkSize, timeout);
         }
 
         #endregion
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqSnapshotManager.cs b/ReactiveXComponent/RabbitMq/RabbitMqSnapshotManager.cs
@@ -73,7 +73,7 @@ private static SnapshotResponse AggregateChunks(ConcurrentBag<SnapshotResponseCh
             return aggregatedResult;
         }
 
-        public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000)
+        public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
             var guid = Guid.NewGuid();
             var requestId = guid.ToString();
@@ -119,7 +119,7 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =
                 SnapshotReceived += snapshotListenerOnMessageReceived;
 
                 SubscribeSnapshot(stateMachine, requestId);
-                SendSnapshotRequest(stateMachine, requestId, chunkSize, _privateCommunicationIdentifier);
+                SendSnapshotRequest(stateMachine, requestId, filter, chunkSize, _privateCommunicationIdentifier);
 
                 if (receivedSnapshotChunksInitialized.WaitOne(timeout))
                 {
@@ -183,12 +183,12 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =
             return result;
         }
 
-        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000)
+        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return Task.Run(() => GetSnapshot(stateMachine, chunkSize, timeout));
+            return Task.Run(() => GetSnapshot(stateMachine, filter, chunkSize, timeout));
         }
 
-        private void SendSnapshotRequest(string stateMachine, string replyTopic, int? chunkSize, string privateCommunicationIdentifier = null)
+        private void SendSnapshotRequest(string stateMachine, string replyTopic, string filter = null, int? chunkSize = null, string privateCommunicationIdentifier = null)
         {
             if (_xcConfiguration == null)
                 return;
@@ -212,6 +212,7 @@ private void SendSnapshotRequest(string stateMachine, string replyTopic, int? ch
                 CallerPrivateTopic = !string.IsNullOrEmpty(privateCommunicationIdentifier)
                             ? new List<string>{ privateCommunicationIdentifier}
                             : null,
+                Filter = filter,
                 ChunkSize = chunkSize
             };
 
diff --git a/ReactiveXComponent/ReactiveXComponent.csproj b/ReactiveXComponent/ReactiveXComponent.csproj
@@ -34,7 +34,6 @@
         <WorkingDirectory>$(MSBuildThisFileDirectory)$(OutputPath)</WorkingDirectory>
     </PropertyGroup>
     <ItemGroup>
-			<InputAssemblies Include="$(WorkingDirectory)\ReactiveXComponent.dll" />
 			<InputAssemblies Include="$(WorkingDirectory)\Newtonsoft.Json.dll" />
 			<InputAssemblies Include="$(WorkingDirectory)\RabbitMQ.Client.dll" />
 			<InputAssemblies Include="$(WorkingDirectory)\WebSocket4Net.dll" />
diff --git a/ReactiveXComponent/WebSocket/WebSocketPublisher.cs b/ReactiveXComponent/WebSocket/WebSocketPublisher.cs
@@ -65,14 +65,14 @@ public void SendEvent(StateMachineRefHeader stateMachineRefHeader, object messag
             _webSocketClient.Send(webSocketRequest);
         }
 
-        public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize, int timeout = 10000)
+        public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return _webSocketSnapshotManager.GetSnapshot(stateMachine, chunkSize, timeout);
+            return _webSocketSnapshotManager.GetSnapshot(stateMachine, filter, chunkSize, timeout);
         }
 
-        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize, int timeout = 10000)
+        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return _webSocketSnapshotManager.GetSnapshotAsync(stateMachine, chunkSize, timeout);
+            return _webSocketSnapshotManager.GetSnapshotAsync(stateMachine, filter, chunkSize, timeout);
         }
 
         #endregion
diff --git a/ReactiveXComponent/WebSocket/WebSocketSnapshotManager.cs b/ReactiveXComponent/WebSocket/WebSocketSnapshotManager.cs
@@ -43,7 +43,7 @@ public WebSocketSnapshotManager(string component, IWebSocketClient webSocketClie
                 h => SnapshotReceived -= h);
         }
         
-        public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000)
+        public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
             var replyTopic = Guid.NewGuid().ToString();
 
@@ -74,7 +74,7 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =
             using (_snapshotStream.Subscribe(observer))
             {
                 SendWebSocketSnapshotSubscriptionResquest(replyTopic);
-                SendWebSocketSnapshotRequest(stateMachine, replyTopic, chunkSize);
+                SendWebSocketSnapshotRequest(stateMachine, replyTopic, filter, chunkSize);
                 lockEvent.WaitOne(timeout);
             }
 
@@ -84,12 +84,12 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =
             return result;
         }
 
-        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000)
+        public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)
         {
-            return Task.Run(() => GetSnapshot(stateMachine, chunkSize, timeout));
+            return Task.Run(() => GetSnapshot(stateMachine, filter, chunkSize, timeout));
         }
 
-        private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic, int? chunkSize)
+        private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic, string filter = null, int? chunkSize = null)
         {
             if (!_webSocketClient.IsOpen) return;
 
@@ -101,7 +101,7 @@ private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic
                 ComponentCode = componentCode,
                 StateMachineCode = stateMachineCode
             };
-            var snapshotMessage = new WebSocketSnapshotMessage(stateMachineCode, componentCode, replyTopic, _privateCommunicationIdentifier, chunkSize);
+            var snapshotMessage = new WebSocketSnapshotMessage(stateMachineCode, componentCode, replyTopic, _privateCommunicationIdentifier, chunkSize, filter);
             var webSocketRequest = WebSocketMessageHelper.SerializeRequest(
                     WebSocketCommand.Snapshot,
                     inputHeader,
diff --git a/ReactiveXComponent/WebSocket/WebSocketSnapshotMessage.cs b/ReactiveXComponent/WebSocket/WebSocketSnapshotMessage.cs
@@ -10,10 +10,11 @@ public class WebSocketSnapshotMessage
         public string ReplyTopic { get; set; }
         public string[] CallerPrivateTopic { get; set; }
         public int? ChunkSize { get; set; }
+        public string Filter { get; set; }
 
         public WebSocketSnapshotMessage() { }
 
-        public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, string replyTopic, string callerPrivateTopic, int? chunkSize)
+        public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, string replyTopic, string callerPrivateTopic, int? chunkSize, string filter = null)
         {
             StateMachineCode = stateMachineCode;
             ComponentCode = componentCode;
@@ -23,6 +24,7 @@ public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, strin
                 CallerPrivateTopic = new []{ callerPrivateTopic };
             }
             ChunkSize = chunkSize;
+            Filter = filter;
         }
     }
 }
diff --git a/ReactiveXComponentTest/Configuration/ConfigurationTests.cs b/ReactiveXComponentTest/Configuration/ConfigurationTests.cs
@@ -83,7 +83,7 @@ public void GetBusDetailsTest()
             Check.That(busDetails.SslServerName).IsEqualTo("XComponent RMq");
             Check.That(busDetails.SslCertificatePath).IsEqualTo("some_cert_path");
             Check.That(busDetails.SslCertificatePassphrase).IsEqualTo("some_cert_pass");
-            Check.That(busDetails.SslProtocol).IsEqualTo(SslProtocols.Default);
+            Check.That(busDetails.SslProtocol).IsEqualTo(SslProtocols.Tls12);
             Check.That(busDetails.SslAllowUntrustedServerCertificate).IsTrue();
         }
 
diff --git a/ReactiveXComponentTest/RabbitMq/RabbitMqTests.cs b/ReactiveXComponentTest/RabbitMq/RabbitMqTests.cs
@@ -403,7 +403,7 @@ public void SnapshotTest(int instancesCount, int chunkSize)
                 List<MessageEventArgs> snapshotInstances = null;
                 Task.Run(async () =>
                 {
-                    snapshotInstances = await publisher.GetSnapshotAsync(StateMachineA, chunkSize);
+                    snapshotInstances = await publisher.GetSnapshotAsync(StateMachineA, chunkSize: chunkSize);
                 }).GetAwaiter().OnCompleted(() =>
                 {
                     snapshotReceivedEvent.Set();
diff --git a/ReactiveXComponentTest/RabbitMqTestApi.xcApi b/ReactiveXComponentTest/RabbitMqTestApi.xcApi
@@ -3,7 +3,7 @@
   <threading />
   <serialization>Binary</serialization>
   <communication>
-    <bus name="rabbitmq" host="127.0.0.1" virtualHost="myVirtualHost" port="5671" user="guest" password="guest" type="RABBIT_MQ" sslEnabled="True" sslServerName="XComponent RMq" sslCertPath="some_cert_path" sslCertPassphrase="some_cert_pass" sslProtocol="Default" sslAllowUntrustedServerCertificate="True" />
+    <bus name="rabbitmq" host="127.0.0.1" virtualHost="myVirtualHost" port="5671" user="guest" password="guest" type="RABBIT_MQ" sslEnabled="True" sslServerName="XComponent RMq" sslCertPath="some_cert_path" sslCertPassphrase="some_cert_pass" sslProtocol="Tls12" sslAllowUntrustedServerCertificate="True" />
   </communication>
   <clientAPICommunication>
     <publish componentCode="-69981087" stateMachineCode="-829536631" eventType="UPDATE" topicType="output" communicationType="BUS" stateCode="0" eventCode="9" event="XComponent.HelloWorld.UserObject.SayHello" communication="rabbitmq">
diff --git a/ReactiveXComponentTest/WebSocket/WebSocketTests.cs b/ReactiveXComponentTest/WebSocket/WebSocketTests.cs
@@ -305,7 +305,7 @@ public void SnapshotTest()
                 List<MessageEventArgs> instances = new List<MessageEventArgs>();
                 Task.Run(async () =>
                 {
-                    instances = await webSocketPublisher.GetSnapshotAsync(stateMachineName, 2);
+                    instances = await webSocketPublisher.GetSnapshotAsync(stateMachineName, chunkSize:2);
                 }).GetAwaiter().OnCompleted(() =>
                 {
                     snapshotReceivedEvent.Set();
diff --git a/docker/integration_tests/XCProjects/HelloWorldV5/CreateInstancesReactiveApi/CreateInstances/Program.cs b/docker/integration_tests/XCProjects/HelloWorldV5/CreateInstancesReactiveApi/CreateInstances/Program.cs
@@ -28,18 +28,25 @@ class ProgramInBatchWithReactiveApi
 
         static int Main()
         {
-            for (int i = 0; i < MaxNumberOfTries; i++)
+            var triesCount = 0;
+
+            while (triesCount < MaxNumberOfTries)
             {
                 try
                 {
+                    triesCount++;
                     int result = TryToRunTest();
                     return result;
                 }
                 catch (Exception e)
                 {
-                    throw new ReactiveXComponentException("Error while running test: " + e.Message, e);
+                    if (triesCount == MaxNumberOfTries)
+                    {
+                        throw new ReactiveXComponentException("Error while running test: " + e.Message, e);
+                    }
                 }
             }
+
             Console.WriteLine("Exited test, failed to run");
             return TooManyTriesExitCode;
         }
@@ -60,6 +67,8 @@ private static int TryToRunTest()
                 {
                     TestCreateInstanceReactiveApi(publisher, createdInstanceIds, subscriber);
 
+                    TestGetSnapshotWithFilter(publisher, createdInstanceIds);
+
                     TestDestroyInstanceReactiveApi(subscriber, createdInstanceIds, publisher);
                     
                     return 0;
@@ -94,7 +103,7 @@ private static void TestDestroyInstanceReactiveApi(IXCSubscriber subscriber, Con
                 List<MessageEventArgs> instanceList = createdInstanceIds.Values.ToList();
                 foreach (var helloWorldResponseInstance in instanceList)
                 {
-                    publisher.SendEvent(helloWorldResponseInstance.StateMachineRefHeader, new SayGoodbye() {Name = "toto"});
+                    publisher.SendEvent(helloWorldResponseInstance.StateMachineRefHeader, new SayGoodbye());
                 }
 
                 if (!waitForInstanceDestruction.WaitOne(TimeToWaitForInstanceState))
@@ -133,7 +142,7 @@ private static void TestCreateInstanceReactiveApi(IXCPublisher publisher, Concur
             {
                 for (int i = 0; i < NumberOfInstances; i++)
                 {
-                    publisher.SendEvent(entryPointSnapshot.StateMachineRefHeader, new SayHello() {Name = "toto"});
+                    publisher.SendEvent(entryPointSnapshot.StateMachineRefHeader, new SayHello() {Name = $"toto{i}"});
                 }
 
                 if (!waitForInstanceCreation.WaitOne(TimeToWaitForInstanceState))
@@ -144,6 +153,32 @@ private static void TestCreateInstanceReactiveApi(IXCPublisher publisher, Concur
             }
         }
 
+        private static void TestGetSnapshotWithFilter(
+            IXCPublisher publisher, 
+            ConcurrentDictionary<string, MessageEventArgs> createdInstanceIds)
+        {
+            Console.WriteLine("Testing get snapshot with filter");
+
+            for (int i = 0; i < NumberOfInstances; i++)
+            {
+                var filter = $"OriginatorName == toto{i}";
+                var instances = publisher.GetSnapshot(ResponseStateMachineName, filter);
+
+                if (instances == null)
+                {
+                    Console.WriteLine($"Received a null snapshot for instance with filter = {filter}");
+                    throw new Exception("Error while testing snapshot");
+                }
+
+                if (instances.Count != 1)
+                {
+                    Console.WriteLine($"Received {instances.Count} instances for get snapshot for instance with filter = {filter}. Expected only one.");
+                    throw new Exception("Error while testing snapshot");
+                }
+            }
+            
+        }
+
         private static void ObserverStateMachineExecuted(int instanceCount, AutoResetEvent waitForInstanceState, string stateInstance)
         {
             Console.WriteLine($"{instanceCount} instance " + stateInstance);

Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ public interface IXCPublisher : IDisposable`
`10`	`10`	`void SendEvent(string stateMachine, object message, string messageType, Visibility visibility = Visibility.Public);`
`11`	`11`	`void SendEvent(string stateMachine, object message, Visibility visibility = Visibility.Public);`
`12`	`12`	`void SendEvent(StateMachineRefHeader stateMachineRefHeader, object message, Visibility visibility = Visibility.Public);`
`13`		`- List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000);`
`14`		`- Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000);`
	`13`	`+ List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000);`
	`14`	`+ Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000);`
`15`	`15`	`}`
`16`	`16`	`}`
Original file line number	Diff line number	Diff line change
`@@ -70,14 +70,14 @@ public void SendEvent(StateMachineRefHeader stateMachineRefHeader, object messag`
`70`	`70`	`Send(message, routingKey, prop);`
`71`	`71`	`}`
`72`	`72`
`73`		`- public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize, int timeout = 10000)`
	`73`	`+ public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`74`	`74`	`{`
`75`		`- return _rabbitMqSnapshotManager.GetSnapshot(stateMachine, chunkSize, timeout);`
	`75`	`+ return _rabbitMqSnapshotManager.GetSnapshot(stateMachine, filter, chunkSize, timeout);`
`76`	`76`	`}`
`77`	`77`
`78`		`- public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize, int timeout = 10000)`
	`78`	`+ public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`79`	`79`	`{`
`80`		`- return _rabbitMqSnapshotManager.GetSnapshotAsync(stateMachine, chunkSize, timeout);`
	`80`	`+ return _rabbitMqSnapshotManager.GetSnapshotAsync(stateMachine, filter, chunkSize, timeout);`
`81`	`81`	`}`
`82`	`82`
`83`	`83`	`#endregion`
Original file line number	Diff line number	Diff line change
`@@ -73,7 +73,7 @@ private static SnapshotResponse AggregateChunks(ConcurrentBag<SnapshotResponseCh`
`73`	`73`	`return aggregatedResult;`
`74`	`74`	`}`
`75`	`75`
`76`		`- public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000)`
	`76`	`+ public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`77`	`77`	`{`
`78`	`78`	`var guid = Guid.NewGuid();`
`79`	`79`	`var requestId = guid.ToString();`
`@@ -119,7 +119,7 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =`
`119`	`119`	`SnapshotReceived += snapshotListenerOnMessageReceived;`
`120`	`120`
`121`	`121`	`SubscribeSnapshot(stateMachine, requestId);`
`122`		`- SendSnapshotRequest(stateMachine, requestId, chunkSize, _privateCommunicationIdentifier);`
	`122`	`+ SendSnapshotRequest(stateMachine, requestId, filter, chunkSize, _privateCommunicationIdentifier);`
`123`	`123`
`124`	`124`	`if (receivedSnapshotChunksInitialized.WaitOne(timeout))`
`125`	`125`	`{`
`@@ -183,12 +183,12 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =`
`183`	`183`	`return result;`
`184`	`184`	`}`
`185`	`185`
`186`		`- public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000)`
	`186`	`+ public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`187`	`187`	`{`
`188`		`- return Task.Run(() => GetSnapshot(stateMachine, chunkSize, timeout));`
	`188`	`+ return Task.Run(() => GetSnapshot(stateMachine, filter, chunkSize, timeout));`
`189`	`189`	`}`
`190`	`190`
`191`		`- private void SendSnapshotRequest(string stateMachine, string replyTopic, int? chunkSize, string privateCommunicationIdentifier = null)`
	`191`	`+ private void SendSnapshotRequest(string stateMachine, string replyTopic, string filter = null, int? chunkSize = null, string privateCommunicationIdentifier = null)`
`192`	`192`	`{`
`193`	`193`	`if (_xcConfiguration == null)`
`194`	`194`	`return;`
`@@ -212,6 +212,7 @@ private void SendSnapshotRequest(string stateMachine, string replyTopic, int? ch`
`212`	`212`	`CallerPrivateTopic = !string.IsNullOrEmpty(privateCommunicationIdentifier)`
`213`	`213`	`? new List<string>{ privateCommunicationIdentifier}`
`214`	`214`	`: null,`
	`215`	`+ Filter = filter,`
`215`	`216`	`ChunkSize = chunkSize`
`216`	`217`	`};`
`217`	`218`
Original file line number	Diff line number	Diff line change
`@@ -65,14 +65,14 @@ public void SendEvent(StateMachineRefHeader stateMachineRefHeader, object messag`
`65`	`65`	`_webSocketClient.Send(webSocketRequest);`
`66`	`66`	`}`
`67`	`67`
`68`		`- public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize, int timeout = 10000)`
	`68`	`+ public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`69`	`69`	`{`
`70`		`- return _webSocketSnapshotManager.GetSnapshot(stateMachine, chunkSize, timeout);`
	`70`	`+ return _webSocketSnapshotManager.GetSnapshot(stateMachine, filter, chunkSize, timeout);`
`71`	`71`	`}`
`72`	`72`
`73`		`- public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize, int timeout = 10000)`
	`73`	`+ public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`74`	`74`	`{`
`75`		`- return _webSocketSnapshotManager.GetSnapshotAsync(stateMachine, chunkSize, timeout);`
	`75`	`+ return _webSocketSnapshotManager.GetSnapshotAsync(stateMachine, filter, chunkSize, timeout);`
`76`	`76`	`}`
`77`	`77`
`78`	`78`	`#endregion`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public WebSocketSnapshotManager(string component, IWebSocketClient webSocketClie`
`43`	`43`	`h => SnapshotReceived -= h);`
`44`	`44`	`}`
`45`	`45`
`46`		`- public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize = null, int timeout = 10000)`
	`46`	`+ public List<MessageEventArgs> GetSnapshot(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`47`	`47`	`{`
`48`	`48`	`var replyTopic = Guid.NewGuid().ToString();`
`49`	`49`
`@@ -74,7 +74,7 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =`
`74`	`74`	`using (_snapshotStream.Subscribe(observer))`
`75`	`75`	`{`
`76`	`76`	`SendWebSocketSnapshotSubscriptionResquest(replyTopic);`
`77`		`- SendWebSocketSnapshotRequest(stateMachine, replyTopic, chunkSize);`
	`77`	`+ SendWebSocketSnapshotRequest(stateMachine, replyTopic, filter, chunkSize);`
`78`	`78`	`lockEvent.WaitOne(timeout);`
`79`	`79`	`}`
`80`	`80`
`@@ -84,12 +84,12 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =`
`84`	`84`	`return result;`
`85`	`85`	`}`
`86`	`86`
`87`		`- public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, int? chunkSize = null, int timeout = 10000)`
	`87`	`+ public Task<List<MessageEventArgs>> GetSnapshotAsync(string stateMachine, string filter = null, int? chunkSize = null, int timeout = 10000)`
`88`	`88`	`{`
`89`		`- return Task.Run(() => GetSnapshot(stateMachine, chunkSize, timeout));`
	`89`	`+ return Task.Run(() => GetSnapshot(stateMachine, filter, chunkSize, timeout));`
`90`	`90`	`}`
`91`	`91`
`92`		`- private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic, int? chunkSize)`
	`92`	`+ private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic, string filter = null, int? chunkSize = null)`
`93`	`93`	`{`
`94`	`94`	`if (!_webSocketClient.IsOpen) return;`
`95`	`95`
`@@ -101,7 +101,7 @@ private void SendWebSocketSnapshotRequest(string stateMachine, string replyTopic`
`101`	`101`	`ComponentCode = componentCode,`
`102`	`102`	`StateMachineCode = stateMachineCode`
`103`	`103`	`};`
`104`		`- var snapshotMessage = new WebSocketSnapshotMessage(stateMachineCode, componentCode, replyTopic, _privateCommunicationIdentifier, chunkSize);`
	`104`	`+ var snapshotMessage = new WebSocketSnapshotMessage(stateMachineCode, componentCode, replyTopic, _privateCommunicationIdentifier, chunkSize, filter);`
`105`	`105`	`var webSocketRequest = WebSocketMessageHelper.SerializeRequest(`
`106`	`106`	`WebSocketCommand.Snapshot,`
`107`	`107`	`inputHeader,`
Original file line number	Diff line number	Diff line change
`@@ -10,10 +10,11 @@ public class WebSocketSnapshotMessage`
`10`	`10`	`public string ReplyTopic { get; set; }`
`11`	`11`	`public string[] CallerPrivateTopic { get; set; }`
`12`	`12`	`public int? ChunkSize { get; set; }`
	`13`	`+ public string Filter { get; set; }`
`13`	`14`
`14`	`15`	`public WebSocketSnapshotMessage() { }`
`15`	`16`
`16`		`- public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, string replyTopic, string callerPrivateTopic, int? chunkSize)`
	`17`	`+ public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, string replyTopic, string callerPrivateTopic, int? chunkSize, string filter = null)`
`17`	`18`	`{`
`18`	`19`	`StateMachineCode = stateMachineCode;`
`19`	`20`	`ComponentCode = componentCode;`
`@@ -23,6 +24,7 @@ public WebSocketSnapshotMessage(long stateMachineCode, long componentCode, strin`
`23`	`24`	`CallerPrivateTopic = new []{ callerPrivateTopic };`
`24`	`25`	`}`
`25`	`26`	`ChunkSize = chunkSize;`
	`27`	`+ Filter = filter;`
`26`	`28`	`}`
`27`	`29`	`}`
`28`	`30`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,7 @@ public void GetBusDetailsTest()`
`83`	`83`	`Check.That(busDetails.SslServerName).IsEqualTo("XComponent RMq");`
`84`	`84`	`Check.That(busDetails.SslCertificatePath).IsEqualTo("some_cert_path");`
`85`	`85`	`Check.That(busDetails.SslCertificatePassphrase).IsEqualTo("some_cert_pass");`
`86`		`- Check.That(busDetails.SslProtocol).IsEqualTo(SslProtocols.Default);`
	`86`	`+ Check.That(busDetails.SslProtocol).IsEqualTo(SslProtocols.Tls12);`
`87`	`87`	`Check.That(busDetails.SslAllowUntrustedServerCertificate).IsTrue();`
`88`	`88`	`}`
`89`	`89`
Original file line number	Diff line number	Diff line change
`@@ -403,7 +403,7 @@ public void SnapshotTest(int instancesCount, int chunkSize)`
`403`	`403`	`List<MessageEventArgs> snapshotInstances = null;`
`404`	`404`	`Task.Run(async () =>`
`405`	`405`	`{`
`406`		`- snapshotInstances = await publisher.GetSnapshotAsync(StateMachineA, chunkSize);`
	`406`	`+ snapshotInstances = await publisher.GetSnapshotAsync(StateMachineA, chunkSize: chunkSize);`
`407`	`407`	`}).GetAwaiter().OnCompleted(() =>`
`408`	`408`	`{`
`409`	`409`	`snapshotReceivedEvent.Set();`