Mercurial > hg > ucis.core

comparison Pml/PmlCommunicator2.cs @ 104:327be9216006

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Improved PML code
author Ivo Smits <Ivo@UCIS.nl>
date Sat, 11 Oct 2014 14:05:41 +0200
parents 8fe322656807
children
comparison
equal deleted inserted replaced
103:8fe322656807 104:327be9216006
5 5
6 namespace UCIS.Pml { 6 namespace UCIS.Pml {
7 public class PmlCommunicator { 7 public class PmlCommunicator {
8 private class CSyncRequest { 8 private class CSyncRequest {
9 internal PmlElement Reply; 9 internal PmlElement Reply;
10 internal ManualResetEvent ResetEvent = new ManualResetEvent(false);
11 } 10 }
12 private interface ISession { 11 private interface ISession {
13 void MessageIn(PmlElement message); 12 void MessageIn(PmlElement message);
14 void CloseIn(); 13 void CloseIn();
15 UInt32 ID { get; } 14 UInt32 ID { get; }
142 } 141 }
143 } 142 }
144 _sessions.Clear(); 143 _sessions.Clear();
145 } 144 }
146 lock (_invocations) { 145 lock (_invocations) {
147 foreach (CSyncRequest T in _invocations.Values) { 146 foreach (CSyncRequest T in _invocations.Values) lock (T) Monitor.Pulse(T);
148 T.ResetEvent.Set();
149 }
150 _invocations.Clear(); 147 _invocations.Clear();
151 } 148 }
152 if (Closed != null) Closed(this, new EventArgs()); 149 if (Closed != null) Closed(this, new EventArgs());
153 } 150 }
154 151
190 Console.WriteLine("UCIS.PML.Connection.Worker Invalid request ID in reply: " + SID.ToString()); 187 Console.WriteLine("UCIS.PML.Connection.Worker Invalid request ID in reply: " + SID.ToString());
191 } 188 }
192 } 189 }
193 if (SRequest != null) { 190 if (SRequest != null) {
194 SRequest.Reply = Message.GetChild("MSG"); 191 SRequest.Reply = Message.GetChild("MSG");
195 SRequest.ResetEvent.Set(); 192 lock (SRequest) Monitor.Pulse(SRequest);
196 } 193 }
197 } else if (Cmd.Equals("REQ") || Cmd.Equals("MSG")) { 194 } else if (Cmd.Equals("REQ") || Cmd.Equals("MSG")) {
198 UCIS.ThreadPool.RunCall(processCall, Message); 195 UCIS.ThreadPool.RunCall(processCall, Message);
199 } else { 196 } else {
200 Console.WriteLine("UCIS.PML.Connection.Worker Invalid command received"); 197 Console.WriteLine("UCIS.PML.Connection.Worker Invalid command received");
286 } 283 }
287 } 284 }
288 } 285 }
289 286
290 public void Call(PmlElement message) { 287 public void Call(PmlElement message) {
291 PmlDictionary Msg = new PmlDictionary(); 288 PmlDictionary Msg = new PmlDictionary() { { "CMD", "MSG" }, { "MSG", message } };
292 Msg.Add("CMD", new PmlString("MSG"));
293 Msg.Add("MSG", message);
294 _WriteMessage(Msg); 289 _WriteMessage(Msg);
295 } 290 }
296 public PmlElement Invoke(PmlElement message) { 291 public PmlElement Invoke(PmlElement message) {
297 return Invoke(message, 60000); 292 return Invoke(message, 60000);
298 } 293 }
299 public PmlElement Invoke(PmlElement message, int timeout) { 294 public PmlElement Invoke(PmlElement message, int timeout) {
300 if (_closed) throw new InvalidOperationException("Sorry, we're closed."); 295 if (_closed) throw new InvalidOperationException("Sorry, we're closed.");
301 CSyncRequest SyncEvent = new CSyncRequest(); 296 CSyncRequest SyncEvent = new CSyncRequest();
302 UInt32 SID = GetNextSessionId(false); 297 UInt32 SID;
303 lock (_invocations) _invocations.Add(SID, SyncEvent); 298 lock (_invocations) {
299 SID = GetNextSessionId(ref pNextSyncRequest, _invocations);
300 _invocations.Add(SID, SyncEvent);
301 }
304 try { 302 try {
305 WriteSyncMessage(SID, false, message); 303 WriteSyncMessage(SID, false, message);
306 if (!SyncEvent.ResetEvent.WaitOne(timeout, false)) { 304 Boolean success;
307 if (!_closed) lock (_invocations) _invocations.Remove(SID); 305 lock (SyncEvent) success = Monitor.Wait(SyncEvent, timeout);
308 throw new TimeoutException("The SyncRequest timed out (SID=" + SID.ToString() + ")"); 306 if (!success) throw new TimeoutException("The SyncRequest timed out (SID=" + SID.ToString() + ")");
309 }
310 } finally { 307 } finally {
311 lock (_invocations) _invocations.Remove(SID); 308 lock (_invocations) _invocations.Remove(SID);
312 } 309 }
313 return SyncEvent.Reply; 310 return SyncEvent.Reply;
314 } 311 }
315 312
316 public IPmlChannel CreateChannel(PmlElement data) { 313 public IPmlChannel CreateChannel(PmlElement data) {
317 UInt32 sid = GetNextSessionId(true); 314 UInt32 sid = GetNextSessionId(ref pNextSession, _sessions);
318 PmlSubChannel ch = new PmlSubChannel(this, sid, true); 315 PmlSubChannel ch = new PmlSubChannel(this, sid, true);
319 WriteSessionMessage(sid, 0, data); 316 WriteSessionMessage(sid, 0, data);
320 if (!ch.IsOpen) return null; 317 if (!ch.IsOpen) return null;
321 return ch; 318 return ch;
322 } 319 }
331 } 328 }
332 private void RemoveSession(ISession session) { 329 private void RemoveSession(ISession session) {
333 RemoveSession(session.ID); 330 RemoveSession(session.ID);
334 } 331 }
335 332
336 private UInt32 GetNextSessionId(bool IsSession) { 333 private static UInt32 GetNextSessionId<T>(ref UInt32 id, IDictionary<UInt32, T> dictionary) {
337 if (IsSession) { 334 lock (dictionary) {
338 lock (_sessions) { 335 do {
339 do { 336 id++;
340 unchecked { pNextSession++; } 337 } while (dictionary.ContainsKey(id));
341 } while (_sessions.ContainsKey(pNextSession)); 338 return id;
342 return pNextSession;
343 }
344 } else {
345 lock (_invocations) {
346 do {
347 unchecked { pNextSyncRequest++; }
348 } while (_invocations.ContainsKey(pNextSyncRequest));
349 return pNextSyncRequest;
350 }
351 } 339 }
352 } 340 }
353 341
354 protected void WriteSyncMessage(UInt32 SID, bool RPL, PmlElement MSG) { 342 protected void WriteSyncMessage(UInt32 SID, bool RPL, PmlElement MSG) {
355 PmlDictionary Msg2 = new PmlDictionary(); 343 PmlDictionary Msg2 = new PmlDictionary() {
356 Msg2.Add("CMD", new PmlString(RPL ? "RPL" : "REQ")); 344 { "CMD", RPL ? "RPL" : "REQ" },
357 Msg2.Add("SID", new PmlInteger(SID)); 345 { "SID", SID },
358 Msg2.Add("MSG", MSG); 346 { "MSG", MSG },
347 };
359 _WriteMessage(Msg2); 348 _WriteMessage(Msg2);
360 } 349 }
361 protected void WriteSessionMessage(UInt32 SID, byte CMD, PmlElement MSG) { 350 protected void WriteSessionMessage(UInt32 SID, byte CMD, PmlElement MSG) {
362 PmlDictionary Msg2 = new PmlDictionary(); 351 PmlDictionary Msg2 = new PmlDictionary() {
363 Msg2.Add("CMD", new PmlString("SES")); 352 { "CMD", "SES" },
364 Msg2.Add("SID", new PmlInteger(SID)); 353 { "SID", SID },
365 Msg2.Add("SCMD", new PmlInteger(CMD)); 354 { "SCMD", CMD },
355 };
366 if (MSG != null) Msg2.Add("MSG", MSG); 356 if (MSG != null) Msg2.Add("MSG", MSG);
367 _WriteMessage(Msg2); 357 _WriteMessage(Msg2);
368 } 358 }
369
370
371 359
372 /* LegacyPmlCommunicator compatibility */ 360 /* LegacyPmlCommunicator compatibility */
373 public PmlElement SyncRequest(PmlElement Request) { 361 public PmlElement SyncRequest(PmlElement Request) {
374 return Invoke(Request); 362 return Invoke(Request);
375 } 363 }