OPENICF-454 - Add periodical message to sync request statuses between client-server

Author: Laszlo Hordos

FrameworkProtoBuf/protobuf/RPCMessages.proto

@@ -40,6 +40,8 @@ message ControlRequest {
         REQUESTS = 2;
     }
     repeated InfoLevel infoLevel = 1;
+    repeated int64 localRequestId = 2;
+    repeated int64 remoteRequestId = 3;
 }
 
 message ControlResponse {

FrameworkRPC/Rpc.cs

@@ -177,6 +177,18 @@ protected internal LocalRequest(long requestId, TH socket)
             _remoteConnectionContext.RemoteConnectionGroup.ReceiveRequest<TV, TE, LocalRequest<TV, TE, TG, TH, TP>>(this);
         }
 
+        /// <summary>
+        /// Check if this object was {@ref Inconsistent}-ed and don't dispose.
+        /// </summary>
+        /// <returns> 'true' when object is still active or 'false' when this can be
+        ///         disposed. </returns>
+        public abstract bool Check();
+
+        /// <summary>
+        /// Signs that the object state is inconsistent.
+        /// </summary>
+        public abstract void Inconsistent();
+
         protected internal abstract bool TryHandleResult(TV result);
 
         protected internal abstract bool TryHandleError(TE error);
@@ -783,6 +795,18 @@ protected RemoteRequest(TP context, Int64 requestId,
             _cancellationToken = cancellationToken;
         }
 
+        /// <summary>
+        /// Check if this object was {@ref Inconsistent}-ed and don't dispose.
+        /// </summary>
+        /// <returns> 'true' when object is still active or 'false' when this can be
+        ///         disposed. </returns>
+        public abstract bool Check();
+
+        /// <summary>
+        /// Signs that the object state is inconsistent.
+        /// </summary>
+        public abstract void Inconsistent();
+
         public abstract void HandleIncomingMessage(TH sourceConnection, Object message);
 
         protected internal abstract RemoteConnectionGroup<TG, TH, TP>.AsyncMessageQueueRecord CreateMessageElement(

FrameworkServer/Client.cs

@@ -75,7 +75,7 @@ public ClientRemoteConnectorInfoManager(RemoteWSFrameworkConnectionInfo info,
                             {
                                 if (result.IsCompleted)
                                 {
-                                    connections[(int) o] = result.Result;
+                                    connections[(int)o] = result.Result;
                                 }
                                 else if (result.IsFaulted)
                                 {
@@ -119,7 +119,8 @@ private sealed class RemoteDelegatingAsyncConnectorInfoManager : DelegatingAsync
         {
             private readonly ClientRemoteConnectorInfoManager _parent;
 
-            public RemoteDelegatingAsyncConnectorInfoManager(ClientRemoteConnectorInfoManager parent) : base(true)
+            public RemoteDelegatingAsyncConnectorInfoManager(ClientRemoteConnectorInfoManager parent)
+                : base(true)
             {
                 _parent = parent;
             }
@@ -169,13 +170,29 @@ private readonly ConcurrentDictionary<RemoteWSFrameworkConnectionInfo, ClientRem
             registry =
                 new ConcurrentDictionary<RemoteWSFrameworkConnectionInfo, ClientRemoteConnectorInfoManager>();
 
+        private readonly Timer _timer;
+
         internal RemoteConnectionInfoManagerFactory(
             IMessageListener<WebSocketConnectionGroup, WebSocketConnectionHolder, RemoteOperationContext>
                 messageListener,
             ConnectionManagerConfig managerConfig)
         {
             this._messageListener = messageListener;
             this.managerConfig = managerConfig;
+
+
+            _timer = new Timer(state =>
+            {
+                if (Running)
+                {
+                    foreach (var connectionGroup in connectionGroups.Values)
+                    {
+                        Trace.TraceInformation("Check ConnectionGroup:{0} - operational={1}", connectionGroup.RemoteSessionId
+                            , connectionGroup.Operational);
+                        connectionGroup.CheckIsActive();
+                    }
+                }
+            }, null, TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(4));
         }
 
         public IRemoteConnectorInfoManager Connect(RemoteWSFrameworkConnectionInfo info)
@@ -215,6 +232,7 @@ public IRemoteConnectorInfoManager Connect(RemoteWSFrameworkConnectionInfo info)
 
         protected internal void doClose()
         {
+            _timer.Dispose();
             foreach (var clientRemoteConnectorInfoManager in registry.Values)
             {
                 clientRemoteConnectorInfoManager.Dispose();
@@ -356,7 +374,7 @@ public virtual long HeartbeatInterval
 
     #endregion
 
-    #region sd
+    #region WebSocketWrapper
 
     public class WebSocketWrapper : WebSocketConnectionHolder
     {
@@ -524,15 +542,15 @@ private static void RunInTask(Action action)
         protected override async Task WriteMessageAsync(byte[] entry, WebSocketMessageType messageType)
         {
             var messageBuffer = entry;
-            var messagesCount = (int) Math.Ceiling((double) messageBuffer.Length/SendChunkSize);
+            var messagesCount = (int)Math.Ceiling((double)messageBuffer.Length / SendChunkSize);
 
             for (var i = 0; i < messagesCount; i++)
             {
-                var offset = (SendChunkSize*i);
+                var offset = (SendChunkSize * i);
                 var count = SendChunkSize;
                 var lastMessage = ((i + 1) == messagesCount);
 
-                if ((count*(i + 1)) > messageBuffer.Length)
+                if ((count * (i + 1)) > messageBuffer.Length)
                 {
                     count = messageBuffer.Length - offset;
                 }

FrameworkServer/FrameworkServer.csproj

@@ -64,6 +64,7 @@
     </Reference>
     <Reference Include="System" />
     <Reference Include="Microsoft.CSharp" />
+    <Reference Include="System.Collections.Concurrent" />
   </ItemGroup>
   <ItemGroup>
     <Compile Include="Async.cs" />