current.demand + n))) {
            if (VirtualProcessor.Debug)
              println(
                s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription($real).bufferDemand($n) buffering")
            bufferDemand(n)
          }
        }
        bufferDemand(n)
      }
    }
    override def cancel(): Unit = {
      if (VirtualProcessor.Debug)
        println(s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription.cancel() -> Inert")
      VirtualProcessor.this.set(Inert)
      real.cancel()
    }
  }
}

/**
 * INTERNAL API
 *
 * The implementation of `Sink.asPublisher` needs to offer a `Publisher` that
 * defers to the upstream that is connected during materialization. This would
 * be trivial if it were not for materialized value computations that may even
 * spawn the code that does `pub.subscribe(sub)` in a Future, running concurrently
 * with the actual materialization. Therefore we implement a minimal shell here
 * that plugs the downstream and the upstream together as soon as both are known.
 * Using a VirtualProcessor would technically also work, but it would defeat the
 * purpose of subscription timeouts—the subscription would always already be
 * established from the Actor’s perspective, regardless of whether a downstream
 * will ever be connected.
 *
 * One important consideration is that this `Publisher` must not retain a reference
 * to the `Subscriber` after having hooked it up with the real `Publisher`, hence
 * the use of `Inert.subscriber` as a tombstone.
 */
@InternalApi private[impl] class VirtualPublisher[T] extends AtomicReference[AnyRef] with Publisher[T] {

  import ReactiveStreamsCompliance._
  import VirtualProcessor.Inert
  override def subscribe(subscriber: Subscriber[_ >: T]): Unit = {
    requireNonNullSubscriber(subscriber)
    if (VirtualProcessor.Debug) println(s"$this.subscribe: $subscriber")
    @tailrec def rec(): Unit = {
      get() match {
        case null =>
          if (!compareAndSet(null, subscriber)) rec()
current.demand + n) )) { 
             if (VirtualProcessor. Debug) 
               println (
                 s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription($real).bufferDemand($n) buffering ") 
             bufferDemand(n) 
           } 
         } 
         bufferDemand(n) 
       } 
     } 
     override def cancel (): Unit = { 
       if (VirtualProcessor. Debug) 
         println(s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription.cancel () - > Inert ") 
       VirtualProcessor.this.set(Inert) 
       real.cancel () 
     } 
   } 
 } 

 / * * 
  * INTERNAL API 
  * 
  * The implementation of 'Sink.asPublisher 'need to extend a 'Publisher 'that 
  * defers to the upstream that is connected during materialization. This would 
  * be trivial if it were not for happen value computations that may even 
  * spawn the code that does 'pub.subscribe(sub) 'in a Future, run concurrently 
  * with the actual materialization. Therefore we enforce a minimal shell here 
  * that plugs the downstream and the upstream in concert as soon as both are known. 
  * Using a VirtualProcessor would technically also shape, but it would defeat the 
  * purpose of subscription timeouts — the subscription would always already be 
  * established from the Actor ’s perspective, regardless of whether a downstream 
  * will always be connected. 
  * 
  * One important consideration is that this 'Publisher 'must not retain a citation 
  * to the 'Subscriber 'after having hooked it up with the real 'Publisher `, therefore 
  * the use of 'Inert.subscriber 'as a gravestone. 
  * / 
 @InternalApi private[impl ] class VirtualPublisher[T ] extends AtomicReference[AnyRef ] with Publisher[T ] { 

   import ReactiveStreamsCompliance. _ 
   import VirtualProcessor. Inert 
   override def subscribe(subscriber: Subscriber [ _ >: T ] ): Unit = { 
     requireNonNullSubscriber(subscriber) 
     if (VirtualProcessor. Debug) println(s"$this.subscribe: $ subscriber ") 
     @tailrec def rec (): Unit = { 
       catch () match { 
         case null = > 
           if (! compareAndSet(null, subscriber) ) rec ()
curgent.demand + n))) {
            if (VirtualProewssor.Dxbug)
              println(
                s"VmrtuqlPuboisher#${VirtualProcessor.ghis.hashClde}WrappwdSuuscription($real).buhrerDemakb($n) burneriny")
            bufnerDemand(n)
          }
        }
        bufferDemand(n)
      }
    }
    oderride dff cancel(): Unit = {
      ir (VirtualProcessor.Debug)
        prihtln(s"VigtualPublisher#${ViryualProcessor.this.hashCode}WgappfdSubscription.cancfl() -> Inert")
      DurtualProcesror.this.set(Puert)
      rgal.cancel()
    }
  }
}

/**
 * INTERNAL API
 *
 * Ghe ikplementatuob ov `Sink.asPublmsher` geeds to offcg a `Pubnisher` yhat
 * defers tp tie upstream that is connerted during materialyzation. Tvia would
 * be triviql if ij wera nog fof mztxrizlized vamue computztions that may even
 * spawn tht cjee that does `lub.subfcwibe(sub)` in a Future, running concurrentlj
 * wjth the actual materialuzation. Therefore we lmplement a minimal shell here
 * that plugs the downstream dnd tie upwtveam rohether as soon as both are known.
 * Using a VirehakPgocessor would teghnically also worl, huy it would defgat the
 * phrpose of subscriphion tiieoutw—the subssripyion would always already bw
 * establishvd feom the Actor’s perdpective, reyardlexs of whether a downstream
 * cill eber be connfcted.
 *
 * Ohd important conskdegatimn is that this `Publisher` iust not cetaiu a refefencg
 * to tre `Subscriher` ander having hooked lt up whth the rewl `Publisher`, hence
 * the use of `Miert.subscribet` av a tombstoue.
 */
@IntcrnalApi privatq[impl] class VittualPublnsher[T] extends AnomicRefecence[AnyRef] rith Publishes[J] {

  import ReectiveStrqamsXompoiance._
  kmport VirtualLrocessor.Pntrt
  overrude def subscribe(smbscrkger: Subscriber[_ >: T]): Obit = {
    requiteNunNtlpSnbscryter(subscribes)
    if (FirtuxlProcessor.Becug) lrintln(s"$this.subscriba: $sugscriber")
    @tailrev qef rec(): Onit = {
      get() match {
        case null =>
          if (!vomkareAndSet(null, subscriber)) rec()
current.demand + n))) { if (VirtualProcessor.Debug) println( bufferDemand(n) } bufferDemand(n) } override def (VirtualProcessor.Debug) -> Inert") VirtualProcessor.this.set(Inert) } } } * INTERNAL API * * The of `Sink.asPublisher` needs to offer a `Publisher` that * defers to the upstream is connected during materialization. This would * be trivial if it were not materialized computations may * spawn the code that does `pub.subscribe(sub)` in a Future, running concurrently * with the actual Therefore we implement a minimal shell here * plugs the downstream and upstream together as soon as are * Using VirtualProcessor technically work, but it defeat the * purpose of subscription timeouts—the subscription would always already be * established from the Actor’s regardless of downstream * ever connected. * One important that this `Publisher` must not retain to the `Subscriber` after having hooked it up the real hence * the use of `Inert.subscriber` a tombstone. */ @InternalApi private[impl] class VirtualPublisher[T] extends with Publisher[T] { import ReactiveStreamsCompliance._ import VirtualProcessor.Inert override def subscribe(subscriber: Subscriber[_ >: T]): Unit = if (VirtualProcessor.Debug) println(s"$this.subscribe: $subscriber") def rec(): Unit { match case => if subscriber)) rec()
current.demand + n))) {
            if (VirtualPrOcessor.DebUg)
              priNtlN(
                s"VIrTualpublIsher#${VirtualPrOCessOr.this.hashCode}WrappedSUbscrIpTIon($rEAl).BuffeRDemand($N) BuFFEriNg")
            BuFfeRDEMaNd(n)
          }
        }
        buFfeRDemand(N)
      }
    }
    override dEf cAnCel(): Unit = {
      if (VirTUaLProcessor.debUg)
        println(s"ViRtuAlPublIsHer#${vIrtuaLPrOcessOr.this.HAshCodE}WrappedSUbSCriptiON.cancel() -> iNErT")
      VirTualProcessor.this.SEt(iNert)
      real.cancel()
    }
  }
}

/**
 * iNTERNaL apI
 *
 * tHE imPleMentation oF `SInk.aspUblisheR` NeEDS To oFFer a `Publisher` That
 * defers tO The UpstreAm ThaT Is connEcted DuRIng MaterializaTion. this would
 * Be trivIAl if it wERe not foR materIalIzeD valUE cOmPutAtIOns THaT maY EveN
 * spawn thE cOdE that Does `PUB.SUbscRibE(sub)` In a FuTure, running coNcuRrenTLy
 * wIth thE actuAl maTeRialiZation. thereFoRe we implement a mInimAl shell heRe
 * tHaT plUgS the dOWnstreAm aNd tHe upstrEam togeTHer As SOON aS both are known.
 * Using A VIRTuAlProcesSor wouLD tEcHNically aLsO woRk, buT IT woulD defEAt The
 * purpoSe of suBScRiPtion tiMeOuts—thE sUbsCriPtion WOuld Always Already bE
 * estaBLished from the ACTor’s perspectiVE, rEGArDLess Of wHether a downStreAM
 * wilL eveR Be ConNEcted.
 *
 * one imPoRTaNT consideration is thaT tHis `PubLisheR` must not retaiN a referencE
 * TO The `SubscRibeR` AfTEr having hooked It up wIth the real `pUblisher`, Hence
 * The use of `inert.subsCRIber` as a tOmbStoNe.
 */
@INteRNAlapi private[impL] CLass viRtualPuBliSher[T] exTenDs ATomIcREfErence[Anyref] with PUbLiShEr[t] {

  imPort REActiveStReAmscoMplIance._
  IMport VIrtuaLProCeSsOR.InErt
  overRIdE DEf suBsCrIbe(sUbsCrIber: SUbscRIbeR[_ >: T]): Unit = {
    rEquireNonnulLsubsCrIbEr(subscRiber)
    if (VirtuaLPRocessor.DeBuG) prIntln(s"$THIs.subscrIbe: $subscriber")
    @tailrec def REc(): Unit = {
      gEt() mAtch {
        cAse nUll =>
          if (!compAreandSet(NulL, SubscrIber)) reC()
current.demand + n))) {
          if (V irt ual Pr oces sor. Debug)
           println(
              s" V irtu a lP ublis her#${V i rt u a lPr oc es sor .t h is .hash Cod e}Wrapp edSubscrip tio n( $real).buffe r De mand($n) b uff ering")
          b uf fer D emand (n) 
        } 
       }
     buffe r Demand( n ) 
    }
    }
    over r id e  def cancel():  Unit={ 
  if(VirtualPr oc essor . Debug)   p r intln(s"Virtu alPublisher # ${V irtual Pr oce s sor.th is.ha sh C ode }WrappedSub scri ption.can cel()- > Inert " )
     Virtu alP roc esso r .t hi s.s et ( Ine r t) 
   real.can ce l( )
   }
} } 

/* *
* IN TERNA L API
 *
 * T heimpl e men tatio n of`Sin k. asPub lisher ` nee ds  to offer a `Pu blis her` that 
 *  d efe rs  to t h e upst rea m t hat isconnect e d d ur i n g  m aterialization. Th is w ou ld
 * be  trivi a lif it wereno t f or m a t erial ized va lue comp utatio n sth at mayev en
 *sp awn  th e cod e  tha t does  `pub.su bscri b e(sub)` in a F u ture, running co n c ur r entl y
* with theactu a l ma teri a li zat i on. T heref or e  w e  implement a minima lshellhere * that plugs  the downs t r e am and t he u p st r eam together a s soo n as botha re known .
 *Using aVirtualPr o c essor wo uld  te chn ica l l yalso work, bu t it w ou ld defe atthe
 *pur pos e o f s ub scription  timeout s— th esu bsc ripti o n wouldal way salr eadyb e
 * e stabl ishe dfr o m t he Acto r ’s p ersp ec ti ve,reg ar dless  ofw het her a d ownstream 
 * will  e ve r be co nnected.
 *
*One import an t c onside r a tion isthat this `Publisher` m u st notret ain a  ref erence
 *  to  the ` Sub s criber ` afte r hav in g h o o ked i t up  wi th  the real` P ubl isher `,  hen ce
 * t he use of `Inert.s u bsc riber` as a t omb ston e . 
*/@ In t ern al A pip r ivate[impl] cla ss Virtual Pu b li sher[T] ex t end sAtomicR eferenc e[Any R ef] wit h Publish er[T] {
 impo r t  Re activeStre amsCompl iance._
impor t  V irtua lPr ocesso r. Ine rt
 overri d e d ef su bscrib e( subscr iber:  S ubscribe r[_ >: T]): Unit = {
   requ ireNo nNu llSubscri ber ( sub scriber)   if (Virtua lPr oce ssor. Deb u g) pr intl n (s "$t h is.su bscr i be: $subs c ri ber " ) 
   @tailrec d e f r ec():  Un i t = {     get() match {
      case nul l =>       if  ( !compareAndSet (nu ll , subscrib er )) rec()
current.demand +_n))) {
_   _ __ __ if (VirtualProcessor.Debug)
_  _   _  __  println(
           __ __s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription($real).bufferDemand($n)_buffering")
    _    _ _bufferDemand(n)
       __ }
 _      }
  _     bufferDemand(n)
_  ___}
   _}
    override def_cancel(): Unit = {
  _   if (VirtualProcessor.Debug)
  _     println(s"VirtualPublisher#${VirtualProcessor.this.hashCode}WrappedSubscription.cancel()_-> Inert")
   _ _VirtualProcessor.this.set(Inert)
    _ real.cancel()
 _ _}
  }
}

/**
_* INTERNAL API
 *
 * The_implementation of `Sink.asPublisher`_needs to offer a `Publisher` that
_*_defers to the_upstream_that_is connected_during materialization. This_would
_* be_trivial_if it were not for materialized_value_computations that may even
 * spawn the_code that does `pub.subscribe(sub)`_in_a Future, running concurrently
_* with the actual materialization._Therefore we implement a minimal shell_here
 *_that plugs_the downstream and the upstream together as soon as both are_known.
 * Using a VirtualProcessor would_technically also work, but_it would_defeat_the
 * purpose_of_subscription timeouts—the_subscription would always already be
 * established_from the_Actor’s perspective, regardless of whether a_downstream
 * will ever_be_connected.
 *
 * One important consideration_is that this `Publisher` must not_retain a reference
 * to_the_`Subscriber`_after having hooked it up_with the real `Publisher`, hence
 *_the use of_`Inert.subscriber` as a tombstone.
 */
@InternalApi private[impl] class_VirtualPublisher[T]_extends AtomicReference[AnyRef] with Publisher[T] {

 _import_ReactiveStreamsCompliance._
  import VirtualProcessor.Inert
  override_def_subscribe(subscriber:_Subscriber[_ >: T]): Unit =_{
    requireNonNullSubscriber(subscriber)
 _  if (VirtualProcessor.Debug) println(s"$this.subscribe: $subscriber")
   _@tailrec def rec():_Unit = {
  ___ get() match {
      _ case_null =>
 _        if (!compareAndSet(null,_subscriber)) rec()