I ended up doing something like that:
let def<'T> =
Unchecked.defaultof<'T>
module Seq =
let mapbi mapping source =
source
|> Seq.scan(fun state item -> fst state + 1I, item) (-1I, def)
|> Seq.skip(1)
|> Seq.map(fun (bi, item) -> mapping bi item)
let mapChange selector mapping source =
source
|> mapbi (fun bi item -> bi, item)
|> Seq.scan(fun (previousSelection, previousMappedItem, _) (bi, item) ->
if bi = 0I then
(selector item, mapping item, item)
else
let currentSelection = selector item
let mappedItem = mapping item
if previousSelection <> currentSelection then
(currentSelection, mappedItem, item)
else
(previousSelection, previousMappedItem, item)
) (def, def, def)
|> Seq.skip 1
|> Seq.map(fun (_, mappedItem, item) -> mappedItem, item)
// and its AsyncSeq counterpart
module AsyncSeq =
let mapbi mapping source =
source
|> AsyncSeq.scan(fun state item -> fst state + 1I, item) (-1I, def)
|> AsyncSeq.skip(1)
|> AsyncSeq.map(fun (bi, item) -> mapping bi item)
let mapChange selector mapping source =
source
|> mapbi (fun bi item -> bi, item)
|> AsyncSeq.scan(fun (previousSelection, previousMappedItem, _) (bi, item) ->
if bi = 0I then
(selector item, mapping item, item)
else
let currentSelection = selector item
let mappedItem = mapping item
if previousSelection <> currentSelection then
(currentSelection, mappedItem, item)
else
(previousSelection, previousMappedItem, item)
) (def, def, def)
|> AsyncSeq.skip 1
|> AsyncSeq.map(fun (_, mappedItem, item) -> mappedItem, item)
(Dummy) Example:
[<EntryPoint>]
let main _ =
Seq.replicate 4 [1 .. 3]
|> Seq.mapi(fun i s ->
s |> Seq.map (fun item -> {| SetIndex = i; RecordIndex = item |} ))
|> Seq.concat
|> AsyncSeq.ofSeq
|> AsyncSeq.mapChange
(fun item -> item.SetIndex)
(fun item -> List.init (item.SetIndex + 1) (fun _ -> item.SetIndex))
|> AsyncSeq.iterAsync(fun (mappedItem, item) ->
async { printfn "%A: %A" item mappedItem })
|> Async.RunSynchronously
|> ignore
0
Output:
{ RecordIndex = 1
SetIndex = 0 }: [0]
{ RecordIndex = 2
SetIndex = 0 }: [0]
{ RecordIndex = 3
SetIndex = 0 }: [0]
{ RecordIndex = 1
SetIndex = 1 }: [1; 1]
{ RecordIndex = 2
SetIndex = 1 }: [1; 1]
{ RecordIndex = 3
SetIndex = 1 }: [1; 1]
{ RecordIndex = 1
SetIndex = 2 }: [2; 2; 2]
{ RecordIndex = 2
SetIndex = 2 }: [2; 2; 2]
{ RecordIndex = 3
SetIndex = 2 }: [2; 2; 2]
{ RecordIndex = 1
SetIndex = 3 }: [3; 3; 3; 3]
{ RecordIndex = 2
SetIndex = 3 }: [3; 3; 3; 3]
{ RecordIndex = 3
SetIndex = 3 }: [3; 3; 3; 3]
Two things that are more idiomatic that my initial piece of code:
- No mutable
- Still work even if the initial sequence is empty
- Still "Stream" everything => memory allocation over long sequence is roughly constant
Not fully satisfied, still kinda too cluttered =|