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

namespace Hopac

open System
open System.Collections.Generic
open System.Threading
open Hopac.Infixes
open Hopac.Job.Infixes
open Hopac.Alt.Infixes
open Hopac.Promise.Infixes
open Hopac.Extensions
open Timer.Global

module Stream =
  let inline (<|>*) x y = x <|>* y :> Alt<_>
  let inline (|>>*) x y = x |>>* y :> Alt<_>
  let inline memo x = Promise.Now.delay x :> Alt<_>
  let inline start x = Job.Global.start x
  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 Cons<'x> =
    | Cons of Value: 'x * Next: Alt<Cons<'x>>
    | Nil

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

  type Src<'x> = {mutable src: IVar<Cons<'x>>}

  module Src =
    let create () = {src = IVar<_> ()}
    let rec value s x = Job.delay <| fun () ->
      let w = IVar<_> ()
      let v = s.src
      if IVar.Now.isFull v then raise <| Exception ("Src closed")
      let v' = Interlocked.CompareExchange (&s.src, w, v)
      if LanguagePrimitives.PhysicalEquality v' v
      then v <-= Cons (x, w)
      else value s x
    let error s e = Job.delay <| fun () -> s.src <-=! e // delay required
    let close s = Job.delay <| fun () -> s.src <-= Nil // delay required
    let tap s = s.src :> Alt<_>

  type Var<'x> = {mutable var: Cons<'x>}

  module Var =
    let create x = {var = Cons (x, IVar<_> ())}
    let get v = match v.var with Cons (x, _) -> x | Nil -> failwith "Impossible"
    let set v x = Job.delay <| fun () ->
      let c = Cons (x, IVar<_> ())
      match Interlocked.Exchange (&v.var, c) with
       | Cons (_, i) -> (i :?> IVar<_>) <-= c
       | Nil -> failwith "Impossible"
    let tap v = Alt.always v.var

  let inline nil<'x> = Alt.always Nil :> Stream<'x>
  let cons x xs = Alt.always (Cons (x, xs))
  let inline error e = Alt.raises e :> Stream<_>
  let inline one x = cons x nil
  let inline delay (u2xs: unit -> #Stream<'x>) =
    memo << Job.delay <| fun () -> u2xs ()

  let fix (xs2xs: Stream<'x> -> #Stream<'x>) =
    let xs = Promise<_> () // XXX publish interface for this?
    xs.Readers <- Promise<_>.Fulfill (Job.delay <| fun () -> xs2xs xs)
    xs :> Stream<_>

  let inline never<'x> = Alt.never () :> Stream<'x>

  let rec ofEnum (xs: IEnumerator<'x>) = memo << Job.thunk <| fun () ->
    if xs.MoveNext () then Cons (xs.Current, ofEnum xs) else xs.Dispose () ; Nil

  let ofSeq (xs: seq<_>) = delay (ofEnum << xs.GetEnumerator)

  let rec onCloseJob (uJ: Job<unit>) xs =
    Job.tryIn xs
       <| function Cons (x, xs) -> cons x (onCloseJob uJ xs) :> Job<_>
                 | Nil -> uJ >>% Nil
       <| fun e -> uJ >>! e
    |> memo
  let onCloseFun u2u xs = onCloseJob (Job.thunk u2u) xs

  type Subscribe<'x> (src: IVar<Cons<'x>>) =
    let mutable src = src
    interface IObserver<'x> with
     override t.OnCompleted () = src <-= Nil |> start
     override t.OnError (e) = src <-=! e |> start
     override t.OnNext (x) =
       let nxt = IVar<_> ()
       src <-= Cons (x, nxt) |> start
       src <- nxt

  let subscribeDuring (xs2ys: Stream<'x> -> #Stream<'y>) (xs: IObservable<'x>) =
    delay <| fun () ->
    let src = IVar<_> ()
    xs2ys src |> onCloseFun (xs.Subscribe (Subscribe (src))).Dispose

  let subscribeOnFirst (xs: IObservable<'x>) = delay <| fun () ->
    let src = IVar<_> () in xs.Subscribe (Subscribe (src)) |> ignore ; src

  let subscribingTo (xs: IObservable<'x>) (xs2yJ: Stream<'x> -> #Job<'y>) =
    Job.delay <| fun () ->
    let src = IVar<_> ()
    Job.using (xs.Subscribe (Subscribe (src))) <| fun _ -> xs2yJ src :> Job<_>

  type Subscriber<'x> (src: IVar<Cons<'x>>) =
    let mutable src = src
    [<DefaultValue>] val mutable disp: IDisposable
    interface IObserver<'x> with
     override t.OnCompleted () = t.disp <- null; src <-= Nil |> start
     override t.OnError (e) = t.disp <- null; src <-=! e |> start
     override t.OnNext (x) =
       let nxt = IVar<_> ()
       src <-= Cons (x, nxt) |> start
       src <- nxt

  let inline post (ctxt: SynchronizationContext) op =
    match ctxt with null -> op () | ctxt -> ctxt.Post ((fun _ -> op ()), null)

  type Guard<'x> (subr: Subscriber<'x>, ctxt: SynchronizationContext) =
    override this.Finalize () =
     match subr.disp with null -> () | disp -> post ctxt disp.Dispose

  let rec guard g xs = xs |>>* function Cons (x, xs) -> Cons (x, guard g xs)
                                      | Nil -> GC.SuppressFinalize g; Nil

  let ofObservableOn ctxt (xO: IObservable<'x>) : Stream<'x> =
    let xs = IVar<_> ()
    let sub = Subscriber (xs)
    post ctxt <| fun () -> sub.disp <- xO.Subscribe sub
    guard (Guard<_> (sub, ctxt)) xs

  let ofObservable xO = ofObservableOn null xO

  let toObservable xs =
    // XXX Use a better approach than naive locking.
    let subs = HashSet<IObserver<_>>()
    let inline iter f =
      Array.iter f << lock subs <| fun () ->
      let xs = Array.zeroCreate subs.Count
      subs.CopyTo xs
      xs
    let rec loop xs =<commit_msg>Must maintain state for correctness.<commit_after>﻿// Copyright (C) by <NAME>

namespace Hopac

open System
open System.Collections.Generic
open System.Threading
open Hopac.Infixes
open Hopac.Job.Infixes
open Hopac.Alt.Infixes
open Hopac.Promise.Infixes
open Hopac.Extensions
open Timer.Global

module Stream =
  let inline (<|>*) x y = x <|>* y :> Alt<_>
  let inline (|>>*) x y = x |>>* y :> Alt<_>
  let inline memo x = Promise.Now.delay x :> Alt<_>
  let inline start x = Job.Global.start x
  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 Cons<'x> =
    | Cons of Value: 'x * Next: Alt<Cons<'x>>
    | Nil

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

  type Src<'x> = {mutable src: IVar<Cons<'x>>}

  module Src =
    let create () = {src = IVar<_> ()}
    let rec value s x = Job.delay <| fun () ->
      let w = IVar<_> ()
      let v = s.src
      if IVar.Now.isFull v then raise <| Exception ("Src closed")
      let v' = Interlocked.CompareExchange (&s.src, w, v)
      if LanguagePrimitives.PhysicalEquality v' v
      then v <-= Cons (x, w)
      else value s x
    let error s e = Job.delay <| fun () -> s.src <-=! e // delay required
    let close s = Job.delay <| fun () -> s.src <-= Nil // delay required
    let tap s = s.src :> Alt<_>

  type Var<'x> = {mutable var: Cons<'x>}

  module Var =
    let create x = {var = Cons (x, IVar<_> ())}
    let get v = match v.var with Cons (x, _) -> x | Nil -> failwith "Impossible"
    let set v x = Job.delay <| fun () ->
      let c = Cons (x, IVar<_> ())
      match Interlocked.Exchange (&v.var, c) with
       | Cons (_, i) -> (i :?> IVar<_>) <-= c
       | Nil -> failwith "Impossible"
    let tap v = Alt.always v.var

  let inline nil<'x> = Alt.always Nil :> Stream<'x>
  let cons x xs = Alt.always (Cons (x, xs))
  let inline error e = Alt.raises e :> Stream<_>
  let inline one x = cons x nil
  let inline delay (u2xs: unit -> #Stream<'x>) =
    memo << Job.delay <| fun () -> u2xs ()

  let fix (xs2xs: Stream<'x> -> #Stream<'x>) =
    let xs = Promise<_> () // XXX publish interface for this?
    xs.Readers <- Promise<_>.Fulfill (Job.delay <| fun () -> xs2xs xs)
    xs :> Stream<_>

  let inline never<'x> = Alt.never () :> Stream<'x>

  let rec ofEnum (xs: IEnumerator<'x>) = memo << Job.thunk <| fun () ->
    if xs.MoveNext () then Cons (xs.Current, ofEnum xs) else xs.Dispose () ; Nil

  let ofSeq (xs: seq<_>) = delay (ofEnum << xs.GetEnumerator)

  let rec onCloseJob (uJ: Job<unit>) xs =
    Job.tryIn xs
       <| function Cons (x, xs) -> cons x (onCloseJob uJ xs) :> Job<_>
                 | Nil -> uJ >>% Nil
       <| fun e -> uJ >>! e
    |> memo
  let onCloseFun u2u xs = onCloseJob (Job.thunk u2u) xs

  type Subscribe<'x> (src: IVar<Cons<'x>>) =
    let mutable src = src
    interface IObserver<'x> with
     override t.OnCompleted () = src <-= Nil |> start
     override t.OnError (e) = src <-=! e |> start
     override t.OnNext (x) =
       let nxt = IVar<_> ()
       src <-= Cons (x, nxt) |> start
       src <- nxt

  let subscribeDuring (xs2ys: Stream<'x> -> #Stream<'y>) (xs: IObservable<'x>) =
    delay <| fun () ->
    let src = IVar<_> ()
    xs2ys src |> onCloseFun (xs.Subscribe (Subscribe (src))).Dispose

  let subscribeOnFirst (xs: IObservable<'x>) = delay <| fun () ->
    let src = IVar<_> () in xs.Subscribe (Subscribe (src)) |> ignore ; src

  let subscribingTo (xs: IObservable<'x>) (xs2yJ: Stream<'x> -> #Job<'y>) =
    Job.delay <| fun () ->
    let src = IVar<_> ()
    Job.using (xs.Subscribe (Subscribe (src))) <| fun _ -> xs2yJ src :> Job<_>

  let inline post (ctxt: SynchronizationContext) op =
    match ctxt with null -> op () | ctxt -> ctxt.Post ((fun _ -> op ()), null)

  type Subscriber<'x> (src: IVar<Cons<'x>>, ctxt: SynchronizationContext) =
    let mutable src = src
    // Initial = 0, Subscribed = 1, Disposed = 2
    [<DefaultValue>] val mutable State: int
    [<DefaultValue>] val mutable disp: IDisposable
    member this.Dispose () =
      if 2 <> this.State then
        this.State <- 2
        match this.disp with
         | null -> ()
         | disp -> post ctxt disp.Dispose
    member this.Subscribe (xO: IObservable<'x>) =
      if 0 = this.State then
        post ctxt <| fun () ->
          this.disp <- xO.Subscribe this
          if 0 <> Interlocked.CompareExchange (&this.State, 1, 0) then
            this.disp.Dispose ()
    interface IObserver<'x> with
     override t.OnCompleted () = t.State <- 2; src <-= Nil |> start
     override t.OnError (e) = t.State <- 2; src <-=! e |> start
     override t.OnNext (x) =
       let nxt = IVar<_> ()
       src <-= Cons (x, nxt) |> start
       src <- nxt

  type Guard<'x> (subr: Subscriber<'x>) =
    override this.Finalize () = subr.Dispose ()

  let rec guard g xs = xs |>>* function Cons (x, xs) -> Cons (x, guard g xs)
                                      | Nil -> GC.SuppressFinalize g; Nil

  let ofObservableOn ctxt (xO: IObservable<'x>) : Stream<'x> =
    let xs = IVar ()
    let sub = Subscriber (xs, ctxt)
    post ctxt <| fun () -> sub.Subscribe xO
    guard (Guard (sub)) xs

  let ofObservable xO = ofObservableOn null xO

  let toObservable xs =
    // XXX Use a better approach than naive locking.
    let subs = HashSet<IObserver<_>>()
    let inline iter f =
      Array.iter f << lock subs <| fun () ->
      let xs = Array.zeroCreate subs.Count
      subs.CopyTo xs
      xs
    let rec loop xs =
{'fraction_non_alphanumeric': 0.15942298743601674, 'fraction_numerical': 0.003257328990228013, 'mean_word_length': 2.4135959339263025, 'pattern_counts': {'":': 0, '<': 227, '<?xml version=': 0, '>': 272, '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': '1416067', 'n_tokens_mistral': 4030, 'n_tokens_neox': 3835, 'n_words': 1257}