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

namespace Hopac

open System
open System.Collections.Generic
open System.Threading
open Hopac.Infixes
open Hopac.Extensions
open Hopac.Timer.Global

module Stream =
  let imp () = failwith "Impossible"
  let inline memo x = Promise.Now.delay x
  let inline queue x = Job.Global.queue x
  let inline start x = Job.Global.start x
  let inline server x = Job.Global.server x
  let inline tryAp x2y x yK eK =
    let mutable e = null
    let y = try x2y x with e' -> e <- e' ; Unchecked.defaultof<_>
    match e with
     | null -> yK y
     | e -> eK e
  let inline tryIn u2v vK eK = tryAp u2v () vK eK
  let inline (|Nothing|Just|) (b, x) = if b then Just x else Nothing

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

  type Stream<'x> = Promise<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 :> Promise<_>

  let ValueChangedEventArgs =
    System.ComponentModel.PropertyChangedEventArgs "Value"

  type Property<'x> (x: 'x) =
    let mutable var = Cons (x, IVar ())
    let propertyChangedEvent = Event<_, _> ()
    interface System.ComponentModel.INotifyPropertyChanged with
      [<CLIEvent>]
      member this.PropertyChanged = propertyChangedEvent.Publish
    member this.Value
     with get () = match var with Cons (x, _) -> x | Nil -> imp ()
      and set x =
        let c = Cons (x, IVar ())
        match Interlocked.Exchange (&var, c) with
         | Cons (_, i) ->
           (i :?> IVar<_>) *<= c |> start
           propertyChangedEvent.Trigger (this, ValueChangedEventArgs)
         | Nil -> imp ()<commit_msg>Fix `Stream.Src.value` to work in contention #101.<commit_after>// Copyright (C) by <NAME>

namespace Hopac

open System
open System.Collections.Generic
open System.Threading
open Hopac.Infixes
open Hopac.Extensions
open Hopac.Timer.Global

module Stream =
  let imp () = failwith "Impossible"
  let inline memo x = Promise.Now.delay x
  let inline queue x = Job.Global.queue x
  let inline start x = Job.Global.start x
  let inline server x = Job.Global.server x
  let inline tryAp x2y x yK eK =
    let mutable e = null
    let y = try x2y x with e' -> e <- e' ; Unchecked.defaultof<_>
    match e with
     | null -> yK y
     | e -> eK e
  let inline tryIn u2v vK eK = tryAp u2v () vK eK
  let inline (|Nothing|Just|) (b, x) = if b then Just x else Nothing

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

  type Stream<'x> = Promise<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
        match IVar.Now.get v with
         | Nil -> raise <| Exception ("Src closed")
         | Cons _ -> value s x
      else
        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 :> Promise<_>

  let ValueChangedEventArgs =
    System.ComponentModel.PropertyChangedEventArgs "Value"

  type Property<'x> (x: 'x) =
    let mutable var = Cons (x, IVar ())
    let propertyChangedEvent = Event<_, _> ()
    interface System.ComponentModel.INotifyPropertyChanged with
      [<CLIEvent>]
      member this.PropertyChanged = propertyChangedEvent.Publish
    member this.Value
     with get () = match var with Cons (x, _) -> x | Nil -> imp ()
      and set x =
        let c = Cons (x, IVar ())
        match Interlocked.Exchange (&var, c) with
         | Cons (_, i) ->
           (i :?> IVar<_>) *<= c |> start
           propertyChangedEvent.Trigger (this, ValueChangedEventArgs)
         | Nil -> imp ()
{'fraction_non_alphanumeric': 0.1250282103362672, 'fraction_numerical': 0.002482509591514331, 'mean_word_length': 2.4329976762199843, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1416068', 'n_tokens_mistral': 1480, 'n_tokens_neox': 1461, 'n_words': 512}