blob: a037b3d00f5142c4dff0ef42d2b64c09ace8e396 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
type _ Effect.t += Yield : Concepts.Tuple.t -> unit Effect.t
type step =
| Element of Concepts.Tuple.t * (unit, step) Effect.Deep.continuation
| Finished of (unit, Concepts.Condition.condition) result
let start produce =
try Finished (produce ~yield:(fun tuple -> Effect.perform (Yield tuple)))
with effect Yield tuple, k -> Element (tuple, k)
let resume k = Effect.Deep.continue k ()
class producer_cursor produce finally =
object (self)
inherit Lifecycle.counted
inherit Identity.of_id
method to_string = "producer-cursor"
val mutable state : [`Fresh | `Live of (unit, step) Effect.Deep.continuation | `Done] = `Fresh
method protocols : Protocols.Handle.protocol list = Protocols.[Cursor.make self]
method private finish =
state <- `Done;
Option.iter (fun f -> f ()) finally
method private step : (Concepts.Tuple.t option, Concepts.Condition.condition) result =
let settle = function
| Element (tuple, k) ->
state <- `Live k;
Ok (Some tuple)
| Finished r ->
self#finish;
Result.map (fun () -> None) r
in
match state with
| `Done -> Ok None
| `Fresh -> settle (start produce)
| `Live k -> settle (resume k)
method fetch limit =
let open Utilities.Result in
let rec fill taken tuples =
if taken >= limit then Ok Protocols.Cursor.{tuples; exhausted= false}
else
let* stepped = self#step in
match stepped with
| None -> Ok Protocols.Cursor.{tuples; exhausted= true}
| Some tuple -> fill (taken + 1) (BatFingerTree.snoc tuples tuple)
in
if limit <= 0 then Ok Protocols.Cursor.{tuples= BatFingerTree.empty; exhausted= state = `Done}
else fill 0 BatFingerTree.empty
method destroy = match state with `Fresh | `Live _ -> self#finish | `Done -> ()
end
let cursor_of ?finally produce = new producer_cursor produce finally |> Protocols.Handle.make
let filter keep cursor =
cursor_of
~finally:(fun () -> Protocols.Handle.release cursor)
(fun ~yield ->
let open Utilities.Result in
let* scan = Protocols.Cursor.require cursor in
let rec pass () =
let* next = Protocols.Cursor.next scan in
match next with
| None -> Ok ()
| Some tuple ->
if keep tuple then yield tuple;
pass ()
in
pass () )
|
