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 { 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); } /// /// 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 { if (connection != null) { if (EnableDelayedCommands) { FlushOnDispose(); } if (connection.IsOpen()) { connection.Close(); } connection.Dispose(); } 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(); 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; } } }