fix issues with websocket (#80)

Fixing issues with websocket

Author: fredericcarre

ReactiveXComponent/Connection/IXCPublisher.cs

@@ -7,6 +7,7 @@ namespace ReactiveXComponent.Connection
 {
     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 timeout = 10000);

ReactiveXComponent/RabbitMq/RabbitMqPublisher.cs

@@ -32,9 +32,9 @@ public RabbitMqPublisher(string component, IXCConfiguration configuration, IConn
 
         #region IXCPublisher implementation
 
-        public void SendEvent(string stateMachine, object message, Visibility visibility = Visibility.Public)
+        public void SendEvent(string stateMachine, object message, string messageType, Visibility visibility = Visibility.Public)
         {
-            var header = CreateHeader(_component, stateMachine, message, visibility);
+            var header = CreateHeader(_component, stateMachine, message, messageType, visibility);
 
             if (header == null) return;
 
@@ -47,6 +47,11 @@ public void SendEvent(string stateMachine, object message, Visibility visibility
             Send(message, routingKey, prop);
         }
 
+        public void SendEvent(string stateMachine, object message, Visibility visibility = Visibility.Public)
+        {
+            SendEvent(stateMachine, message, message?.GetType().ToString() ?? string.Empty, visibility);
+        }
+
         public void SendEvent(StateMachineRefHeader stateMachineRefHeader, object message, Visibility visibility = Visibility.Public)
         {
             if (stateMachineRefHeader == null) return;
@@ -87,10 +92,9 @@ private void CreatePublisherChannel(IConnection connection)
             _publisherChannel.ExchangeDeclare(_exchangeName, ExchangeType.Topic);
         }
 
-        private Header CreateHeader(string component, string stateMachine, object message, Visibility visibility)
+        private Header CreateHeader(string component, string stateMachine, object message, string messageType, Visibility visibility)
         {
             var defaultValue = -1;
-            var messageType = message?.GetType().ToString() ?? string.Empty;
 
             if (_configuration == null)
             {

ReactiveXComponent/WebSocket/WebSocketPublisher.cs

@@ -26,18 +26,23 @@ public WebSocketPublisher(string component, IWebSocketClient webSocketClient, IX
         #region IXCPublisher implementation
 
         public void SendEvent(string stateMachine, object message, Visibility visibility = Visibility.Public)
+        {
+            SendEvent(stateMachine, message, message?.GetType().ToString(), visibility);
+        }
+
+        public void SendEvent(string stateMachine, object message, string messageType, Visibility visibility = Visibility.Public)
         {
             if (!_webSocketClient.IsOpen) return;
-            
-            var inputHeader = CreateWebSocketHeader(stateMachine, message, visibility);
+
+            var inputHeader = CreateWebSocketHeader(stateMachine, messageType, visibility);
             var componentCode = _xcConfiguration.GetComponentCode(_component);
             var topic = _xcConfiguration.GetPublisherTopic(_component, stateMachine);
             var webSocketRequest = WebSocketMessageHelper.SerializeRequest(
-                    WebSocketCommand.Input,
-                    inputHeader,
-                    message,
-                    componentCode.ToString(),
-                    topic);
+                WebSocketCommand.Input,
+                inputHeader,
+                message,
+                componentCode.ToString(),
+                topic);
 
             _webSocketClient.Send(webSocketRequest);
         }
@@ -71,15 +76,14 @@ public void GetSnapshotAsync(string stateMachine, Action<List<MessageEventArgs>>
 
         #endregion
 
-        private WebSocketEngineHeader CreateWebSocketHeader(string stateMachine, object message, Visibility visibility = Visibility.Public)
+        private WebSocketEngineHeader CreateWebSocketHeader(string stateMachine, string messageType, Visibility visibility = Visibility.Public)
         {
-            var messageType = message?.GetType();
             var webSocketEngineHeader = new WebSocketEngineHeader
             {
                 ComponentCode = _xcConfiguration.GetComponentCode(_component),
                 StateMachineCode = _xcConfiguration.GetStateMachineCode(_component, stateMachine),
-                EventCode = _xcConfiguration.GetPublisherEventCode(messageType?.ToString()),
-                MessageType = messageType?.ToString(),
+                EventCode = _xcConfiguration.GetPublisherEventCode(messageType),
+                MessageType = messageType,
                 PublishTopic = visibility == Visibility.Private && !string.IsNullOrEmpty(_privateCommunicationIdentifier)? _privateCommunicationIdentifier : null
             };
 

ReactiveXComponent/WebSocket/WebSocketXCApiManager.cs

@@ -1,7 +1,10 @@
 using System;
 using System.Collections.Generic;
+using System.IO;
+using System.IO.Compression;
 using System.Reactive;
 using System.Reactive.Linq;
+using System.Text;
 using System.Threading;
 using Newtonsoft.Json;
 using ReactiveXComponent.Common;
@@ -62,8 +65,18 @@ public List<string> GetXCApiList(string requestId = null, TimeSpan ? timeout = n
 
             return result;
         }
+     
+        private string UnzipString(byte[] data)
+        {
+            using (var memoryStream = new MemoryStream(data, false))
+            using (var gzipStream = new GZipStream(memoryStream, CompressionMode.Decompress))
+            using (var reader = new StreamReader(gzipStream, Encoding.UTF8))
+            {
+                return reader.ReadToEnd();
+            }
+        }
 
-
+      
         public string GetXCApi(string apiFullName, string requestId = null, TimeSpan ? timeout = null)
         {
             var delay = timeout ?? TimeSpan.FromSeconds(10);
@@ -73,10 +86,11 @@ public string GetXCApi(string apiFullName, string requestId = null, TimeSpan ? t
             {
                 if (response.RequestId == requestId && response.ApiFound)
                 {
-                    result = response.ApiName;
+                    result = UnzipString(Convert.FromBase64String(response.Content));
                     lockEvent.Set();
                 }
             });
+            
             var request = new WebSocketXCApiCommand
             {
                 Command = WebSocketCommand.GetXCApi,

ReactiveXComponent/XComponentApi.cs

@@ -3,6 +3,7 @@
 using ReactiveXComponent.Configuration;
 using ReactiveXComponent.Connection;
 using ReactiveXComponent.Parser;
+using ReactiveXComponent.WebSocket;
 
 namespace ReactiveXComponent
 {
@@ -20,6 +21,16 @@ private XComponentApi(Stream xcApiStream, string privateCommunicationIdentifier
             _xcConnection = connectionFactory.CreateConnection(connectionType);
         }
 
+        private XComponentApi(Stream xcApiStream, WebSocketEndpoint webSocketEndpoint, string privateCommunicationIdentifier = null)
+        {
+            var parser = new XCApiConfigParser();
+            var xcConfiguration = new XCConfiguration(parser);
+            xcConfiguration.Init(xcApiStream);
+            _xcConnection = new WebSocketConnection(xcConfiguration, webSocketEndpoint, 10000, privateCommunicationIdentifier);
+        }
+
+
+
         public static IXComponentApi CreateFromXCApi(string xcApiFilePath, string privateCommunicationIdentifier = null)
         {
             if (!File.Exists(xcApiFilePath))
@@ -38,6 +49,11 @@ public static IXComponentApi CreateFromXCApi(Stream xcApiStream, string privateC
             return new XComponentApi(xcApiStream, privateCommunicationIdentifier);
         }
 
+        public static IXComponentApi CreateFromXCApi(Stream xcApiStream, WebSocketEndpoint webSocketEndpoint, string privateCommunicationIdentifier = null)
+        {
+            return new XComponentApi(xcApiStream, webSocketEndpoint, privateCommunicationIdentifier);
+        }
+
         public IXCSession CreateSession(ConfigurationOverrides configurationOverrides = null)
         {
             return _xcConnection.CreateSession(configurationOverrides);