typelevel/cats-effect

Most discussion is on Typelevel Discord: https://discord.gg/bSQBZA3Ced

Mark Tomko

@mtomko

great! thanks so much!

Fabio Labella

@SystemFw

basically with this approach you want to model this sort of problems as a finite state machine

for this scenario you have two cases: the first call, where there is no value initially, and the subsequent calls where there is a new value that needs updating

the first scenario requires some waiting, in case the first get call arrives before the variable can be populated

waiting means having a deferred

so overall the state looks like this

sealed trait State[A]
case class FirstCall(wait: Deferred[F, A]) extends State[A]
case class NextCalls(v: A) extends State[A]

normally I put this as an inner class in the create method of the abstraction (I'll show you that in a bit), so that's why there is no F[_] parameter, it's normally in the outer definition

Mark Tomko

@mtomko

I'm with you so far

Fabio Labella

@SystemFw

then you want to write the transitions

well, I should specify that you have a Ref[F, State[A]] as your main component

Christopher Davenport

@ChristopherDavenport

What is first call parameterized?

That looks to lose the F

Deferred holds some F

Fabio Labella

@SystemFw

you need an initial value for that, which looks like

for {
  d <- Deferred[F, A]
  r <- Ref[F].of(FirstCall(d): State)
} yield r

@ChristopherDavenport do you mean

, so that's why there is no F[_] parameter, it's normally in the outer definition

let me just write the whole thing down, then we can talk about it, it's clearer

be back in 5

Mark Tomko

@mtomko

thanks!

Fabio Labella

@SystemFw

ah btw, there is one hard bit you need to think about in general, which is what to do when update fails. I'm going to write a buggy version that does not account for that, then we can discuss

Mark Tomko

@mtomko

oh, right, good point!

Mark Tomko

@mtomko

I guess depending on how we use State[A] we could add a case class Unavailable(reason: Throwable) extends State[Nothing] and pass the error on to the consumer.

Fabio Labella

@SystemFw

trait Thing[F[_], A] {
  def get: F[A]
}
object Thing {
  def create[F[_]: Concurrent: Timer, A](
      action: F[A],
      t: FiniteDuration
  ): Resource[F, Thing[F, A]] = {
    sealed trait State
    case class FirstCall(waitV: Deferred[F, A]) extends State
    case class NextCalls(v: A) extends State

    val initial = for {
      d <- Deferred[F, A]
      r <- Ref[F].of(FirstCall(d): State)
    } yield r

    Stream
      .eval(initial)
      .flatMap { state =>
        val read = new Thing[F, A] {
          def get = state.get.flatMap {
            case FirstCall(wait) => wait.get
            case NextCalls(v) => v.pure[F]
          }
        }

        val write = action.flatMap { v =>
          state.modify {
            case FirstCall(wait) => NextCalls(v) -> wait.complete(v).void
            case NextCalls(_) => NextCalls(v) -> ().pure[F]
          }.flatten
        }

        Stream.emit(read).concurrently(Stream.repeatEval(write).metered(t))
      }
      .compile
      .resource
      .lastOrError

  }
}

something along these lines

only checked for compilation so might be buggy :P

also needs to deal with errors and interruption, but shouldn't be too hard depending on the semantics you want

Mark Tomko

@mtomko

okay, reading over it now!

Fabio Labella

@SystemFw

your comment about how to handle errors make sense btw, I would recommend that as the easiest approach (you need an Either in the Deferred too, in case the error happens in the first call)

for subsequent calls you have the choice of propagating the error to the consumer or leaving the old value there

Mark Tomko

@mtomko

right, both of those are fine options. I think in my case, propagating it to the caller might be right, but I could see leaving the old one in place and logging vehemently

also I'd presumably add some retry logic

oh, this is really interesting. I think I see what's going on and I wouldn't have thought of this.

Fabio Labella

@SystemFw

it's a technique that I find invaluable once complexity grows

since you can focus on deriving the transition function

and you transform, as far as possible, a concurrency problem into a data modelling problem (which is far easier to solve)

Mark Tomko

@mtomko

yeah. I think I'd been thinking the Thing class would contain an MVar[F, A] and it's get would translate to a read on the mvar.

Fabio Labella

@SystemFw

interruption safety is the biggest complication in general, but I hope my proposal for interruption in CE3 will address that

Mark Tomko

@mtomko

But here, everything's done in one place.

Fabio Labella

@SystemFw

let me edit a small but critical bug :)

done

Mark Tomko

@mtomko

thanks!

Fabio Labella

@SystemFw

:+1:

Mark Tomko

@mtomko

I am pretty sure I see what all is going on, thank you so much! I think I see how to add the error handling (as discussed) and I'll spend some time wrapping my heard around this. I have to step away, but I really appreciate your generous help. Be well, and thanks again!

Fabio Labella

@SystemFw

:) :wave:

Christopher Davenport

@ChristopherDavenport

How does the compiler know that F is correct outside that limited scope, I'm still missing an alignment.

Oh I see it's inside the F declaration.

Mark Tomko

@mtomko

Hmm. For F[_] that have MonadError[T, A] you can call .attempt to go from F[A] to F[Either[T, A]]. Is there an inverse operation that takes an F[Either[T, A]] and converts it to an F[A], raising T in the Left case?

maybe this belongs in cats instead.

Daniel Spiewak

@djspiewak

.rethrow

Mark Tomko

@mtomko

sweet, thanks!

that's better than what I was going to call it, unattempt

Where communities thrive

People

Repo info

Activity