using MediaBrowser.Model.Logging;
using System;
using System.Collections.Concurrent;
using System.Data;
using System.Data.Common;
using System.Data.SQLite;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
namespace MediaBrowser.Server.Implementations.Sqlite
{
///
/// Class SqliteRepository
///
public abstract class SqliteRepository : IDisposable
{
///
/// The db file name
///
protected string dbFileName;
///
/// The connection
///
protected SQLiteConnection connection;
///
/// The delayed commands
///
protected ConcurrentQueue delayedCommands = new ConcurrentQueue();
///
/// The flush interval
///
private const int FlushInterval = 2000;
///
/// The flush timer
///
private Timer FlushTimer;
///
/// Gets the logger.
///
/// The logger.
protected ILogger Logger { get; private set; }
///
/// Gets a value indicating whether [enable delayed commands].
///
/// true if [enable delayed commands]; otherwise, false.
protected virtual bool EnableDelayedCommands
{
get
{
return true;
}
}
///
/// Initializes a new instance of the class.
///
/// The log manager.
/// logger
protected SqliteRepository(ILogManager logManager)
{
if (logManager == null)
{
throw new ArgumentNullException("logManager");
}
Logger = logManager.GetLogger(GetType().Name);
}
///
/// Connects to DB.
///
/// The db path.
/// Task{System.Boolean}.
/// dbPath
protected async Task ConnectToDb(string dbPath)
{
if (string.IsNullOrEmpty(dbPath))
{
throw new ArgumentNullException("dbPath");
}
dbFileName = dbPath;
var connectionstr = new SQLiteConnectionStringBuilder
{
PageSize = 4096,
CacheSize = 40960,
SyncMode = SynchronizationModes.Off,
DataSource = dbPath,
JournalMode = SQLiteJournalModeEnum.Memory
};
connection = new SQLiteConnection(connectionstr.ConnectionString);
await connection.OpenAsync().ConfigureAwait(false);
if (EnableDelayedCommands)
{
// Run once
FlushTimer = new Timer(Flush, null, TimeSpan.FromMilliseconds(FlushInterval), TimeSpan.FromMilliseconds(-1));
}
}
///
/// Runs the queries.
///
/// The queries.
/// true if XXXX, false otherwise
/// queries
protected void RunQueries(string[] queries)
{
if (queries == null)
{
throw new ArgumentNullException("queries");
}
using (var tran = connection.BeginTransaction())
{
try
{
using (var cmd = connection.CreateCommand())
{
foreach (var query in queries)
{
cmd.Transaction = tran;
cmd.CommandText = query;
cmd.ExecuteNonQuery();
}
}
tran.Commit();
}
catch (Exception e)
{
Logger.ErrorException("Error running queries", e);
tran.Rollback();
throw;
}
}
}
///
/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
private readonly object _disposeLock = new object();
///
/// Releases unmanaged and - optionally - managed resources.
///
/// true to release both managed and unmanaged resources; false to release only unmanaged resources.
protected virtual void Dispose(bool dispose)
{
if (dispose)
{
try
{
lock (_disposeLock)
{
if (connection != null)
{
if (EnableDelayedCommands)
{
FlushOnDispose();
}
if (connection.IsOpen())
{
connection.Close();
}
connection.Dispose();
connection = null;
}
if (FlushTimer != null)
{
FlushTimer.Dispose();
FlushTimer = null;
}
}
}
catch (Exception ex)
{
Logger.ErrorException("Error disposing database", ex);
}
}
}
///
/// Flushes the on dispose.
///
private void FlushOnDispose()
{
// If we're not already flushing, do it now
if (!_isFlushing)
{
Flush(null);
}
// Don't dispose in the middle of a flush
while (_isFlushing)
{
Thread.Sleep(25);
}
}
///
/// Queues the command.
///
/// The CMD.
/// cmd
protected void QueueCommand(SQLiteCommand cmd)
{
if (cmd == null)
{
throw new ArgumentNullException("cmd");
}
delayedCommands.Enqueue(cmd);
}
///
/// The is flushing
///
private bool _isFlushing;
///
/// Flushes the specified sender.
///
/// The sender.
private void Flush(object sender)
{
// Cannot call Count on a ConcurrentQueue since it's an O(n) operation
// Use IsEmpty instead
if (delayedCommands.IsEmpty)
{
FlushTimer.Change(TimeSpan.FromMilliseconds(FlushInterval), TimeSpan.FromMilliseconds(-1));
return;
}
if (_isFlushing)
{
return;
}
_isFlushing = true;
var numCommands = 0;
using (var tran = connection.BeginTransaction())
{
try
{
while (!delayedCommands.IsEmpty)
{
SQLiteCommand command;
delayedCommands.TryDequeue(out command);
command.Connection = connection;
command.Transaction = tran;
command.ExecuteNonQuery();
command.Dispose();
numCommands++;
}
tran.Commit();
}
catch (Exception e)
{
Logger.ErrorException("Failed to commit transaction.", e);
tran.Rollback();
}
}
Logger.Debug("SQL Delayed writer executed " + numCommands + " commands");
FlushTimer.Change(TimeSpan.FromMilliseconds(FlushInterval), TimeSpan.FromMilliseconds(-1));
_isFlushing = false;
}
///
/// Executes the command.
///
/// The CMD.
/// Task.
/// cmd
public async Task ExecuteCommand(DbCommand cmd)
{
if (cmd == null)
{
throw new ArgumentNullException("cmd");
}
using (var tran = connection.BeginTransaction())
{
try
{
cmd.Connection = connection;
cmd.Transaction = tran;
await cmd.ExecuteNonQueryAsync().ConfigureAwait(false);
tran.Commit();
}
catch (Exception e)
{
Logger.ErrorException("Failed to commit transaction.", e);
tran.Rollback();
}
}
}
///
/// Gets a stream from a DataReader at a given ordinal
///
/// The reader.
/// The ordinal.
/// Stream.
/// reader
protected static Stream GetStream(IDataReader reader, int ordinal)
{
if (reader == null)
{
throw new ArgumentNullException("reader");
}
var memoryStream = new MemoryStream();
var num = 0L;
var array = new byte[4096];
long bytes;
do
{
bytes = reader.GetBytes(ordinal, num, array, 0, array.Length);
memoryStream.Write(array, 0, (int)bytes);
num += bytes;
}
while (bytes > 0L);
memoryStream.Position = 0;
return memoryStream;
}
}
}