Advanced Task Sequence Operations

This page covers advanced TaskSeq<'T> operations: grouping, stateful transformation with mapFold and threadState, deduplication, set-difference, partitioning, counting by key, lexicographic comparison, cancellation, and positional editing.

open System.Threading
open System.Threading.Tasks
open FSharp.Control

groupBy and groupByAsync

TaskSeq.groupBy partitions a sequence into groups by a key-projection function. The result is an array of (key, elements[]) pairs, one per distinct key, in order of first occurrence.

Note: groupBy consumes the entire source before returning. Do not use it on potentially infinite sequences.

type Event = { EntityId: int; Payload: string }

let events : TaskSeq<Event> =
    TaskSeq.ofList
        [ { EntityId = 1; Payload = "A" }
          { EntityId = 2; Payload = "B" }
          { EntityId = 1; Payload = "C" }
          { EntityId = 3; Payload = "D" }
          { EntityId = 2; Payload = "E" } ]

// groups: (1, [A;C]), (2, [B;E]), (3, [D])
let grouped : Task<(int * Event[])[]> =
    events |> TaskSeq.groupBy (fun e -> e.EntityId)

TaskSeq.groupByAsync accepts an async key projection:

let groupedAsync : Task<(int * Event[])[]> =
    events |> TaskSeq.groupByAsync (fun e -> task { return e.EntityId })

countBy and countByAsync

TaskSeq.countBy counts how many elements map to each key, returning (key, count)[]:

let counts : Task<(int * int)[]> =
    events |> TaskSeq.countBy (fun e -> e.EntityId)
// (1,2), (2,2), (3,1)

mapFold and mapFoldAsync

TaskSeq.mapFold threads a state accumulator through a sequence while simultaneously mapping each element to a result value. The output is a task returning a pair of (result[], finalState):

// Number each word sequentially while building a running concatenation
let words : TaskSeq<string> =
    TaskSeq.ofList [ "hello"; "world"; "foo" ]

let numbered : Task<string[] * int> =
    words
    |> TaskSeq.mapFold (fun count w -> $"{count}: {w}", count + 1) 0

// result: ([| "0: hello"; "1: world"; "2: foo" |], 3)

TaskSeq.mapFoldAsync is the same but the mapping function returns Task<'Result * 'State>.

threadState and threadStateAsync

TaskSeq.threadState is the lazy, streaming counterpart to mapFold. It threads a state accumulator through the sequence while yielding each mapped result — but unlike mapFold it never materialises the results into an array, and it discards the final state. This makes it suitable for infinite sequences and pipelines where intermediate results should be streamed rather than buffered:

let numbers : TaskSeq<int> = TaskSeq.ofSeq (seq { 1..5 })

// Produce a running total without collecting the whole sequence first
let runningSum : TaskSeq<int> =
    numbers
    |> TaskSeq.threadState (fun acc x -> acc + x, acc + x) 0

// yields lazily: 1, 3, 6, 10, 15

Compare with scan, which also emits a running result but prepends the initial state:

let viaScan = numbers |> TaskSeq.scan (fun acc x -> acc + x) 0
// yields: 0, 1, 3, 6, 10, 15  (one extra initial element)

let viaThreadState = numbers |> TaskSeq.threadState (fun acc x -> acc + x, acc + x) 0
// yields: 1, 3, 6, 10, 15  (no initial element; result == new state here)

TaskSeq.threadStateAsync accepts an asynchronous folder:

let asyncRunningSum : TaskSeq<int> =
    numbers
    |> TaskSeq.threadStateAsync (fun acc x -> Task.fromResult (acc + x, acc + x)) 0

TaskSeq.scan is the streaming sibling of fold: it emits each intermediate state as a new element, starting with the initial state:

let runningTotals : TaskSeq<int> =
    numbers |> TaskSeq.scan (fun acc n -> acc + n) 0

// yields: 0, 1, 3, 6, 10, 15

distinct and distinctBy

TaskSeq.distinct removes duplicates (keeps first occurrence), using generic equality:

let withDups : TaskSeq<int> = TaskSeq.ofList [ 1; 2; 2; 3; 1; 4 ]

let deduped : TaskSeq<int> = withDups |> TaskSeq.distinct // 1, 2, 3, 4

TaskSeq.distinctBy deduplicates by a key projection:

let strings : TaskSeq<string> =
    TaskSeq.ofList [ "hello"; "HELLO"; "world"; "WORLD" ]

let caseInsensitiveDistinct : TaskSeq<string> =
    strings |> TaskSeq.distinctBy (fun s -> s.ToLowerInvariant())
// "hello", "world"

Note: both distinct and distinctBy buffer all unique keys in a hash set. Do not use them on potentially infinite sequences.

TaskSeq.distinctByAsync accepts an async key projection.

distinctUntilChanged

TaskSeq.distinctUntilChanged removes consecutive duplicates only — it does not buffer the whole sequence, so it is safe on infinite streams:

let run : TaskSeq<int> = TaskSeq.ofList [ 1; 1; 2; 2; 2; 3; 1; 1 ]

let noConsecDups : TaskSeq<int> = run |> TaskSeq.distinctUntilChanged
// 1, 2, 3, 1

except and exceptOfSeq

TaskSeq.except itemsToExclude source returns elements of source that do not appear in itemsToExclude. The exclusion set is materialised eagerly before iteration:

let exclusions : TaskSeq<int> = TaskSeq.ofList [ 2; 4 ]
let source : TaskSeq<int> = TaskSeq.ofSeq (seq { 1..5 })

let filtered : TaskSeq<int> = TaskSeq.except exclusions source // 1, 3, 5

TaskSeq.exceptOfSeq accepts a plain seq<'T> as the exclusion set.

partition and partitionAsync

TaskSeq.partition splits the sequence into two arrays in a single pass. Elements for which the predicate returns true go into the first array; the rest into the second:

let partitioned : Task<int[] * int[]> =
    source |> TaskSeq.partition (fun n -> n % 2 = 0)
// trueItems: [|2;4|]   falseItems: [|1;3;5|]

TaskSeq.partitionAsync accepts an async predicate.

compareWith and compareWithAsync

TaskSeq.compareWith performs a lexicographic comparison of two sequences using a custom comparer. It returns the first non-zero comparison result, or 0 if the sequences are element-wise equal and have the same length:

let a : TaskSeq<int> = TaskSeq.ofList [ 1; 2; 3 ]
let b : TaskSeq<int> = TaskSeq.ofList [ 1; 2; 4 ]

let cmp : Task<int> =
    TaskSeq.compareWith (fun x y -> compare x y) a b
// negative (a < b)

TaskSeq.compareWithAsync accepts an async comparer.

withCancellation

TaskSeq.withCancellation token source injects a CancellationToken into the underlying IAsyncEnumerable<'T>. This is equivalent to calling .WithCancellation(token) in C# and is useful when consuming sequences from libraries (e.g. Entity Framework Core) that require a token at the enumeration site:

let cts = new CancellationTokenSource()

let cancellable : TaskSeq<int> =
    source |> TaskSeq.withCancellation cts.Token

Positional editing

TaskSeq.insertAt, TaskSeq.insertManyAt, TaskSeq.removeAt, TaskSeq.removeManyAt, and TaskSeq.updateAt produce new sequences with an element inserted, removed, or replaced at a given zero-based index:

let original : TaskSeq<int> = TaskSeq.ofList [ 1; 2; 4; 5 ]

let inserted : TaskSeq<int> = original |> TaskSeq.insertAt 2 3 // 1,2,3,4,5

let removed : TaskSeq<int> = original |> TaskSeq.removeAt 1 // 1,4,5

let updated : TaskSeq<int> = original |> TaskSeq.updateAt 0 99 // 99,2,4,5

let manyInserted : TaskSeq<int> =
    original
    |> TaskSeq.insertManyAt 2 (TaskSeq.ofList [ 10; 11 ])
// 1, 2, 10, 11, 4, 5

let manyRemoved : TaskSeq<int> = original |> TaskSeq.removeManyAt 1 2 // 1, 5