Merge pull request #85 from xcomponent/ISSUE-84

Andrei Marculescu · web-flow · commit 7ab9c10dbc1a · 2018-09-19T17:39:35.000+02:00
Adding support for rabbitmq virtual hosts. Adding WorkerId and MessageIds
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`