3 Commits

Author SHA1 Message Date
1bfdcf165c Launch all tasks as background tasks
- Standardize on sync/async prefixes
  - affects toList, CreateConnection, ignore*
- Target netstandard-2.0 for max compatibility
- Update dependencies
2024-12-15 22:59:10 -05:00
a09181eee0 Add logging CreateConnection overloads 2022-07-19 09:12:57 -04:00
5d15de240d Add URI configuration option 2022-07-18 22:32:32 -04:00
9 changed files with 796 additions and 566 deletions

View File

@@ -41,13 +41,15 @@ let fetchPost (postId : string) =
} }
``` ```
### A standard way to translate JSON into a strongly-typed configuration ### A standard way to translate JSON or a URI into a strongly-typed configuration
```fsharp ```fsharp
/// type: DataConfig /// type: DataConfig
let config = DataConfig.fromJsonFile "data-config.json" let config = DataConfig.fromJsonFile "data-config.json"
// OR // OR
let config = DataConfig.fromConfiguration (config.GetSection "RethinkDB") let config = DataConfig.fromConfiguration (config.GetSection "RethinkDB")
// OR
let config = DataConfig.fromUri (config.GetConnectionString "RethinkDB")
/// type: IConnection /// type: IConnection
let conn = config.Connect () let conn = config.Connect ()

File diff suppressed because it is too large Load Diff

View File

@@ -1,10 +1,11 @@
namespace RethinkDb.Driver.FSharp namespace RethinkDb.Driver.FSharp
open System
open Microsoft.Extensions.Configuration open Microsoft.Extensions.Configuration
open Microsoft.Extensions.Logging
open Newtonsoft.Json.Linq open Newtonsoft.Json.Linq
open RethinkDb.Driver open RethinkDb.Driver
open RethinkDb.Driver.Net open RethinkDb.Driver.Net
open System.Threading.Tasks
/// Parameters for the RethinkDB configuration /// Parameters for the RethinkDB configuration
type DataConfigParameter = type DataConfigParameter =
@@ -36,17 +37,42 @@ type DataConfig =
static member empty = static member empty =
{ Parameters = [] } { Parameters = [] }
/// Create a RethinkDB connection /// Build the connection from the given parameters
member this.CreateConnection () : IConnection = member private this.BuildConnection() =
this.Parameters this.Parameters
|> Seq.fold ConnectionBuilder.build (RethinkDB.R.Connection()) |> Seq.fold ConnectionBuilder.build (RethinkDB.R.Connection())
|> function builder -> builder.Connect ()
/// Create a RethinkDB connection /// Create a RethinkDB connection (task async)
member this.CreateConnectionAsync () : Task<Connection> = member this.CreateConnection() = backgroundTask {
this.Parameters let! conn = this.BuildConnection().ConnectAsync()
|> Seq.fold ConnectionBuilder.build (RethinkDB.R.Connection ()) return conn :> IConnection
|> function builder -> builder.ConnectAsync () }
/// Create a RethinkDB connection, logging the connection settings (task async)
member this.CreateConnection(log: ILogger) = backgroundTask {
let builder = this.BuildConnection()
if not (isNull log) then log.LogInformation $"Connecting to {this.EffectiveUri}"
let! conn = builder.ConnectAsync()
return conn :> IConnection
}
/// Create a RethinkDB connection (F# async)
member this.CreateAsyncConnection() =
this.CreateConnection() |> Async.AwaitTask
/// Create a RethinkDB connection, logging the connection settings (F# async)
member this.CreateAsyncConnection(log: ILogger) =
this.CreateConnection(log) |> Async.AwaitTask
/// Create a RethinkDB connection (sync)
member this.CreateSyncConnection() : IConnection =
this.BuildConnection().Connect()
/// Create a RethinkDB connection, logging the connection settings (sync)
member this.CreateSyncConnection(log: ILogger) : IConnection =
let builder = this.BuildConnection()
if not (isNull log) then log.LogInformation $"Connecting to {this.EffectiveUri}"
builder.Connect()
/// The effective hostname /// The effective hostname
member this.Hostname = member this.Hostname =
@@ -72,6 +98,20 @@ type DataConfig =
| Some (Database x) -> x | Some (Database x) -> x
| _ -> RethinkDBConstants.DefaultDbName | _ -> RethinkDBConstants.DefaultDbName
/// The effective configuration URI (excludes password / auth key)
member this.EffectiveUri =
seq {
"rethinkdb://"
match this.Parameters |> List.tryPick (fun x -> match x with User _ -> Some x | _ -> None) with
| Some (User(username, _)) -> $"{username}:***pw***@"
| _ ->
match this.Parameters |> List.tryPick (fun x -> match x with AuthKey _ -> Some x | _ -> None) with
| Some (AuthKey _) -> "****key****@"
| _ -> ()
$"{this.Hostname}:{this.Port}/{this.Database}?timeout={this.Timeout}"
}
|> String.concat ""
/// Parse settings from JSON /// Parse settings from JSON
/// ///
/// A sample JSON object with all the possible properties filled in: /// A sample JSON object with all the possible properties filled in:
@@ -116,3 +156,26 @@ type DataConfig =
match cfg["username"], cfg["password"] with null, _ | _, null -> () | user -> User user match cfg["username"], cfg["password"] with null, _ | _, null -> () | user -> User user
] ]
} }
/// Parse settings from a URI
///
/// rethinkdb://user:password@host:port/database?timeout=##
/// OR
/// rethinkdb://authkey@host:port/database?timeout=##
///
/// Scheme and host are required; all other settings optional
static member FromUri(uri: string) =
let it = Uri uri
if it.Scheme <> "rethinkdb" then invalidArg "Scheme" $"""URI scheme must be "rethinkdb" (was {it.Scheme})"""
{ Parameters =
[ Hostname it.Host
if it.Port <> -1 then Port it.Port
if it.UserInfo <> "" then
if it.UserInfo.Contains ":" then
let parts = it.UserInfo.Split ':' |> Array.truncate 2
User(parts[0], parts[1])
else AuthKey it.UserInfo
if it.Segments.Length > 1 then Database it.Segments[1]
if it.Query.Contains "?timeout=" then Timeout(int it.Query[9..])
]
}

View File

@@ -18,15 +18,17 @@ module private Helpers =
// ~~ WRITES ~~ // ~~ WRITES ~~
/// Write a ReQL command with a cancellation token, always returning a result /// Write a ReQL command with a cancellation token, always returning a result
let runWriteResultWithCancel cancelToken (expr : ReqlExpr) = fun conn -> let runWriteResultWithCancel cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunWriteAsync (conn, cancelToken) return! expr.RunWriteAsync(conn, cancelToken)
}
/// Write a ReQL command, always returning a result /// Write a ReQL command, always returning a result
let runWriteResult expr = runWriteResultWithCancel CancellationToken.None expr let runWriteResult expr = runWriteResultWithCancel CancellationToken.None expr
/// Write a ReQL command with optional arguments and a cancellation token, always returning a result /// Write a ReQL command with optional arguments and a cancellation token, always returning a result
let runWriteResultWithOptArgsAndCancel args cancelToken (expr : ReqlExpr) = fun conn -> let runWriteResultWithOptArgsAndCancel args cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunWriteAsync (conn, RunOptArg.create args, cancelToken) return! expr.RunWriteAsync(conn, RunOptArg.create args, cancelToken)
}
/// Write a ReQL command with optional arguments, always returning a result /// Write a ReQL command with optional arguments, always returning a result
let runWriteResultWithOptArgs args expr = runWriteResultWithOptArgsAndCancel args CancellationToken.None expr let runWriteResultWithOptArgs args expr = runWriteResultWithOptArgsAndCancel args CancellationToken.None expr
@@ -108,12 +110,14 @@ let syncWriteWithOptArgs args expr = fun conn ->
// ~~ Full results (atom / sequence) ~~ // ~~ Full results (atom / sequence) ~~
/// Run the ReQL command using a cancellation token, returning the result as the type specified /// Run the ReQL command using a cancellation token, returning the result as the type specified
let runResultWithCancel<'T> cancelToken (expr : ReqlExpr) = fun conn -> let runResultWithCancel<'T> cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunResultAsync<'T> (conn, cancelToken) return! expr.RunResultAsync<'T>(conn, cancelToken)
}
/// Run the ReQL command using optional arguments and a cancellation token, returning the result as the type specified /// Run the ReQL command using optional arguments and a cancellation token, returning the result as the type specified
let runResultWithOptArgsAndCancel<'T> args cancelToken (expr : ReqlExpr) = fun conn -> let runResultWithOptArgsAndCancel<'T> args cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunResultAsync<'T> (conn, RunOptArg.create args, cancelToken) return! expr.RunResultAsync<'T>(conn, RunOptArg.create args, cancelToken)
}
/// Run the ReQL command, returning the result as the type specified /// Run the ReQL command, returning the result as the type specified
let runResult<'T> expr = runResultWithCancel<'T> CancellationToken.None expr let runResult<'T> expr = runResultWithCancel<'T> CancellationToken.None expr
@@ -165,20 +169,22 @@ let asAsyncOption (f : IConnection -> Async<'T>) conn = async {
let asSyncOption (f: IConnection -> 'T) conn = nullToOption (f conn) let asSyncOption (f: IConnection -> 'T) conn = nullToOption (f conn)
/// Ignore the result of a task-based query /// Ignore the result of a task-based query
let ignoreResult<'T> (f : IConnection -> Tasks.Task<'T>) conn = task { let ignoreResult<'T> (f: IConnection -> Tasks.Task<'T>) conn = backgroundTask {
let! _ = (f conn).ConfigureAwait false let! _ = f conn
() ()
} }
// ~~ Cursors / partial results (sequence / partial) ~~ // ~~ Cursors / partial results (sequence / partial) ~~
/// Run the ReQL command using a cancellation token, returning a cursor for the type specified /// Run the ReQL command using a cancellation token, returning a cursor for the type specified
let runCursorWithCancel<'T> cancelToken (expr : ReqlExpr) = fun conn -> let runCursorWithCancel<'T> cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunCursorAsync<'T> (conn, cancelToken) return! expr.RunCursorAsync<'T>(conn, cancelToken)
}
/// Run the ReQL command using optional arguments and a cancellation token, returning a cursor for the type specified /// Run the ReQL command using optional arguments and a cancellation token, returning a cursor for the type specified
let runCursorWithOptArgsAndCancel<'T> args cancelToken (expr : ReqlExpr) = fun conn -> let runCursorWithOptArgsAndCancel<'T> args cancelToken (expr: ReqlExpr) = fun conn -> backgroundTask {
expr.RunCursorAsync<'T> (conn, RunOptArg.create args, cancelToken) return! expr.RunCursorAsync<'T>(conn, RunOptArg.create args, cancelToken)
}
/// Run the ReQL command, returning a cursor for the type specified /// Run the ReQL command, returning a cursor for the type specified
let runCursor<'T> expr = runCursorWithCancel<'T> CancellationToken.None expr let runCursor<'T> expr = runCursorWithCancel<'T> CancellationToken.None expr
@@ -229,13 +235,13 @@ let toList<'T> (f : IConnection -> Task<Cursor<'T>>) = fun conn -> backgroundTas
} }
/// Convert a cursor to a list of items /// Convert a cursor to a list of items
let toListAsync<'T> (f : IConnection -> Async<Cursor<'T>>) = fun conn -> async { let asyncToList<'T> (f: IConnection -> Async<Cursor<'T>>) = fun conn -> async {
use! cursor = f conn use! cursor = f conn
return! cursorToList<'T> cursor |> Async.AwaitTask return! cursorToList<'T> cursor |> Async.AwaitTask
} }
/// Convert a cursor to a list of items /// Convert a cursor to a list of items
let toListSync<'T> (f : IConnection -> Cursor<'T>) = fun conn -> let syncToList<'T> (f: IConnection -> Cursor<'T>) = fun conn ->
use cursor = f conn use cursor = f conn
cursorToList cursor |> Async.AwaitTask |> Async.RunSynchronously cursorToList cursor |> Async.AwaitTask |> Async.RunSynchronously
@@ -640,7 +646,9 @@ let withRetry<'T> intervals f =
/// Convert an async function to a task function (Polly does not understand F# Async) /// Convert an async function to a task function (Polly does not understand F# Async)
let private asyncFuncToTask<'T> (f : IConnection -> Async<'T>) = let private asyncFuncToTask<'T> (f : IConnection -> Async<'T>) =
fun conn -> f conn |> Async.StartAsTask fun conn -> backgroundTask {
return! f conn |> Async.StartAsTask
}
/// Retry, delaying for each the seconds provided (if required) /// Retry, delaying for each the seconds provided (if required)
let withAsyncRetry<'T> intervals f = fun conn -> let withAsyncRetry<'T> intervals f = fun conn ->

View File

@@ -159,10 +159,10 @@ val syncCursorWithOptArgs<'T> : RunOptArg list -> ReqlExpr -> (IConnection -> Cu
val toList<'T> : (IConnection -> Task<Cursor<'T>>) -> IConnection -> Task<'T list> val toList<'T> : (IConnection -> Task<Cursor<'T>>) -> IConnection -> Task<'T list>
/// Convert a cursor to a list of items /// Convert a cursor to a list of items
val toListAsync<'T> : (IConnection -> Async<Cursor<'T>>) -> IConnection -> Async<'T list> val asyncToList<'T> : (IConnection -> Async<Cursor<'T>>) -> IConnection -> Async<'T list>
/// Convert a cursor to a list of items /// Convert a cursor to a list of items
val toListSync<'T> : (IConnection -> Cursor<'T>) -> IConnection -> 'T list val syncToList<'T> : (IConnection -> Cursor<'T>) -> IConnection -> 'T list
/// Apply a connection to the query pipeline (typically the final step) /// Apply a connection to the query pipeline (typically the final step)
val withConn<'T> : IConnection -> (IConnection -> 'T) -> 'T val withConn<'T> : IConnection -> (IConnection -> 'T) -> 'T

View File

@@ -374,4 +374,3 @@ module UpdateOptArg =
| NonAtomic non -> upd.OptArg("non_atomic", non) | NonAtomic non -> upd.OptArg("non_atomic", non)
| IgnoreWriteHook ign -> upd.OptArg("ignore_write_hook", ign)) | IgnoreWriteHook ign -> upd.OptArg("ignore_write_hook", ign))
u u

View File

@@ -10,6 +10,8 @@ open RethinkDb.Driver.FSharp
let dataCfg = DataConfig.fromJson "rethink-config.json" let dataCfg = DataConfig.fromJson "rethink-config.json"
// - or - // - or -
let dataCfg = DataConfig.fromConfiguration [config-section] let dataCfg = DataConfig.fromConfiguration [config-section]
// - or -
let dataCfg = DataConfig.fromUri [connection-string]
let conn = dataCfg.CreateConnection () // IConnection let conn = dataCfg.CreateConnection () // IConnection
``` ```

View File

@@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>net6.0;netstandard2.0</TargetFrameworks> <TargetFramework>netstandard2.0</TargetFramework>
<Description>Idiomatic F# extensions on the official RethinkDB C# driver</Description> <Description>Idiomatic F# extensions on the official RethinkDB C# driver</Description>
<Authors>Daniel J. Summers,Bit Badger Solutions</Authors> <Authors>Daniel J. Summers,Bit Badger Solutions</Authors>
<PackageLicenseExpression>Apache-2.0</PackageLicenseExpression> <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
@@ -13,10 +13,9 @@
<PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance> <PackageRequireLicenseAcceptance>false</PackageRequireLicenseAcceptance>
<Copyright>See LICENSE</Copyright> <Copyright>See LICENSE</Copyright>
<PackageTags>RethinkDB document F#</PackageTags> <PackageTags>RethinkDB document F#</PackageTags>
<VersionPrefix>0.9.0</VersionPrefix> <VersionPrefix>1.0.0</VersionPrefix>
<VersionSuffix>beta-06</VersionSuffix>
<PackageReleaseNotes> <PackageReleaseNotes>
Add optional arguments for tableCreate Add URI config option and logging CreateConnection overloads
</PackageReleaseNotes> </PackageReleaseNotes>
</PropertyGroup> </PropertyGroup>
@@ -32,11 +31,11 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Microsoft.Extensions.Configuration" Version="6.0.1" /> <PackageReference Include="Microsoft.Extensions.Configuration" Version="9.0.0" />
<PackageReference Include="Newtonsoft.Json" Version="13.0.1" /> <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="Polly" Version="7.2.3" /> <PackageReference Include="Polly" Version="8.5.0" />
<PackageReference Include="RethinkDb.Driver" Version="2.*" /> <PackageReference Include="RethinkDb.Driver" Version="2.*" />
<PackageReference Update="FSharp.Core" Version="6.0.3" /> <PackageReference Update="FSharp.Core" Version="9.0.100" />
</ItemGroup> </ItemGroup>
</Project> </Project>