using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;
using log4net;
using ACE.Database.Entity;
using ACE.Database.Models.Shard;
using ACE.Entity.Enum;
namespace ACE.Database
{
public class SerializedShardDatabase
{
private static readonly ILog log = LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);
///
/// This is the base database that SerializedShardDatabase is a wrapper for.
///
public readonly ShardDatabase BaseDatabase;
private readonly BlockingCollection _queue = new BlockingCollection();
private Thread _workerThread;
internal SerializedShardDatabase(ShardDatabase shardDatabase)
{
BaseDatabase = shardDatabase;
}
public void Start()
{
_workerThread = new Thread(DoWork);
_workerThread.Name = "Serialized Shard Database";
_workerThread.Start();
}
public void Stop()
{
_queue.CompleteAdding();
_workerThread.Join();
}
private void DoWork()
{
while (!_queue.IsCompleted)
{
try
{
Task t = _queue.Take();
try
{
t.Start();
t.Wait();
}
catch (Exception ex)
{
log.Error($"[DATABASE] DoWork task failed with exception: {ex}");
// perhaps add failure callbacks?
// swallow for now. can't block other db work because 1 fails.
}
}
catch (ObjectDisposedException)
{
// the _queue has been disposed, we're good
break;
}
catch (InvalidOperationException)
{
// _queue is empty and CompleteForAdding has been called -- we're done here
break;
}
}
}
public int QueueCount => _queue.Count;
public void GetCurrentQueueWaitTime(Action callback)
{
var initialCallTime = DateTime.UtcNow;
_queue.Add(new Task(() =>
{
callback?.Invoke(DateTime.UtcNow - initialCallTime);
}));
}
///
/// Will return uint.MaxValue if no records were found within the range provided.
///
public void GetMaxGuidFoundInRange(uint min, uint max, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.GetMaxGuidFoundInRange(min, max);
callback?.Invoke(result);
}));
}
///
/// This will return available id's, in the form of sequence gaps starting from min.
/// If a gap is just 1 value wide, then both start and end will be the same number.
///
public void GetSequenceGaps(uint min, uint limitAvailableIDsReturned, Action> callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.GetSequenceGaps(min, limitAvailableIDsReturned);
callback?.Invoke(result);
}));
}
public void SaveBiota(ACE.Entity.Models.Biota biota, ReaderWriterLockSlim rwLock, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.SaveBiota(biota, rwLock);
callback?.Invoke(result);
}));
}
public void SaveBiotasInParallel(IEnumerable<(ACE.Entity.Models.Biota biota, ReaderWriterLockSlim rwLock)> biotas, Action callback, bool doNotAddToCache = false)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.SaveBiotasInParallel(biotas, doNotAddToCache);
callback?.Invoke(result);
}));
}
public void RemoveBiota(uint id, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.RemoveBiota(id);
callback?.Invoke(result);
}));
}
public void RemoveBiota(uint id, Action callback, Action performanceResults)
{
var initialCallTime = DateTime.UtcNow;
_queue.Add(new Task(() =>
{
var taskStartTime = DateTime.UtcNow;
var result = BaseDatabase.RemoveBiota(id);
var taskCompletedTime = DateTime.UtcNow;
callback?.Invoke(result);
performanceResults?.Invoke(taskStartTime - initialCallTime, taskCompletedTime - taskStartTime);
}));
}
public void RemoveBiotasInParallel(IEnumerable ids, Action callback, Action performanceResults)
{
var initialCallTime = DateTime.UtcNow;
_queue.Add(new Task(() =>
{
var taskStartTime = DateTime.UtcNow;
var result = BaseDatabase.RemoveBiotasInParallel(ids);
var taskCompletedTime = DateTime.UtcNow;
callback?.Invoke(result);
performanceResults?.Invoke(taskStartTime - initialCallTime, taskCompletedTime - taskStartTime);
}));
}
public void GetPossessedBiotasInParallel(uint id, Action callback)
{
_queue.Add(new Task(() =>
{
var c = BaseDatabase.GetPossessedBiotasInParallel(id);
callback?.Invoke(c);
}));
}
public void GetInventoryInParallel(uint parentId, bool includedNestedItems, Action> callback)
{
_queue.Add(new Task(() =>
{
var c = BaseDatabase.GetInventoryInParallel(parentId, includedNestedItems);
callback?.Invoke(c);
}));
}
public void IsCharacterNameAvailable(string name, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.IsCharacterNameAvailable(name);
callback?.Invoke(result);
}));
}
public void GetCharacters(uint accountId, bool includeDeleted, Action> callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.GetCharacters(accountId, includeDeleted);
callback?.Invoke(result);
}));
}
public void GetCharacter(uint characterId, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.GetCharacter(characterId);
callback?.Invoke(result);
}));
}
public void SaveCharacter(Character character, ReaderWriterLockSlim rwLock, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.SaveCharacter(character, rwLock);
callback?.Invoke(result);
}));
}
public void RenameCharacter(Character character, string newName, ReaderWriterLockSlim rwLock, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.RenameCharacter(character, newName, rwLock);
callback?.Invoke(result);
}));
}
public void SetCharacterAccessLevelByName(string name, AccessLevel accessLevel, Action callback)
{
// TODO
throw new NotImplementedException();
}
public void AddCharacterInParallel(ACE.Entity.Models.Biota biota, ReaderWriterLockSlim biotaLock, IEnumerable<(ACE.Entity.Models.Biota biota, ReaderWriterLockSlim rwLock)> possessions, Character character, ReaderWriterLockSlim characterLock, Action callback)
{
_queue.Add(new Task(() =>
{
var result = BaseDatabase.AddCharacterInParallel(biota, biotaLock, possessions, character, characterLock);
callback?.Invoke(result);
}));
}
}
}