Split out cache
This commit is contained in:
@@ -5,86 +5,6 @@ open System.Net.Http
|
||||
open System.Threading
|
||||
open System.Threading.Tasks
|
||||
|
||||
type private CacheMessage<'a> =
|
||||
| TriggerUpdate
|
||||
| UpdateStored of 'a Task
|
||||
| Get of AsyncReplyChannel<'a>
|
||||
| Quit of AsyncReplyChannel<unit>
|
||||
|
||||
type Cache<'a> (obtainNew : CancellationToken -> 'a Task, expiry : 'a -> DateTime option) =
|
||||
let cts = new CancellationTokenSource ()
|
||||
|
||||
let initialValue = obtainNew cts.Token
|
||||
|
||||
let rec handle (value : 'a Task) (mailbox : MailboxProcessor<CacheMessage<'a>>) : Async<unit> =
|
||||
async {
|
||||
let! message = mailbox.Receive ()
|
||||
|
||||
match message with
|
||||
| Quit channel ->
|
||||
channel.Reply ()
|
||||
return ()
|
||||
| CacheMessage.UpdateStored newValue -> return! handle newValue mailbox
|
||||
| CacheMessage.TriggerUpdate ->
|
||||
async {
|
||||
let! a = Async.AwaitTask (obtainNew cts.Token)
|
||||
let expiry = expiry a
|
||||
|
||||
match expiry with
|
||||
| None -> return ()
|
||||
| Some expiry ->
|
||||
|
||||
// a bit sloppy but :shrug:
|
||||
do! Async.Sleep ((expiry - DateTime.Now) - TimeSpan.FromMinutes 1.0)
|
||||
|
||||
try
|
||||
mailbox.Post CacheMessage.TriggerUpdate
|
||||
with _ ->
|
||||
// Post during shutdown sequence: drop it on the floor
|
||||
()
|
||||
|
||||
return ()
|
||||
}
|
||||
|> fun a -> Async.Start (a, cancellationToken = cts.Token)
|
||||
|
||||
return! handle value mailbox
|
||||
| CacheMessage.Get reply ->
|
||||
let! valueAwaited = Async.AwaitTask value
|
||||
reply.Reply valueAwaited
|
||||
return! handle value mailbox
|
||||
}
|
||||
|
||||
let mailbox = new MailboxProcessor<_> (handle initialValue)
|
||||
|
||||
do
|
||||
mailbox.Start ()
|
||||
mailbox.Post CacheMessage.TriggerUpdate
|
||||
|
||||
let isDisposing = ref 0
|
||||
let hasDisposed = TaskCompletionSource<unit> ()
|
||||
|
||||
member this.GetCurrentValue () =
|
||||
try
|
||||
mailbox.PostAndAsyncReply CacheMessage.Get
|
||||
with
|
||||
// TODO I think this is the right exception...
|
||||
| :? InvalidOperationException ->
|
||||
raise (ObjectDisposedException (nameof (Cache)))
|
||||
|
||||
interface IDisposable with
|
||||
member _.Dispose () =
|
||||
if Interlocked.Increment isDisposing = 1 then
|
||||
mailbox.PostAndReply CacheMessage.Quit
|
||||
(mailbox :> IDisposable).Dispose ()
|
||||
// We can't terminate the CTS until the mailbox has processed all client requests.
|
||||
// Otherwise we terminate the mailbox's state Task before it has finished querying that
|
||||
// task on behalf of clients.
|
||||
cts.Cancel ()
|
||||
cts.Dispose ()
|
||||
hasDisposed.SetResult ()
|
||||
else
|
||||
hasDisposed.Task.Result
|
||||
|
||||
/// Methods for interacting with the PureGym REST API.
|
||||
[<RequireQualifiedAccess>]
|
||||
module Api =
|
||||
|
||||
86
PureGym/Cache.fs
Normal file
86
PureGym/Cache.fs
Normal file
@@ -0,0 +1,86 @@
|
||||
namespace PureGym
|
||||
|
||||
open System
|
||||
open System.Threading
|
||||
open System.Threading.Tasks
|
||||
|
||||
type private CacheMessage<'a> =
|
||||
| TriggerUpdate
|
||||
| UpdateStored of 'a Task
|
||||
| Get of AsyncReplyChannel<'a>
|
||||
| Quit of AsyncReplyChannel<unit>
|
||||
|
||||
type internal Cache<'a> (obtainNew : CancellationToken -> 'a Task, expiry : 'a -> DateTime option) =
|
||||
let cts = new CancellationTokenSource ()
|
||||
|
||||
let initialValue = obtainNew cts.Token
|
||||
|
||||
let rec handle (value : 'a Task) (mailbox : MailboxProcessor<CacheMessage<'a>>) : Async<unit> =
|
||||
async {
|
||||
let! message = mailbox.Receive ()
|
||||
|
||||
match message with
|
||||
| Quit channel ->
|
||||
channel.Reply ()
|
||||
return ()
|
||||
| CacheMessage.UpdateStored newValue -> return! handle newValue mailbox
|
||||
| CacheMessage.TriggerUpdate ->
|
||||
async {
|
||||
let! a = Async.AwaitTask (obtainNew cts.Token)
|
||||
let expiry = expiry a
|
||||
|
||||
match expiry with
|
||||
| None -> return ()
|
||||
| Some expiry ->
|
||||
|
||||
// a bit sloppy but :shrug:
|
||||
do! Async.Sleep ((expiry - DateTime.Now) - TimeSpan.FromMinutes 1.0)
|
||||
|
||||
try
|
||||
mailbox.Post CacheMessage.TriggerUpdate
|
||||
with _ ->
|
||||
// Post during shutdown sequence: drop it on the floor
|
||||
()
|
||||
|
||||
return ()
|
||||
}
|
||||
|> fun a -> Async.Start (a, cancellationToken = cts.Token)
|
||||
|
||||
return! handle value mailbox
|
||||
| CacheMessage.Get reply ->
|
||||
let! valueAwaited = Async.AwaitTask value
|
||||
reply.Reply valueAwaited
|
||||
return! handle value mailbox
|
||||
}
|
||||
|
||||
let mailbox = new MailboxProcessor<_> (handle initialValue)
|
||||
|
||||
do
|
||||
mailbox.Start ()
|
||||
mailbox.Post CacheMessage.TriggerUpdate
|
||||
|
||||
let isDisposing = ref 0
|
||||
let hasDisposed = TaskCompletionSource<unit> ()
|
||||
|
||||
member this.GetCurrentValue () =
|
||||
try
|
||||
mailbox.PostAndAsyncReply CacheMessage.Get
|
||||
with
|
||||
// TODO I think this is the right exception...
|
||||
| :? InvalidOperationException ->
|
||||
raise (ObjectDisposedException (nameof (Cache)))
|
||||
|
||||
interface IDisposable with
|
||||
member _.Dispose () =
|
||||
if Interlocked.Increment isDisposing = 1 then
|
||||
mailbox.PostAndReply CacheMessage.Quit
|
||||
(mailbox :> IDisposable).Dispose ()
|
||||
// We can't terminate the CTS until the mailbox has processed all client requests.
|
||||
// Otherwise we terminate the mailbox's state Task before it has finished querying that
|
||||
// task on behalf of clients.
|
||||
cts.Cancel ()
|
||||
cts.Dispose ()
|
||||
hasDisposed.SetResult ()
|
||||
else
|
||||
hasDisposed.Task.Result
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
<Compile Include="GeneratedClient.fs">
|
||||
<MyriadFile>Client.fs</MyriadFile>
|
||||
</Compile>
|
||||
<Compile Include="Cache.fs" />
|
||||
<Compile Include="Api.fs" />
|
||||
<Compile Include="GymSelector.fs" />
|
||||
<EmbeddedResource Include="SurfaceBaseline.txt" />
|
||||
|
||||
@@ -38,9 +38,6 @@ PureGym.AuthToken.get_ExpiryTime [method]: unit -> System.DateTime option
|
||||
PureGym.AuthTokenModule inherit obj
|
||||
PureGym.AuthTokenModule.get [static method]: PureGym.UsernamePin -> System.Threading.CancellationToken -> PureGym.AuthToken System.Threading.Tasks.Task
|
||||
PureGym.AuthTokenModule.ofBearerToken [static method]: string -> PureGym.AuthToken
|
||||
PureGym.Cache`1 inherit obj, implements IDisposable
|
||||
PureGym.Cache`1..ctor [constructor]: (System.Threading.CancellationToken -> 'a System.Threading.Tasks.Task, 'a -> System.DateTime option)
|
||||
PureGym.Cache`1.GetCurrentValue [method]: unit -> 'a Microsoft.FSharp.Control.FSharpAsync
|
||||
PureGym.Gym inherit obj, implements PureGym.Gym System.IEquatable, System.Collections.IStructuralEquatable, PureGym.Gym System.IComparable, System.IComparable, System.Collections.IStructuralComparable
|
||||
PureGym.Gym..ctor [constructor]: (string, int, int, PureGym.GymAddress, string, string, PureGym.GymOpeningHours, PureGym.GymAccessOptions, PureGym.GymLocation, string, string)
|
||||
PureGym.Gym.AccessOptions [property]: [read-only] PureGym.GymAccessOptions
|
||||
|
||||
Reference in New Issue
Block a user