#84: adding support for rabbit mq virtual hosts. Adding WorkerId to StateMachineRefHeader to avoid performance issue with context based messages (without the worker id, the message goes to all the workers of the statemachine. Adding message ids to all the events (context based or not) to be able to use the api against a xcomponent cluster

Andrei Marculescu · Andrei Marculescu · commit 75487d4de770 · 2018-09-19T17:06:44.000+02:00
diff --git a/ReactiveXComponent/Common/Header.cs b/ReactiveXComponent/Common/Header.cs
@@ -1,3 +1,5 @@
+using System;
+
 namespace ReactiveXComponent.Common
 {
     public class Header
@@ -17,5 +19,7 @@ public class Header
         public string PublishTopic { get; set; }
 
         public string ErrorMessage { get; set; }
+
+        public string MessageId { get; set; } = Guid.NewGuid().ToString();
     }
 }
diff --git a/ReactiveXComponent/Common/HeaderElement.cs b/ReactiveXComponent/Common/HeaderElement.cs
@@ -13,5 +13,7 @@ public static class HeaderElement
         public const string SessionData = "SessionData";
         public const string EventType = "EventType";
         public const string ErrorMessage = "ErrorMessage";
+        public const string MessageId = "MessageId";
+        public const string WorkerId = "WorkerId";
     }
 }
diff --git a/ReactiveXComponent/Common/StateMachineRefHeader.cs b/ReactiveXComponent/Common/StateMachineRefHeader.cs
@@ -1,37 +1,10 @@
 ﻿
+using System;
+
 namespace ReactiveXComponent.Common
 {
     public class StateMachineRefHeader
     {
-        protected bool Equals(StateMachineRefHeader other)
-        {
-            return StateMachineId == other.StateMachineId && StateCode == other.StateCode && StateMachineCode == other.StateMachineCode && ComponentCode == other.ComponentCode && string.Equals(MessageType, other.MessageType) && string.Equals(PrivateTopic, other.PrivateTopic) && string.Equals(SessionData, other.SessionData) && string.Equals(ErrorMessage, other.ErrorMessage);
-        }
-
-        public override bool Equals(object obj)
-        {
-            if (ReferenceEquals(null, obj)) return false;
-            if (ReferenceEquals(this, obj)) return true;
-            if (obj.GetType() != this.GetType()) return false;
-            return Equals((StateMachineRefHeader) obj);
-        }
-
-        public override int GetHashCode()
-        {
-            unchecked
-            {
-                var hashCode = (StateMachineId != null ? StateMachineId.GetHashCode() : 0);
-                hashCode = (hashCode * 397) ^ StateCode;
-                hashCode = (hashCode * 397) ^ StateMachineCode;
-                hashCode = (hashCode * 397) ^ ComponentCode;
-                hashCode = (hashCode * 397) ^ (MessageType != null ? MessageType.GetHashCode() : 0);
-                hashCode = (hashCode * 397) ^ (PrivateTopic != null ? PrivateTopic.GetHashCode() : 0);
-                hashCode = (hashCode * 397) ^ (SessionData != null ? SessionData.GetHashCode() : 0);
-                hashCode = (hashCode * 397) ^ (ErrorMessage != null ? ErrorMessage.GetHashCode() : 0);
-                return hashCode;
-            }
-        }
-
         public string StateMachineId { get; set; }
 
         public int StateCode { get; set; }
@@ -47,5 +20,9 @@ public override int GetHashCode()
         public string SessionData { get; set; }
 
         public string ErrorMessage { get; set; }
+
+        public string MessageId { get; set; } = Guid.NewGuid().ToString();
+
+        public int WorkerId { get; set; }
     }
 }
diff --git a/ReactiveXComponent/Configuration/BusDetails.cs b/ReactiveXComponent/Configuration/BusDetails.cs
@@ -8,12 +8,14 @@ public BusDetails()
             
         }
 
-        public BusDetails(string username, string password, string host, int port)
+        public BusDetails(string username, string password, string host, string virtualHost, int port)
         {
             Username = username;
             Password = password;
             Host = host;
+            VirtualHost = virtualHost;
             Port = port;
+
         }
 
         public string Username { get; set; }
@@ -22,14 +24,17 @@ public BusDetails(string username, string password, string host, int port)
 
         public string Host { get; set; }
 
+        public string VirtualHost { get; set; }
+
         public int Port { get; set; }
 
         public BusDetails Clone()
         {
             return new BusDetails(
-                (string)Username?.Clone(),
-                (string)Password?.Clone(),
-                (string)Host?.Clone(),
+                Username,
+                Password,
+                Host,
+                VirtualHost,
                 Port);
         }
     }
diff --git a/ReactiveXComponent/Configuration/ConfigurationOverrides.cs b/ReactiveXComponent/Configuration/ConfigurationOverrides.cs
@@ -6,6 +6,8 @@ public class ConfigurationOverrides
     {
         public string Host { get; set; }
 
+        public string VirtualHost { get; set; }
+
         public string Port { get; set; }
 
         public string Username { get; set; }
diff --git a/ReactiveXComponent/Parser/XCApiConfigParser.cs b/ReactiveXComponent/Parser/XCApiConfigParser.cs
@@ -186,6 +186,7 @@ public BusDetails GetBusDetails()
                 busInfos?.Attribute("user")?.Value,
                 busInfos?.Attribute("password")?.Value,
                 busInfos?.Attribute("host")?.Value,
+                busInfos?.Attribute("virtualHost")?.Value,
                 Convert.ToInt32(busInfos?.Attribute("port")?.Value));
 
             return busDetails;
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqConnection.cs b/ReactiveXComponent/RabbitMq/RabbitMqConnection.cs
@@ -30,6 +30,11 @@ public IXCSession CreateSession(ConfigurationOverrides configurationOverrides =
                 busDetails.Host = configurationOverrides.Host;
             }
 
+            if (!string.IsNullOrEmpty(configurationOverrides.VirtualHost))
+            {
+                busDetails.VirtualHost = configurationOverrides.VirtualHost;
+            }
+
             if (configurationOverrides.Port != null)
             {
                 busDetails.Port = int.Parse(configurationOverrides.Port);
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqHeaderConverter.cs b/ReactiveXComponent/RabbitMq/RabbitMqHeaderConverter.cs
@@ -21,7 +21,10 @@ public static Dictionary<string, object> ConvertHeader(Header header)
                                {HeaderElement.IncomingEventType, header?.IncomingEventType},
                                {HeaderElement.PublishTopic, header?.PublishTopic != null ? encoding.GetBytes(header.PublishTopic) : encoding.GetBytes(string.Empty) },
                                {HeaderElement.MessageType, header?.MessageType != null ? encoding.GetBytes(header.MessageType) : encoding.GetBytes(string.Empty)},
-                               {HeaderElement.ErrorMessage, header?.ErrorMessage != null ? encoding.GetBytes(header.ErrorMessage) : encoding.GetBytes(string.Empty)}
+                               {HeaderElement.ErrorMessage, header?.ErrorMessage != null ? encoding.GetBytes(header.ErrorMessage) : encoding.GetBytes(string.Empty)},
+                               {HeaderElement.MessageId, header?.MessageId != null ? encoding.GetBytes(header.MessageId) : encoding.GetBytes(string.Empty) },
+
+
                            };
             return dico;
         }
@@ -40,7 +43,10 @@ public static Dictionary<string, object> CreateHeaderFromStateMachineRefHeader(S
                                 {HeaderElement.MessageType, stateMachineRefHeader?.MessageType != null ? encoding.GetBytes(stateMachineRefHeader.MessageType) : encoding.GetBytes(string.Empty)},
                                 {HeaderElement.EventType, eventCode},
                                 {HeaderElement.IncomingEventType, (int)incomingEventType},
-                                {HeaderElement.ErrorMessage, stateMachineRefHeader?.ErrorMessage != null ? encoding.GetBytes(stateMachineRefHeader.ErrorMessage) : encoding.GetBytes(string.Empty)}
+                                {HeaderElement.ErrorMessage, stateMachineRefHeader?.ErrorMessage != null ? encoding.GetBytes(stateMachineRefHeader.ErrorMessage) : encoding.GetBytes(string.Empty)},
+                                {HeaderElement.MessageId, stateMachineRefHeader?.MessageId != null ? encoding.GetBytes(stateMachineRefHeader.MessageId) : encoding.GetBytes(string.Empty) },
+                                {HeaderElement.WorkerId, stateMachineRefHeader?.WorkerId},
+
                            };
             return dico;
         }
@@ -56,6 +62,8 @@ public static StateMachineRefHeader ConvertStateMachineRefHeader(IDictionary<str
             var messageType = string.Empty;
             var sessionData = string.Empty;
             var errorMessage = string.Empty;
+            var messageId = string.Empty;
+            var workerId = -1;
 
             if (stateMachineRefHeader.ContainsKey(HeaderElement.StateMachineId) && stateMachineRefHeader[HeaderElement.StateMachineId] != null)
                 stateMachineId = Encoding.UTF8.GetString((byte[])stateMachineRefHeader[HeaderElement.StateMachineId]);
@@ -73,6 +81,10 @@ public static StateMachineRefHeader ConvertStateMachineRefHeader(IDictionary<str
                 sessionData = encoding.GetString(stateMachineRefHeader[HeaderElement.SessionData] as byte[]);
             if (stateMachineRefHeader.ContainsKey(HeaderElement.ErrorMessage))
                 errorMessage = encoding.GetString(stateMachineRefHeader[HeaderElement.ErrorMessage] as byte[]);
+            if (stateMachineRefHeader.ContainsKey(HeaderElement.MessageId))
+                messageId = encoding.GetString(stateMachineRefHeader[HeaderElement.MessageId] as byte[]);
+            if (stateMachineRefHeader.ContainsKey(HeaderElement.WorkerId))
+                workerId = Convert.ToInt32(stateMachineRefHeader[HeaderElement.WorkerId]);
 
             return new StateMachineRefHeader()
             {
@@ -83,7 +95,9 @@ public static StateMachineRefHeader ConvertStateMachineRefHeader(IDictionary<str
                 MessageType  = messageType,
                 PrivateTopic = publishTopic,
                 SessionData = sessionData,
-                ErrorMessage = errorMessage
+                ErrorMessage = errorMessage,
+                MessageId = messageId,
+                WorkerId = workerId,
             };                
         }
     }
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqPublisher.cs b/ReactiveXComponent/RabbitMq/RabbitMqPublisher.cs
@@ -110,7 +110,7 @@ private Header CreateHeader(string component, string stateMachine, object messag
                 StateCode = defaultValue,
                 EventCode = _configuration.GetPublisherEventCode(messageType),
                 IncomingEventType = (int)IncomingEventType.Transition,
-                PublishTopic = visibility == Visibility.Private && !string.IsNullOrEmpty(_privateCommunicationIdentifier)? _privateCommunicationIdentifier : string.Empty                
+                PublishTopic = visibility == Visibility.Private && !string.IsNullOrEmpty(_privateCommunicationIdentifier)? _privateCommunicationIdentifier : string.Empty,
             };
 
             return header;
@@ -133,7 +133,9 @@ private StateMachineRefHeader CreateStateMachineRefHeader(StateMachineRefHeader
                 StateMachineCode = stateMachineRefHeader.StateMachineCode,
                 ComponentCode = stateMachineRefHeader.ComponentCode,
                 MessageType = messageType,
-                PrivateTopic = visibility == Visibility.Private && !string.IsNullOrEmpty(_privateCommunicationIdentifier) ? _privateCommunicationIdentifier : string.Empty
+                PrivateTopic = visibility == Visibility.Private && !string.IsNullOrEmpty(_privateCommunicationIdentifier) ? _privateCommunicationIdentifier : string.Empty,
+                MessageId = stateMachineRefHeader.MessageId,
+                WorkerId = stateMachineRefHeader.WorkerId,
             };
 
             return stateMachineRefheader;
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqSession.cs b/ReactiveXComponent/RabbitMq/RabbitMqSession.cs
@@ -33,7 +33,7 @@ private void InitConnection(BusDetails busDetails)
                 {
                     UserName = busDetails.Username ?? XCApiTags.DefaultUserName,
                     Password = busDetails.Password ?? XCApiTags.DefaultPassword,
-                    VirtualHost = ConnectionFactory.DefaultVHost,
+                    VirtualHost = string.IsNullOrEmpty(busDetails.VirtualHost) ? ConnectionFactory.DefaultVHost : busDetails.VirtualHost,
                     HostName = busDetails.Host,
                     Port = busDetails.Port,
                     Protocol = Protocols.DefaultProtocol
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqSnapshotManager.cs b/ReactiveXComponent/RabbitMq/RabbitMqSnapshotManager.cs
@@ -134,7 +134,8 @@ public List<MessageEventArgs> GetSnapshot(string stateMachine, int? chunkSize =
                                     StateMachineId = item.StateMachineId,
                                     StateMachineCode = item.StateMachineCode,
                                     ComponentCode = item.ComponentCode,
-                                    StateCode = item.StateCode
+                                    StateCode = item.StateCode,
+                                    WorkerId = item.WorkerId,
                                 },
                                 item.PublicMember,
                                 _serializationType)).ToList();
diff --git a/ReactiveXComponent/RabbitMq/RabbitMqSubscriber.cs b/ReactiveXComponent/RabbitMq/RabbitMqSubscriber.cs
diff --git a/ReactiveXComponentTest/Configuration/ConfigurationTests.cs b/ReactiveXComponentTest/Configuration/ConfigurationTests.cs
diff --git a/ReactiveXComponentTest/Connection/ConnectionTests.cs b/ReactiveXComponentTest/Connection/ConnectionTests.cs
diff --git a/ReactiveXComponentTest/RabbitMq/RabbitMqTests.cs b/ReactiveXComponentTest/RabbitMq/RabbitMqTests.cs
diff --git a/ReactiveXComponentTest/RabbitMqTestApi.xcApi b/ReactiveXComponentTest/RabbitMqTestApi.xcApi

Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,5 @@`
	`1`	`+using System;`
	`2`	`+`
`1`	`3`	`namespace ReactiveXComponent.Common`
`2`	`4`	`{`
`3`	`5`	`public class Header`
`@@ -17,5 +19,7 @@ public class Header`
`17`	`19`	`public string PublishTopic { get; set; }`
`18`	`20`
`19`	`21`	`public string ErrorMessage { get; set; }`
	`22`	`+`
	`23`	`+ public string MessageId { get; set; } = Guid.NewGuid().ToString();`
`20`	`24`	`}`
`21`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -13,5 +13,7 @@ public static class HeaderElement`
`13`	`13`	`public const string SessionData = "SessionData";`
`14`	`14`	`public const string EventType = "EventType";`
`15`	`15`	`public const string ErrorMessage = "ErrorMessage";`
	`16`	`+ public const string MessageId = "MessageId";`
	`17`	`+ public const string WorkerId = "WorkerId";`
`16`	`18`	`}`
`17`	`19`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,12 +8,14 @@ public BusDetails()`
`8`	`8`
`9`	`9`	`}`
`10`	`10`
`11`		`- public BusDetails(string username, string password, string host, int port)`
	`11`	`+ public BusDetails(string username, string password, string host, string virtualHost, int port)`
`12`	`12`	`{`
`13`	`13`	`Username = username;`
`14`	`14`	`Password = password;`
`15`	`15`	`Host = host;`
	`16`	`+ VirtualHost = virtualHost;`
`16`	`17`	`Port = port;`
	`18`	`+`
`17`	`19`	`}`
`18`	`20`
`19`	`21`	`public string Username { get; set; }`
`@@ -22,14 +24,17 @@ public BusDetails(string username, string password, string host, int port)`
`22`	`24`
`23`	`25`	`public string Host { get; set; }`
`24`	`26`
	`27`	`+ public string VirtualHost { get; set; }`
	`28`	`+`
`25`	`29`	`public int Port { get; set; }`
`26`	`30`
`27`	`31`	`public BusDetails Clone()`
`28`	`32`	`{`
`29`	`33`	`return new BusDetails(`
`30`		`- (string)Username?.Clone(),`
`31`		`- (string)Password?.Clone(),`
`32`		`- (string)Host?.Clone(),`
	`34`	`+ Username,`
	`35`	`+ Password,`
	`36`	`+ Host,`
	`37`	`+ VirtualHost,`
`33`	`38`	`Port);`
`34`	`39`	`}`
`35`	`40`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,8 @@ public class ConfigurationOverrides`
`6`	`6`	`{`
`7`	`7`	`public string Host { get; set; }`
`8`	`8`
	`9`	`+ public string VirtualHost { get; set; }`
	`10`	`+`
`9`	`11`	`public string Port { get; set; }`
`10`	`12`
`11`	`13`	`public string Username { get; set; }`
Original file line number	Diff line number	Diff line change
`@@ -30,6 +30,11 @@ public IXCSession CreateSession(ConfigurationOverrides configurationOverrides =`
`30`	`30`	`busDetails.Host = configurationOverrides.Host;`
`31`	`31`	`}`
`32`	`32`
	`33`	`+ if (!string.IsNullOrEmpty(configurationOverrides.VirtualHost))`
	`34`	`+ {`
	`35`	`+ busDetails.VirtualHost = configurationOverrides.VirtualHost;`
	`36`	`+ }`
	`37`	`+`
`33`	`38`	`if (configurationOverrides.Port != null)`
`34`	`39`	`{`
`35`	`40`	`busDetails.Port = int.Parse(configurationOverrides.Port);`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ private void InitConnection(BusDetails busDetails)`
`33`	`33`	`{`
`34`	`34`	`UserName = busDetails.Username ?? XCApiTags.DefaultUserName,`
`35`	`35`	`Password = busDetails.Password ?? XCApiTags.DefaultPassword,`
`36`		`- VirtualHost = ConnectionFactory.DefaultVHost,`
	`36`	`+ VirtualHost = string.IsNullOrEmpty(busDetails.VirtualHost) ? ConnectionFactory.DefaultVHost : busDetails.VirtualHost,`
`37`	`37`	`HostName = busDetails.Host,`
`38`	`38`	`Port = busDetails.Port,`
`39`	`39`	`Protocol = Protocols.DefaultProtocol`