Advanced Task Sequence Operations

This page covers advanced TaskSeq<'T> operations: grouping, stateful transformation with mapFold, 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>.

scan and scanAsync

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

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

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