secure subscriptions and methodCalls from parallel access using lock

Author: brean

unity-project/Assets/Moulin/DDP/ddp/DdpConnection.cs

@@ -107,6 +107,9 @@ public enum ConnectionState {
 		private ConnectionState ddpConnectionState;
 		private string sessionId;
 
+		private object subscriptionsLock = new object();
+		private object methodCallsLock = new object();
+		
 		private Dictionary<string, Subscription> subscriptions = new Dictionary<string, Subscription>();
 		private Dictionary<string, MethodCall> methodCalls = new Dictionary<string, MethodCall>();
 
@@ -155,14 +158,18 @@ public void OnWebSocketOpen()
         {
             OnDebugMessage?.Invoke("Websocket open");
             Send(GetConnectMessage());
-            foreach (Subscription subscription in subscriptions.Values)
-            {
-                Send(GetSubscriptionMessage(subscription));
-            }
-            foreach (MethodCall methodCall in methodCalls.Values)
-            {
-                Send(GetMethodCallMessage(methodCall));
-            }
+			lock (subscriptionsLock) {
+				foreach (Subscription subscription in subscriptions.Values)
+				{
+					Send(GetSubscriptionMessage(subscription));
+				}
+			}
+			lock (methodCallsLock) {
+				foreach (MethodCall methodCall in methodCalls.Values)
+				{
+					Send(GetMethodCallMessage(methodCall));
+				}
+			}
         }
 
 		private void OnWebSocketError(string reason) {
@@ -177,8 +184,14 @@ private void OnWebSocketClose(bool wasClean) {
 			if (wasClean) {
 				ddpConnectionState = ConnectionState.CLOSED;
 				sessionId = null;
-				subscriptions.Clear();
-				methodCalls.Clear();
+				lock (subscriptionsLock)
+				{
+					subscriptions.Clear();
+				}
+				lock (methodCallsLock)
+				{
+					methodCalls.Clear();
+				}
                 OnDisconnected?.Invoke(this);
 			} else {
 				ddpConnectionState = ConnectionState.DISCONNECTED;
@@ -230,7 +243,10 @@ private void HandleMessage(JSONObject message) {
 
 			    case MessageType.NOSUB: {
 				    string subscriptionId = message[Field.ID].str;
-				    subscriptions.Remove(subscriptionId);
+					lock (subscriptionsLock)
+					{
+						subscriptions.Remove(subscriptionId);
+					}
 
 				    if (message.HasField(Field.ERROR)) {
                         OnError?.Invoke(GetError(message[Field.ERROR]));
@@ -266,7 +282,11 @@ private void HandleMessage(JSONObject message) {
 				    string[] subscriptionIds = ToStringArray(message[Field.SUBS]);
 
 				    foreach (string subscriptionId in subscriptionIds) {
-					    Subscription subscription = subscriptions[subscriptionId];
+					    Subscription subscription;
+						lock (subscriptionsLock)
+						{
+							subscription = subscriptions[subscriptionId];
+						}
 					    if (subscription != null) {
 						    subscription.isReady = true;
                             subscription.OnReady?.Invoke(subscription);
@@ -294,14 +314,21 @@ private void HandleMessage(JSONObject message) {
 
 			    case MessageType.RESULT: {
 					string methodCallId = message[Field.ID].str;
-					MethodCall methodCall = methodCalls[methodCallId];
+					MethodCall methodCall;
+					lock (methodCallsLock)
+					{
+						methodCall = methodCalls[methodCallId];
+					}
 					if (methodCall != null) {
 						if (message.HasField(Field.ERROR)) {
 							methodCall.error = GetError(message[Field.ERROR]);
 						}
 						methodCall.result = message[Field.RESULT];
 						if (methodCall.hasUpdated) {
-							methodCalls.Remove(methodCallId);
+							lock (methodCallsLock)
+							{
+								methodCalls.Remove(methodCallId);
+							}
 						}
 						methodCall.hasResult = true;
                         methodCall.OnResult?.Invoke(methodCall);
@@ -312,10 +339,17 @@ private void HandleMessage(JSONObject message) {
 			    case MessageType.UPDATED: {
 					string[] methodCallIds = ToStringArray(message[Field.METHODS]);
 					foreach (string methodCallId in methodCallIds) {
-						MethodCall methodCall = methodCalls[methodCallId];
+						MethodCall methodCall;
+						lock (methodCallsLock)
+						{
+							methodCall = methodCalls[methodCallId];
+						}
 						if (methodCall != null) {
 							if (methodCall.hasResult) {
-								methodCalls.Remove(methodCallId);
+								lock (methodCallsLock)
+								{
+									methodCalls.Remove(methodCallId);
+								}
 							}
 							methodCall.hasUpdated = true;
                             methodCall.OnUpdated?.Invoke(methodCall);
@@ -503,7 +537,10 @@ public async Task<MethodCall> CallAsync(string methodName, params JSONObject[] i
                 methodName = methodName,
                 items = items
             };
-            methodCalls[methodCall.id] = methodCall;
+			lock (methodCallsLock)
+			{
+				methodCalls[methodCall.id] = methodCall;
+			}
             await SendAsync(GetMethodCallMessage(methodCall));
             return methodCall;
         }
@@ -514,7 +551,10 @@ public MethodCall Call(string methodName, params JSONObject[] items) {
 				methodName = methodName,
 				items = items
 			};
-			methodCalls[methodCall.id] = methodCall;
+			lock (methodCallsLock)
+			{
+				methodCalls[methodCall.id] = methodCall;
+			}
 			Send(GetMethodCallMessage(methodCall));
 			return methodCall;
 		}