watsonb8/aurora
1
1
Fork 0
Code Issues 1 Pull Requests Projects 1 Releases Wiki Activity

EventManager rewritten to support a push arch instead of an internal poll. Crash finally fixed!!! The dangers of threading.

Browse Source
This commit is contained in:
watsonb8 2019-07-15 12:14:38 -04:00
parent d78dce44f0
commit 00b39b1d84
6 changed files with 254 additions and 327 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
Aurora/Design/Views/Party/PartyViewModel.cs
View File

@ -31,7 +31,6 @@ namespace Aurora.Design.Views.Party
private RemotePartyService.RemotePartyServiceClient _remotePartyClient;
private RemotePlaybackService.RemotePlaybackServiceClient _remotePlaybackClient;
private RemoteEventService.RemoteEventServiceClient _remoteEventsClient;
private Task _eventsTask;
CancellationTokenSource _eventCancellationTokenSource;
@ -129,7 +128,6 @@ namespace Aurora.Design.Views.Party
_remoteEventsClient = new RemoteEventService.RemoteEventServiceClient(_channel);
//Assign but don't start task
_eventsTask = new Task(GetEvents);
_eventCancellationTokenSource = new CancellationTokenSource();
}
@ -159,6 +157,7 @@ namespace Aurora.Design.Views.Party
try
{
Console.WriteLine(string.Format("CLIENT {0} - SubscribeToEvents called from client with id", SettingsService.Instance.ClientId));
_remoteEventsClient.SubscribeToEvents(req);
}
catch (Exception ex)
@ -166,8 +165,7 @@ namespace Aurora.Design.Views.Party
Console.WriteLine("Error subscribing to events: " + ex.Message);
}
_eventsTask.Start();
GetEvents();
}
@ -200,22 +198,31 @@ namespace Aurora.Design.Views.Party
OnPropertyChanged("IsNotSelectingHost");
}
private void AddMember(PartyMember member)
{
Members.Add(member);
}
/// <summary>
/// Asynchronous function for processing events off of the event stream.
/// </summary>
/// <returns></returns>
private async void GetEvents()
{
Console.WriteLine(string.Format("CLIENT {0} - GetEvents called from client with id", SettingsService.Instance.ClientId));
using (AsyncServerStreamingCall<BaseEvent> eventStream = _remoteEventsClient
.GetEvents(new EventsRequest { ClientId = SettingsService.Instance.ClientId }))
{
while (!_eventCancellationTokenSource.Token.IsCancellationRequested &&
await eventStream.ResponseStream.MoveNext(_eventCancellationTokenSource.Token))
{
Console.WriteLine(string.Format("CLIENT {0} - Event received for client with id", SettingsService.Instance.ClientId));
try
{
//Convert derived event type
BaseEvent e = eventStream.ResponseStream.Current;
BaseEvent e = new BaseEvent(eventStream.ResponseStream.Current);
switch (e.DerivedEventCase)
{
case BaseEvent.DerivedEventOneofCase.None:
@ -225,7 +232,16 @@ namespace Aurora.Design.Views.Party
case BaseEvent.DerivedEventOneofCase.PartyMemberJoinedEvent:
{
PartyMemberJoinedEvent derivedEvent = e.PartyMemberJoinedEvent;
Members.Add(derivedEvent.Member);
PartyMember member = new PartyMember
{
UserName = derivedEvent.Member.UserName,
Id = derivedEvent.Member.Id,
IpAddress = derivedEvent.Member.IpAddress,
Port = derivedEvent.Member.Port
};
AddMember(member);
break;
}
case BaseEvent.DerivedEventOneofCase.PartyMemberLeftEvent:
@ -234,7 +250,7 @@ namespace Aurora.Design.Views.Party
var found = Members.Where(x => x.Id == derivedEvent.Member.Id);
foreach (PartyMember member in found)
{
Members.Remove(member);
_members.Remove(member);
}
break;
}
@ -245,6 +261,8 @@ namespace Aurora.Design.Views.Party
{
Console.WriteLine(string.Format("EXCEPTION --- " + ex.Message));
}
OnPropertyChanged("Members");
}
}
}

35
Aurora/RemoteImpl/RemoteEventImpl.cs
View File

@ -1,10 +1,11 @@
using System;
using System.Threading.Tasks;
using System.Threading;
using System.Collections.Generic;
using System.Linq;
using Aurora.Services;
using System.Linq;
using Aurora.Services.EventManager;
using Aurora.Proto.Events;
using Aurora.Proto.General;
namespace Aurora.RemoteImpl
{
@ -24,24 +25,19 @@ namespace Aurora.RemoteImpl
/// <returns></returns>
public async override Task GetEvents(EventsRequest request, Grpc.Core.IServerStreamWriter<BaseEvent> responseStream, Grpc.Core.ServerCallContext context)
{
try
string peerId = Combine(new string[] { context.Peer, request.ClientId });
Console.WriteLine(string.Format("SERVER - Events request received from peer: {0}", peerId));
AutoResetEvent are = new AutoResetEvent(false);
Action<BaseEvent> callback = (BaseEvent bEvent) =>
{
while (EventManager.Instance.GetSubscriptionCount(Combine(new string[] { context.Peer, request.ClientId })) > 0)
{
Console.WriteLine("Peer " + context.Peer);
//TODO this causes crashes when two or more members are connected
//TODO Change this to events based stream instead of a poll based...
List<BaseEvent> events = EventManager.Instance.GetSessionEvents(Combine(new string[] { context.Peer, request.ClientId }));
foreach (BaseEvent currentEvent in events)
{
await responseStream.WriteAsync(currentEvent);
}
}
}
catch (Exception ex)
{
Console.WriteLine("Exception caught:" + ex.Message);
}
Console.WriteLine(string.Format("SERVER - Event fired for peer: {0}", peerId));
responseStream.WriteAsync(bEvent);
};
EventManager.Instance.AddEventHandler(callback, Combine(new string[] { context.Peer, request.ClientId }));
are.WaitOne();
}
/// <summary>
@ -52,6 +48,7 @@ namespace Aurora.RemoteImpl
/// <returns></returns>
public override Task<SubscriptionResponse> SubscribeToEvents(SubscribeRequest request, Grpc.Core.ServerCallContext context)
{
Console.WriteLine(string.Format("SERVER - Subscription from client with id: {0}", request.ClientId));
EventManager.Instance.AddSubscriptionList(Combine(new string[] { context.Peer, request.ClientId }), request.EventTypes.ToList());
return Task.FromResult(new SubscriptionResponse { Successful = true });

11
Aurora/RemoteImpl/RemotePartyImpl.cs
View File

@ -2,12 +2,9 @@ using System;
using System.Threading.Tasks;
using System.Collections.ObjectModel;
using System.Linq;
using Google.Protobuf.WellKnownTypes;
using Aurora.Proto.Party;
using Aurora.Proto.General;
using Aurora.Proto.Events;
using Aurora.Models;
using Aurora.Services;
using Aurora.Services.EventManager;
namespace Aurora.RemoteImpl
{
@ -39,6 +36,8 @@ namespace Aurora.RemoteImpl
IpAddress = context.Host,
};
Console.WriteLine("SERVER - Client joined party: " + partyMember.Id);
_partyMembers.Add(partyMember);
BaseEvent e = new BaseEvent
@ -50,7 +49,7 @@ namespace Aurora.RemoteImpl
}
};
EventManager.Instance.PushEvent(e);
EventManager.Instance.FireEvent(e);
JoinPartyResponse response = new JoinPartyResponse() { Status = PartyJoinedStatusEnum.Connected, ClientId = partyMember.Id };
return Task.FromResult(response);
@ -71,7 +70,7 @@ namespace Aurora.RemoteImpl
}
};
EventManager.Instance.PushEvent(bv);
EventManager.Instance.FireEvent(bv);
LeavePartyResponse response = new LeavePartyResponse() { Status = PartyJoinedStatusEnum.Disconnected };
return Task.FromResult(response);

294
Aurora/Services/EventManager.cs
View File

@ -1,294 +0,0 @@
using System;
using System.IO;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using System.Linq;
using System.Threading;
using Google.Protobuf.WellKnownTypes;
using Google.Protobuf.Reflection;
using Aurora.Proto.Events;
using Aurora.Models;
namespace Aurora.Services
{
public class EventManager : BaseService<EventManager>
{
//TODO purge inactive sessions from the list
public EventManager()
{
_eventQueues = new ConcurrentDictionary<string, BlockingCollection<BaseEvent>>();
_subscriptionList = new Dictionary<string, List<EventType>>();
_unprocessedEventsList = new ObservableCollection<BaseEvent>();
_unprocessedEventsList.CollectionChanged += UnprocessedEventsList_Changed;
}
#region Fields
private ConcurrentDictionary<string, BlockingCollection<BaseEvent>> _eventQueues;
private Dictionary<string, List<EventType>> _subscriptionList;
private ObservableCollection<BaseEvent> _unprocessedEventsList;
#endregion Fields
#region Private Methods
/// <summary>
/// Add event to appropriate events list for subscribed sessions
/// </summary>
/// <param name="unprocessedEvent"></param>
private void ProcessEvent(BaseEvent unprocessedEvent)
{
lock (_subscriptionList)
{
lock (_unprocessedEventsList)
{
//Duplicate events into client event queues
foreach (KeyValuePair<string, List<EventType>> subscription in _subscriptionList)
{
//Active Events contains the session in question
if (_eventQueues.ContainsKey(subscription.Key))
{
//Add all of the events to active events that are subscribed
if (subscription.Value.Contains(unprocessedEvent.EventType))
{
BlockingCollection<BaseEvent> eventList;
_eventQueues.TryGetValue(subscription.Key, out eventList);
if (eventList != null)
{
eventList.Add(unprocessedEvent);
}
}
}
}
//Remove Event from list
_unprocessedEventsList.Remove(unprocessedEvent);
}
}
}
#endregion Private Methods
#region Public Methods
/// <summary>
/// Get the list of event type subscriptions for a given session id.
/// </summary>
/// <param name="session">Session Id</param>
/// <returns></returns>
public List<EventType> GetSubscriptionList(string session)
{
List<EventType> eventList = new List<EventType>();
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.TryGetValue(session, out eventList);
}
return eventList;
}
/// <summary>
/// Get the number of event subscriptions for a given session
/// </summary>
/// <param name="session">Session Id</param>
/// <returns></returns>
public int GetSubscriptionCount(string session)
{
List<EventType> eventList = new List<EventType>();
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.TryGetValue(session, out eventList);
}
return eventList.Count();
}
/// <summary>
/// Add a new subscription
/// </summary>
/// <param name="session"></param>
/// <param name="type"></param>
public bool AddSubscription(string session, EventType type)
{
bool success = false;
if (!_subscriptionList.ContainsKey(session))
{
//Add session to subscription list
List<EventType> eventList = new List<EventType>();
eventList.Add(type);
_subscriptionList.Add(session, eventList);
success = true;
//base.LogInformation(string.Format("Subscription removed for event type {0} subscription on session {1}", type.ToString(), session));
}
else
{
List<EventType> eventList;
_subscriptionList.TryGetValue(session, out eventList);
if (eventList != null)
{
eventList.Add(type);
success = true;
//base.LogInformation(string.Format("Subscription removed for event type {0} subscription on session {1}", type.ToString(), session));
}
}
//Add activeEvents if it doesn't exist
if (!_eventQueues.ContainsKey(session))
{
//Add session to active events
_eventQueues.TryAdd(session, new BlockingCollection<BaseEvent>());
}
return success;
}
/// <summary>
/// Add a list of subscriptions. This unsubscribes from unused events.
/// </summary>
/// <param name="session">The browser session id.</param>
/// <param name="types">The list of event types to subscribe to.</param>
public void AddSubscriptionList(string session, List<EventType> types)
{
RemoveAllSubscriptions(session);
foreach (EventType e in types)
{
AddSubscription(session, e);
}
}
/// <summary>
/// Unsubscribe from a given event type.
/// </summary>
/// <param name="session">Session Id</param>
/// <param name="type">Event Type to be removed</param>
public void RemoveSubscription(string session, EventType type)
{
if (_subscriptionList.ContainsKey(session))
{
List<EventType> eventTypeList;
_subscriptionList.TryGetValue(session, out eventTypeList);
if (eventTypeList != null && eventTypeList.Contains(type))
{
eventTypeList.Remove(type);
//base.LogInformation(string.Format("Subscription removed for event type {0} subscription on session {1}", type.ToString(), session));
}
}
}
public void RemoveSubscriptionList(string session, List<EventType> types)
{
foreach (EventType e in types)
{
RemoveSubscription(session, e);
}
}
/// <summary>
/// Remove all subscriptons for a given session.
/// </summary>
/// <param name="session">Session Id</param>
public void RemoveAllSubscriptions(string session)
{
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.Remove(session);
}
if (_eventQueues.ContainsKey(session))
{
BlockingCollection<BaseEvent> rem = null;
_eventQueues.TryRemove(session, out rem);
}
//base.LogInformation(string.Format("All subscriptions removed for event type {0}", session));
}
/// <summary>
/// Get a list of accumulated events for a given session. Timeout after 5 seconds if list is empty.
/// </summary>
/// <param name="session"></param>
/// <returns>List<BaseEvent></returns>
public List<BaseEvent> GetSessionEvents(string session)
{
BlockingCollection<BaseEvent> eList;
_eventQueues.TryGetValue(session, out eList);
List<BaseEvent> returnList = new List<BaseEvent>();
if (eList == null)
{
return returnList;
}
//Continue to take until the eList is not empty.
while (true)
{
BaseEvent e;
eList.TryTake(out e);
if (e != null)
{
returnList.Add(e);
}
else
{
break;
}
}
//In the event that eList was empty to begin with, wait for something to be appear or cancel
if (eList.Count == 0)
{
CancellationTokenSource tkSrc = new CancellationTokenSource(5000);
BaseEvent e = null;
try
{
e = eList.Take(tkSrc.Token);
}
catch (OperationCanceledException)
{
// eat this
}
catch (Exception)
{
}
if (e != null)
{
returnList.Add(e);
}
}
return returnList;
}
/// <summary>
/// Push a new event to the event queue.
/// </summary>
/// <param name="newEvent">The event to be pushed</param>
public void PushEvent(BaseEvent newEvent)
{
_unprocessedEventsList.Add(newEvent);
}
#endregion Public Methods
#region Events
private void UnprocessedEventsList_Changed(object sender, NotifyCollectionChangedEventArgs e)
{
switch (e.Action)
{
case NotifyCollectionChangedAction.Add:
{
foreach (BaseEvent bEvent in e.NewItems)
{
ProcessEvent(bEvent);
}
break;
}
}
}
#endregion Events
}
}

208
Aurora/Services/EventManager/EventManager.cs Normal file
View File

@ -0,0 +1,208 @@
using System;
using System.IO;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Collections.Specialized;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf.WellKnownTypes;
using Google.Protobuf.Reflection;
using Aurora.Proto.Events;
using Aurora.Models;
namespace Aurora.Services.EventManager
{
public class EventManager : BaseService<EventManager>
{
#region Fields
private Dictionary<string, List<EventType>> _subscriptionList;
private Dictionary<string, Action<BaseEvent>> _actionList;
#endregion Fields
public EventManager()
{
_subscriptionList = new Dictionary<string, List<EventType>>();
_actionList = new Dictionary<string, Action<BaseEvent>>();
}
#region Private Methods
#endregion Private Methods
#region Public Methods
/// <summary>
/// Get the list of event type subscriptions for a given session id.
/// </summary>
/// <param name="session">Session Id</param>
/// <returns></returns>
public List<EventType> GetSubscriptionList(string session)
{
List<EventType> eventList = new List<EventType>();
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.TryGetValue(session, out eventList);
}
return eventList;
}
/// <summary>
/// Get the number of event subscriptions for a given session
/// </summary>
/// <param name="session">Session Id</param>
/// <returns></returns>
public int GetSubscriptionCount(string session)
{
List<EventType> eventList = new List<EventType>();
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.TryGetValue(session, out eventList);
}
return eventList.Count();
}
/// <summary>
/// Add a new subscription
/// </summary>
/// <param name="session"></param>
/// <param name="type"></param>
public bool AddSubscription(string session, EventType type)
{
bool success = false;
lock (_subscriptionList)
{
if (!_subscriptionList.ContainsKey(session))
{
//Add session to subscription list
List<EventType> eventList = new List<EventType>();
eventList.Add(type);
_subscriptionList.Add(session, eventList);
success = true;
}
else
{
_subscriptionList.TryGetValue(session, out List<EventType> eventList);
if (eventList != null)
{
eventList.Add(type);
success = true;
}
}
}
return success;
}
/// <summary>
/// Add a list of subscriptions. This unsubscribes from unused events.
/// </summary>
/// <param name="session">The browser session id.</param>
/// <param name="types">The list of event types to subscribe to.</param>
public void AddSubscriptionList(string session, List<EventType> types)
{
RemoveAllSubscriptions(session);
foreach (EventType e in types)
{
AddSubscription(session, e);
}
}
/// <summary>
/// Unsubscribe from a given event type.
/// </summary>
/// <param name="session">Session Id</param>
/// <param name="type">Event Type to be removed</param>
public void RemoveSubscription(string session, EventType type)
{
lock (_subscriptionList)
{
if (_subscriptionList.ContainsKey(session))
{
List<EventType> eventTypeList;
_subscriptionList.TryGetValue(session, out eventTypeList);
if (eventTypeList != null && eventTypeList.Contains(type))
{
eventTypeList.Remove(type);
//base.LogInformation(string.Format("Subscription removed for event type {0} subscription on session {1}", type.ToString(), session));
}
}
}
}
public void RemoveSubscriptionList(string session, List<EventType> types)
{
foreach (EventType e in types)
{
RemoveSubscription(session, e);
}
}
/// <summary>
/// Remove all subscriptons for a given session.
/// </summary>
/// <param name="session">Session Id</param>
public void RemoveAllSubscriptions(string session)
{
if (_subscriptionList.ContainsKey(session))
{
_subscriptionList.Remove(session);
}
}
public void AddEventHandler(Action<BaseEvent> action, string sessionId)
{
lock (_actionList)
{
_actionList.Add(sessionId, action);
}
}
public void RemoveEventHandler(string sessionId)
{
_actionList.Remove(sessionId);
}
public void FireEvent(BaseEvent bEvent)
{
Dictionary<string, Action<BaseEvent>> actionsCopy = new Dictionary<string, Action<BaseEvent>>();
//Copy actions list
lock (_actionList)
{
foreach (KeyValuePair<string, Action<BaseEvent>> pair in _actionList)
{
actionsCopy.Add(pair.Key, pair.Value);
}
}
lock (_subscriptionList)
{
foreach (KeyValuePair<string, List<EventType>> pair in _subscriptionList)
{
Console.WriteLine("SERVER - Invoking action for client: " + pair.Key);
Task.Delay(1000);
//If action list contains an action for id, invoke
if (actionsCopy.ContainsKey(pair.Key))
{
actionsCopy.TryGetValue(pair.Key, out Action<BaseEvent> action);
ParameterizedThreadStart operation = new ParameterizedThreadStart(obj => action((BaseEvent)obj));
Thread executionThread = new Thread(operation, 1024 * 1024);
executionThread.Start(bEvent);
executionThread.Join();
}
}
}
}
#endregion Public Methods
}
}

1
Aurora/Services/ServerService.cs
View File

@ -86,7 +86,6 @@ namespace Aurora.Services
RegisterService(RemotePlaybackService.BindService(_remotePlaybackImpl));
RegisterService(RemoteEventService.BindService(_remoteEventImpl));
}
_server.Start();
}
catch (Exception ex)