<commit_before>﻿// Copyright (C) by <NAME>

namespace Hopac.Extra

open System
open System.Collections.Generic
open Hopac
open Hopac.Infixes
open Hopac.Job.Infixes
open Hopac.Alt.Infixes
open Hopac.Extensions

[<AutoOpen>]
module internal Util =
  let inline memo x = Promise.Now.delay x :> Alt<_>
  let inline (>>=*) x f = x >>= f |> memo
  let inline (|>>*) x f = x |>> f |> memo
  let inline (<|>*) x y = x <|>? y |> memo

  let inline tryIn u2v vK eK =
    let mutable e = null
    let v = try u2v () with e' -> e <- e' ; Unchecked.defaultof<_>
    match e with
     | null -> vK v
     | e -> eK e

  let inline (|Nothing|Just|) (b, x) = if b then Just x else Nothing

type Stream<'x> =
  | Nil
  | Cons of Value: 'x * Next: Alt<Stream<'x>>

type Streams<'x> = Alt<Stream<'x>>

type StreamSrc<'x> = {tail: MVar<IVar<Stream<'x>>>}

module StreamSrc =
  let create () = {tail = mvarFull (ivar ())}
  let inline doTail ss op =
    ss.tail >>=? fun tail ->
    if IVar.Now.isFull tail
    then ss.tail <<-= tail >>! Exception ("StreamSrc: closed")
    else op tail
  let value ss x =
    doTail ss <| fun tail ->
    let next = ivar ()
    ss.tail <<-= next >>. (tail <-= Cons (x, next))
  let error ss e = doTail ss <| fun tail -> tail <-=! e >>. (ss.tail <<-= tail)
  let close ss = doTail ss <| fun tail -> tail <-= Nil >>. (ss.tail <<-= tail)
  let tap ss = MVar.read ss.tail >>= IVar.read |> memo

type StreamVar<'x> = {state: MVar<Stream<'x>>}

module StreamVar =
  let create x = {state = mvarFull (Cons (x, ivar ()))}
  let inline mapc f = function Nil -> raise <| Exception ("StreamVar: closed")
                             | Cons (x, n) -> f x n
  let get sv = MVar.read sv.state |>>? mapc (fun x _ -> x)
  let inline update sv (n: Alt<_>) x' =
    let c = Cons (x', ivar ()) in (n :?> IVar<_>) <-= c >>. (sv.state <<-= c)
  let set sv x = sv.state >>=? mapc (fun _ n -> update sv n x)
  let updateJob sv x2xJ = sv.state >>=? mapc (fun x n -> x2xJ x >>= update sv n)
  let updateFun sv x2x = sv.state >>=? mapc (fun x n -> update sv n (x2x x))
  let maybeUpdateFun sv x2xO =
    sv.state >>=? fun c -> c |> mapc (fun x n ->
    match x2xO x with
     | None -> sv.state <<-= c
     | Some x' -> update sv n x')
  let tap sv = MVar.read sv.state |> memo
  let setIfNotEq sv n =
    maybeUpdateFun sv (fun o -> if n <> o then Some n else None)
<commit_msg>Refactored a bit.<commit_after>﻿// Copyright (C) by <NAME>

namespace Hopac.Extra

open System
open System.Collections.Generic
open Hopac
open Hopac.Infixes
open Hopac.Job.Infixes
open Hopac.Alt.Infixes
open Hopac.Extensions

[<AutoOpen>]
module internal Util =
  let inline memo x = Promise.Now.delay x :> Alt<_>
  let inline (>>=*) x f = x >>= f |> memo
  let inline (|>>*) x f = x |>> f |> memo
  let inline (<|>*) x y = x <|>? y |> memo

  let inline tryIn u2v vK eK =
    let mutable e = null
    let v = try u2v () with e' -> e <- e' ; Unchecked.defaultof<_>
    match e with
     | null -> vK v
     | e -> eK e

  let inline (|Nothing|Just|) (b, x) = if b then Just x else Nothing

type Stream<'x> =
  | Nil
  | Cons of Value: 'x * Next: Alt<Stream<'x>>

type Streams<'x> = Alt<Stream<'x>>

type StreamSrc<'x> = {src: MVar<IVar<Stream<'x>>>}

module StreamSrc =
  let create () = {src = mvarFull (ivar ())}
  let inline doTail ss op =
    ss.src >>=? fun tail ->
    if IVar.Now.isFull tail
    then ss.src <<-= tail >>! Exception ("StreamSrc: closed")
    else op tail
  let value ss x =
    doTail ss <| fun tail ->
    let next = ivar ()
    ss.src <<-= next >>. (tail <-= Cons (x, next))
  let error ss e = doTail ss <| fun tail -> tail <-=! e >>. (ss.src <<-= tail)
  let close ss = doTail ss <| fun tail -> tail <-= Nil >>. (ss.src <<-= tail)
  let tap ss = MVar.read ss.src >>= IVar.read |> memo

type StreamVar<'x> = {var: MVar<Stream<'x>>}

module StreamVar =
  let create x = {var = mvarFull (Cons (x, ivar ()))}
  let inline op sv f =
    sv.var >>=? function
     | Cons (x, n) as c -> f c x n
     | Nil as c -> sv.var <<-= c >>! Exception ("StreamVar: closed") // XXX
  let get sv = op sv <| fun c x _ -> sv.var <<-= c >>% x
  let inline value sv (n: Alt<_>) x' =
    let c = Cons (x', ivar ()) in (n :?> IVar<_>) <-= c >>. (sv.var <<-= c)
  let set sv x = op sv <| fun _ _ n -> value sv n x
  let updateJob sv x2xJ = op sv <| fun _ x n -> x2xJ x >>= value sv n
  let updateFun sv x2x = op sv <| fun _ x n -> value sv n (x2x x)
  let maybeUpdateFun sv x2xO =
    op sv <| fun c x n -> match x2xO x with None -> sv.var <<-= c
                                          | Some x' -> value sv n x'
  let tap sv = MVar.read sv.var |> memo
  let setIfNotEq sv n =
    maybeUpdateFun sv (fun o -> if n <> o then Some n else None)

{'fraction_non_alphanumeric': 0.16769263426024197, 'fraction_numerical': 0.003396306516663129, 'mean_word_length': 2.4294032023289667, 'pattern_counts': {'":': 0, '<': 97, '<?xml version=': 0, '>': 161, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6930021', 'n_tokens_mistral': 1820, 'n_tokens_neox': 1805, 'n_words': 665}