@ -30,115 +30,126 @@ using System.Threading.Tasks;
namespace SQLite namespace SQLite
{ {
public partial class SQLiteAsyncConnection public partial class SQLiteAsyncConnection
{ {
SQLiteConnectionString _ connectionString ; SQLiteConnectionString _ connectionString ;
public SQLiteAsyncConnection ( string databasePath , bool storeDateTimeAsTicks = false ) public SQLiteAsyncConnection ( string databasePath , bool storeDateTimeAsTicks = false )
{ {
_ connectionString = new SQLiteConnectionString ( databasePath , storeDateTimeAsTicks ) ; _ connectionString = new SQLiteConnectionString ( databasePath , storeDateTimeAsTicks ) ;
} }
SQLiteConnectionWithLock GetConnection ( ) SQLiteConnectionWithLock GetConnection ( )
{ {
return SQLiteConnectionPool . Shared . GetConnection ( _ connectionString ) ; return SQLiteConnectionPool . Shared . GetConnection ( _ connectionString ) ;
} }
public Task < CreateTablesResult > CreateTableAsync < T > ( ) public Task < CreateTablesResult > CreateTableAsync < T > ( )
where T : new ( ) where T : new ( )
{ {
return CreateTablesAsync ( typeof ( T ) ) ; return CreateTablesAsync ( typeof ( T ) ) ;
} }
public Task < CreateTablesResult > CreateTablesAsync < T , T2 > ( ) public Task < CreateTablesResult > CreateTablesAsync < T , T2 > ( )
where T : new ( ) where T : new ( )
where T2 : new ( ) where T2 : new ( )
{ {
return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) ) ; return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) ) ;
} }
public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 > ( ) public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 > ( )
where T : new ( ) where T : new ( )
where T2 : new ( ) where T2 : new ( )
where T3 : new ( ) where T3 : new ( )
{ {
return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) ) ; return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) ) ;
} }
public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 , T4 > ( ) public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 , T4 > ( )
where T : new ( ) where T : new ( )
where T2 : new ( ) where T2 : new ( )
where T3 : new ( ) where T3 : new ( )
where T4 : new ( ) where T4 : new ( )
{ {
return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) , typeof ( T4 ) ) ; return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) , typeof ( T4 ) ) ;
} }
public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 , T4 , T5 > ( ) public Task < CreateTablesResult > CreateTablesAsync < T , T2 , T3 , T4 , T5 > ( )
where T : new ( ) where T : new ( )
where T2 : new ( ) where T2 : new ( )
where T3 : new ( ) where T3 : new ( )
where T4 : new ( ) where T4 : new ( )
where T5 : new ( ) where T5 : new ( )
{ {
return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) , typeof ( T4 ) , typeof ( T5 ) ) ; return CreateTablesAsync ( typeof ( T ) , typeof ( T2 ) , typeof ( T3 ) , typeof ( T4 ) , typeof ( T5 ) ) ;
} }
public Task < CreateTablesResult > CreateTablesAsync ( params Type [ ] types ) public Task < CreateTablesResult > CreateTablesAsync ( params Type [ ] types )
{ {
return Task . Factory . StartNew ( ( ) = > { return Task . Factory . StartNew ( ( ) = >
CreateTablesResult result = new CreateTablesResult ( ) ; {
var conn = GetConnection ( ) ; CreateTablesResult result = new CreateTablesResult ( ) ;
using ( conn . Lock ( ) ) { var conn = GetConnection ( ) ;
foreach ( Type type in types ) { using ( conn . Lock ( ) )
int aResult = conn . CreateTable ( type ) ; {
result . Results [ type ] = aResult ; foreach ( Type type in types )
} {
} int aResult = conn . CreateTable ( type ) ;
return result ; result . Results [ type ] = aResult ;
} ) ; }
} }
return result ;
public Task < int > DropTableAsync < T > ( ) } ) ;
where T : new ( ) }
{
return Task . Factory . StartNew ( ( ) = > { public Task < int > DropTableAsync < T > ( )
var conn = GetConnection ( ) ; where T : new ( )
using ( conn . Lock ( ) ) { {
return conn . DropTable < T > ( ) ; return Task . Factory . StartNew ( ( ) = >
} {
} ) ; var conn = GetConnection ( ) ;
} using ( conn . Lock ( ) )
{
public Task < int > InsertAsync ( object item ) return conn . DropTable < T > ( ) ;
{ }
return Task . Factory . StartNew ( ( ) = > { } ) ;
var conn = GetConnection ( ) ; }
using ( conn . Lock ( ) ) {
return conn . Insert ( item ) ; public Task < int > InsertAsync ( object item )
} {
} ) ; return Task . Factory . StartNew ( ( ) = >
} {
var conn = GetConnection ( ) ;
public Task < int > UpdateAsync ( object item ) using ( conn . Lock ( ) )
{ {
return Task . Factory . StartNew ( ( ) = > { return conn . Insert ( item ) ;
var conn = GetConnection ( ) ; }
using ( conn . Lock ( ) ) { } ) ;
return conn . Update ( item ) ; }
}
} ) ; public Task < int > UpdateAsync ( object item )
} {
return Task . Factory . StartNew ( ( ) = >
public Task < int > DeleteAsync ( object item ) {
{ var conn = GetConnection ( ) ;
return Task . Factory . StartNew ( ( ) = > { using ( conn . Lock ( ) )
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { return conn . Update ( item ) ;
return conn . Delete ( item ) ; }
} } ) ;
} ) ; }
}
public Task < int > DeleteAsync ( object item )
{
return Task . Factory . StartNew ( ( ) = >
{
var conn = GetConnection ( ) ;
using ( conn . Lock ( ) )
{
return conn . Delete ( item ) ;
}
} ) ;
}
public Task < T > GetAsync < T > ( object pk ) public Task < T > GetAsync < T > ( object pk )
where T : new ( ) where T : new ( )
@ -153,18 +164,20 @@ namespace SQLite
} ) ; } ) ;
} }
public Task < T > FindAsync < T > ( object pk ) public Task < T > FindAsync < T > ( object pk )
where T : new ( ) where T : new ( )
{ {
return Task . Factory . StartNew ( ( ) = > { return Task . Factory . StartNew ( ( ) = >
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { var conn = GetConnection ( ) ;
return conn . Find < T > ( pk ) ; using ( conn . Lock ( ) )
} {
} ) ; return conn . Find < T > ( pk ) ;
} }
} ) ;
}
public Task < T > GetAsync < T > ( Expression < Func < T , bool > > predicate ) public Task < T > GetAsync < T > ( Expression < Func < T , bool > > predicate )
where T : new ( ) where T : new ( )
{ {
return Task . Factory . StartNew ( ( ) = > return Task . Factory . StartNew ( ( ) = >
@ -172,60 +185,70 @@ namespace SQLite
var conn = GetConnection ( ) ; var conn = GetConnection ( ) ;
using ( conn . Lock ( ) ) using ( conn . Lock ( ) )
{ {
return conn . Get < T > ( predicate ) ; return conn . Get < T > ( predicate ) ;
} }
} ) ; } ) ;
} }
public Task < T > FindAsync < T > ( Expression < Func < T , bool > > predicate ) public Task < T > FindAsync < T > ( Expression < Func < T , bool > > predicate )
where T : new ( ) where T : new ( )
{ {
return Task . Factory . StartNew ( ( ) = > { return Task . Factory . StartNew ( ( ) = >
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { var conn = GetConnection ( ) ;
return conn . Find < T > ( predicate ) ; using ( conn . Lock ( ) )
} {
} ) ; return conn . Find < T > ( predicate ) ;
} }
} ) ;
public Task < int > ExecuteAsync ( string query , params object [ ] args ) }
{
return Task < int > . Factory . StartNew ( ( ) = > { public Task < int > ExecuteAsync ( string query , params object [ ] args )
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { return Task < int > . Factory . StartNew ( ( ) = >
return conn . Execute ( query , args ) ; {
} var conn = GetConnection ( ) ;
} ) ; using ( conn . Lock ( ) )
} {
return conn . Execute ( query , args ) ;
public Task < int > InsertAllAsync ( IEnumerable items ) }
{ } ) ;
return Task . Factory . StartNew ( ( ) = > { }
var conn = GetConnection ( ) ;
using ( conn . Lock ( ) ) { public Task < int > InsertAllAsync ( IEnumerable items )
return conn . InsertAll ( items ) ; {
} return Task . Factory . StartNew ( ( ) = >
} ) ; {
} var conn = GetConnection ( ) ;
using ( conn . Lock ( ) )
{
return conn . InsertAll ( items ) ;
}
} ) ;
}
[Obsolete("Will cause a deadlock if any call in action ends up in a different thread. Use RunInTransactionAsync(Action<SQLiteConnection>) instead.")] [Obsolete("Will cause a deadlock if any call in action ends up in a different thread. Use RunInTransactionAsync(Action<SQLiteConnection>) instead.")]
public Task RunInTransactionAsync ( Action < SQLiteAsyncConnection > action ) public Task RunInTransactionAsync ( Action < SQLiteAsyncConnection > action )
{ {
return Task . Factory . StartNew ( ( ) = > { return Task . Factory . StartNew ( ( ) = >
var conn = this . GetConnection ( ) ; {
using ( conn . Lock ( ) ) { var conn = this . GetConnection ( ) ;
conn . BeginTransaction ( ) ; using ( conn . Lock ( ) )
try { {
action ( this ) ; conn . BeginTransaction ( ) ;
conn . Commit ( ) ; try
} {
catch ( Exception ) { action ( this ) ;
conn . Rollback ( ) ; conn . Commit ( ) ;
throw ; }
} catch ( Exception )
} {
} ) ; conn . Rollback ( ) ;
} throw ;
}
}
} ) ;
}
public Task RunInTransactionAsync ( Action < SQLiteConnection > action ) public Task RunInTransactionAsync ( Action < SQLiteConnection > action )
{ {
@ -249,238 +272,256 @@ namespace SQLite
} ) ; } ) ;
} }
public AsyncTableQuery < T > Table < T > ( ) public AsyncTableQuery < T > Table < T > ( )
where T : new ( ) where T : new ( )
{ {
//
//
// This isn't async as the underlying connection doesn't go out to the database
// This isn't async as the underlying connection doesn't go out to the database
// until the query is performed. The Async methods are on the query iteself.
// until the query is performed. The Async methods are on the query iteself.
//
//
var conn = GetConnection ( ) ; var conn = GetConnection ( ) ;
return new AsyncTableQuery < T > ( conn . Table < T > ( ) ) ; return new AsyncTableQuery < T > ( conn . Table < T > ( ) ) ;
} }
public Task < T > ExecuteScalarAsync < T > ( string sql , params object [ ] args ) public Task < T > ExecuteScalarAsync < T > ( string sql , params object [ ] args )
{ {
return Task < T > . Factory . StartNew ( ( ) = > { return Task < T > . Factory . StartNew ( ( ) = >
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { var conn = GetConnection ( ) ;
var command = conn . CreateCommand ( sql , args ) ; using ( conn . Lock ( ) )
return command . ExecuteScalar < T > ( ) ; {
} var command = conn . CreateCommand ( sql , args ) ;
} ) ; return command . ExecuteScalar < T > ( ) ;
} }
} ) ;
public Task < List < T > > QueryAsync < T > ( string sql , params object [ ] args ) }
where T : new ( )
{ public Task < List < T > > QueryAsync < T > ( string sql , params object [ ] args )
return Task < List < T > > . Factory . StartNew ( ( ) = > { where T : new ( )
var conn = GetConnection ( ) ; {
using ( conn . Lock ( ) ) { return Task < List < T > > . Factory . StartNew ( ( ) = >
return conn . Query < T > ( sql , args ) ; {
} var conn = GetConnection ( ) ;
} ) ; using ( conn . Lock ( ) )
} {
} return conn . Query < T > ( sql , args ) ;
}
//
} ) ;
// TODO: Bind to AsyncConnection.GetConnection instead so that delayed
}
// execution can still work after a Pool.Reset.
//
public class AsyncTableQuery < T >
where T : new ( )
{
TableQuery < T > _ innerQuery ;
public AsyncTableQuery ( TableQuery < T > innerQuery )
{
_ innerQuery = innerQuery ;
}
public AsyncTableQuery < T > Where ( Expression < Func < T , bool > > predExpr )
{
return new AsyncTableQuery < T > ( _ innerQuery . Where ( predExpr ) ) ;
}
public AsyncTableQuery < T > Skip ( int n )
{
return new AsyncTableQuery < T > ( _ innerQuery . Skip ( n ) ) ;
}
public AsyncTableQuery < T > Take ( int n )
{
return new AsyncTableQuery < T > ( _ innerQuery . Take ( n ) ) ;
}
public AsyncTableQuery < T > OrderBy < U > ( Expression < Func < T , U > > orderExpr )
{
return new AsyncTableQuery < T > ( _ innerQuery . OrderBy < U > ( orderExpr ) ) ;
}
public AsyncTableQuery < T > OrderByDescending < U > ( Expression < Func < T , U > > orderExpr )
{
return new AsyncTableQuery < T > ( _ innerQuery . OrderByDescending < U > ( orderExpr ) ) ;
}
public Task < List < T > > ToListAsync ( )
{
return Task . Factory . StartNew ( ( ) = > {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) ) {
return _ innerQuery . ToList ( ) ;
}
} ) ;
}
public Task < int > CountAsync ( )
{
return Task . Factory . StartNew ( ( ) = > {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) ) {
return _ innerQuery . Count ( ) ;
}
} ) ;
}
public Task < T > ElementAtAsync ( int index )
{
return Task . Factory . StartNew ( ( ) = > {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) ) {
return _ innerQuery . ElementAt ( index ) ;
}
} ) ;
}
public Task < T > FirstAsync ( )
{
return Task < T > . Factory . StartNew ( ( ) = > {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) ) {
return _ innerQuery . First ( ) ;
}
} ) ;
}
public Task < T > FirstOrDefaultAsync ( )
{
return Task < T > . Factory . StartNew ( ( ) = > {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) ) {
return _ innerQuery . FirstOrDefault ( ) ;
}
} ) ;
}
} }
public class CreateTablesResult //
{ // TODO: Bind to AsyncConnection.GetConnection instead so that delayed
public Dictionary < Type , int > Results { get ; private set ; } // execution can still work after a Pool.Reset.
//
internal CreateTablesResult ( ) public class AsyncTableQuery < T >
{ where T : new ( )
this . Results = new Dictionary < Type , int > ( ) ; {
} TableQuery < T > _ innerQuery ;
}
public AsyncTableQuery ( TableQuery < T > innerQuery )
class SQLiteConnectionPool {
{ _ innerQuery = innerQuery ;
class Entry }
{
public SQLiteConnectionString ConnectionString { get ; private set ; } public AsyncTableQuery < T > Where ( Expression < Func < T , bool > > predExpr )
public SQLiteConnectionWithLock Connection { get ; private set ; } {
return new AsyncTableQuery < T > ( _ innerQuery . Where ( predExpr ) ) ;
public Entry ( SQLiteConnectionString connectionString ) }
{
ConnectionString = connectionString ; public AsyncTableQuery < T > Skip ( int n )
Connection = new SQLiteConnectionWithLock ( connectionString ) ; {
} return new AsyncTableQuery < T > ( _ innerQuery . Skip ( n ) ) ;
}
public void OnApplicationSuspended ( )
{ public AsyncTableQuery < T > Take ( int n )
Connection . Dispose ( ) ; {
Connection = null ; return new AsyncTableQuery < T > ( _ innerQuery . Take ( n ) ) ;
} }
}
public AsyncTableQuery < T > OrderBy < U > ( Expression < Func < T , U > > orderExpr )
readonly Dictionary < string , Entry > _ entries = new Dictionary < string , Entry > ( ) ; {
readonly object _ entriesLock = new object ( ) ; return new AsyncTableQuery < T > ( _ innerQuery . OrderBy < U > ( orderExpr ) ) ;
}
static readonly SQLiteConnectionPool _ shared = new SQLiteConnectionPool ( ) ;
public AsyncTableQuery < T > OrderByDescending < U > ( Expression < Func < T , U > > orderExpr )
/// <summary>
{
/// Gets the singleton instance of the connection tool.
return new AsyncTableQuery < T > ( _ innerQuery . OrderByDescending < U > ( orderExpr ) ) ;
/// </summary>
}
public static SQLiteConnectionPool Shared
{ public Task < List < T > > ToListAsync ( )
get {
{ return Task . Factory . StartNew ( ( ) = >
return _ shared ; {
} using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) )
} {
return _ innerQuery . ToList ( ) ;
public SQLiteConnectionWithLock GetConnection ( SQLiteConnectionString connectionString ) }
{ } ) ;
lock ( _ entriesLock ) { }
Entry entry ;
string key = connectionString . ConnectionString ; public Task < int > CountAsync ( )
{
if ( ! _ entries . TryGetValue ( key , out entry ) ) { return Task . Factory . StartNew ( ( ) = >
entry = new Entry ( connectionString ) ; {
_ entries [ key ] = entry ; using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) )
} {
return _ innerQuery . Count ( ) ;
return entry . Connection ; }
} } ) ;
} }
/// <summary>
public Task < T > ElementAtAsync ( int index )
/// Closes all connections managed by this pool.
{
/// </summary>
return Task . Factory . StartNew ( ( ) = >
public void Reset ( ) {
{ using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) )
lock ( _ entriesLock ) { {
foreach ( var entry in _ entries . Values ) { return _ innerQuery . ElementAt ( index ) ;
entry . OnApplicationSuspended ( ) ; }
} } ) ;
_ entries . Clear ( ) ; }
}
} public Task < T > FirstAsync ( )
{
/// <summary>
return Task < T > . Factory . StartNew ( ( ) = >
/// Call this method when the application is suspended.
{
/// </summary>
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) )
/// <remarks>Behaviour here is to close any open connections.</remarks>
{
public void ApplicationSuspended ( ) return _ innerQuery . First ( ) ;
{ }
Reset ( ) ; } ) ;
} }
}
public Task < T > FirstOrDefaultAsync ( )
class SQLiteConnectionWithLock : SQLiteConnection {
{ return Task < T > . Factory . StartNew ( ( ) = >
readonly object _l ockPoint = new object ( ) ; {
using ( ( ( SQLiteConnectionWithLock ) _ innerQuery . Connection ) . Lock ( ) )
public SQLiteConnectionWithLock ( SQLiteConnectionString connectionString ) {
: base ( connectionString . DatabasePath , connectionString . StoreDateTimeAsTicks ) return _ innerQuery . FirstOrDefault ( ) ;
{ }
} } ) ;
}
public IDisposable Lock ( ) }
{
return new LockWrapper ( _l ockPoint ) ; public class CreateTablesResult
} {
public Dictionary < Type , int > Results { get ; private set ; }
private class LockWrapper : IDisposable
{ internal CreateTablesResult ( )
object _l ockPoint ; {
this . Results = new Dictionary < Type , int > ( ) ;
public LockWrapper ( object lockPoint ) }
{ }
_l ockPoint = lockPoint ;
Monitor . Enter ( _l ockPoint ) ; class SQLiteConnectionPool
} {
class Entry
public void Dispose ( ) {
{ public SQLiteConnectionString ConnectionString { get ; private set ; }
Monitor . Exit ( _l ockPoint ) ; public SQLiteConnectionWithLock Connection { get ; private set ; }
}
} public Entry ( SQLiteConnectionString connectionString )
} {
ConnectionString = connectionString ;
Connection = new SQLiteConnectionWithLock ( connectionString ) ;
}
public void OnApplicationSuspended ( )
{
Connection . Dispose ( ) ;
Connection = null ;
}
}
readonly Dictionary < string , Entry > _ entries = new Dictionary < string , Entry > ( ) ;
readonly object _ entriesLock = new object ( ) ;
static readonly SQLiteConnectionPool _ shared = new SQLiteConnectionPool ( ) ;
/// <summary>
/// Gets the singleton instance of the connection tool.
/// </summary>
public static SQLiteConnectionPool Shared
{
get
{
return _ shared ;
}
}
public SQLiteConnectionWithLock GetConnection ( SQLiteConnectionString connectionString )
{
lock ( _ entriesLock )
{
Entry entry ;
string key = connectionString . ConnectionString ;
if ( ! _ entries . TryGetValue ( key , out entry ) )
{
entry = new Entry ( connectionString ) ;
_ entries [ key ] = entry ;
}
return entry . Connection ;
}
}
/// <summary>
/// Closes all connections managed by this pool.
/// </summary>
public void Reset ( )
{
lock ( _ entriesLock )
{
foreach ( var entry in _ entries . Values )
{
entry . OnApplicationSuspended ( ) ;
}
_ entries . Clear ( ) ;
}
}
/// <summary>
/// Call this method when the application is suspended.
/// </summary>
/// <remarks>Behaviour here is to close any open connections.</remarks>
public void ApplicationSuspended ( )
{
Reset ( ) ;
}
}
class SQLiteConnectionWithLock : SQLiteConnection
{
readonly object _l ockPoint = new object ( ) ;
public SQLiteConnectionWithLock ( SQLiteConnectionString connectionString )
: base ( connectionString . DatabasePath , connectionString . StoreDateTimeAsTicks )
{
}
public IDisposable Lock ( )
{
return new LockWrapper ( _l ockPoint ) ;
}
private class LockWrapper : IDisposable
{
object _l ockPoint ;
public LockWrapper ( object lockPoint )
{
_l ockPoint = lockPoint ;
Monitor . Enter ( _l ockPoint ) ;
}
public void Dispose ( )
{
Monitor . Exit ( _l ockPoint ) ;
}
}
}
} }
James South
13 years ago
