I am currently working on the `Functor instance Qu...

# arrow-contributors

[kioba]

I am currently working on the Functor instance Queue ticket. I know this should be a fairly simple task sorry if I ask something obbvious :S ,
I have no previous experience with the Queue solution so I looked into the mentioned ZIO Queue (to be precise the trait ZQueue[-RA, +EA, -RB, +EB, -A, +B]) implementation.
If I understand correctly we currently dont have a ZQueue equivalent implementation in arrow, our queue is representing a general zio.internal.MutableConcurrentQueue.
To be able to implement the Functor we have to make sure the Queue captures the previous queue reference and the f function to be able to deplete the original queue.

TL;DR: we have to create a ZQueue equivalent in arrow and implement Functor on it instead of Queue.

I believe it is a slightly more work than a simple typeclass. I would be super happy to implement it, but before that I just want to make sure I am on the right path here 🙂 @simon.vergauwen

[kioba]

the mentioned issues is: https://github.com/arrow-kt/arrow/issues/1827

[kioba]

zio solution is just this:

/*
   * Transforms elements dequeued from this queue with a function.
   */
  def map[C](f: B => C): ZQueue[RA, EA, RB, EB, A, C] =
    new ZQueue[RA, EA, RB, EB, A, C] {
      def take: ZIO[RB, EB, C] = 
            self.take.map(f)
      def takeAll: ZIO[RB, EB, List[C]] = 
            self.takeAll.map(_.map(f))
      def takeUpTo(max: Int): ZIO[RB, EB, List[C]] =
           self.takeUpTo(max).map(_.map(f))
    }

[simon.vergauwen]

Hey @kioba!
Awesome that you want to contribute to this!

[simon.vergauwen]

Ah I think I see the problem you’re having. If you look at Ref, Semaphore, Promise, or MVar you will see they’re all defined as `interface`s that define the contract.

If you have a constraint of fun CF(): Concurrent<F> in your interface you can implement the above method as following.

fun <B> map(f: (A) -> B): Queue<F, B> = CF().run {
  object : Queue<F, B> {
    override fun take(): Kind<F, B> =
      [e-mail]().map(f)

     ...
  }
}

[simon.vergauwen]

FS2 has a very nice polymoprhic Queue impl as well.
https://github.com/functional-streams-for-scala/fs2/blob/master/core/shared/src/main/scala/fs2/concurrent/Queue.scala

[simon.vergauwen]

I’m be happy to help with any questions you have along the way! 👍

[kioba]

🤔 that is a fair point, the concern I have is:

queue.map { it.toString() }.shutdown()

will not shut down the original queue.

Unless we override all the functions to delegate the call to the original or

object : Queue<F, B> by this@map

[kioba]

not sure if delegating by this@map works tho 😅
What I was thinking maybe to introduce a Queue interface to make sure we never skip any important function?

[kioba]

will look into FS2! Thanks for the link

[simon.vergauwen]

Ah I think I see the problem you’re having. If you look at Ref, Semaphore, Promise, or MVar you will see they’re all defined as `interface`s that define the contract.

Yes, you need to split Queue in interface and implementations as done int he other concurrency primitives. They probably also deserve a Functor, FunctorK etc.

[simon.vergauwen]

will not shut down the original queue.

Should not be possible. It’s not a new Queue, we just wrap the Queue with Functor#map

[kioba]

@simon.vergauwen I looked into the FS2 Queue but they only define imap only 🤔 . The reason for that is we need to offer any element to the original queue.
ZQueue solves this with extending the type parameters so the offer only takes the original types but take can return any mapped type.

my current implementation is based on FS2:

override fun <A, B> QueueOf<F, A>.imap(f: (A) -> B, g: (B) -> A): Queue<F, B> =
    object : Queue<F, B> {
      override fun offer(a: B): Kind<F, Unit> = [e-mail]().offer(g(a))
      override fun take(): Kind<F, B> = with(functorF()) {
        [e-mail]().take().map(f)
      }
      // ....
  }

Without knowing the g: (B) -> A, we can not offer for the original Queue.
Should we extend the type arguments? what would be your suggestion?

[kioba]

extending the type arguments would allow us to provide the map functions:

interface QueueFunctor<F, A> : Functor<QueuePartialOf<F, A>> {
  fun functorF(): Functor<F>
  override fun <B, C> QueueOf<F, A, B>.map(f: (B) -> C): Queue<F, A, C> =
    object : Queue<F, A, C> {
      override fun offer(a: A): Kind<F, Unit> = [e-mail]().offer(a)
      override fun take(): Kind<F, C> = with(functorF()) {
        [e-mail]().take().map(f)
      }
    }
}

but I am not sure how could I bridge the gap between the FS2 and Zio 🤔

[kioba]

created a PR for the second solution: https://github.com/arrow-kt/arrow/pull/1904/files