feat: STTP stream #880
feat: STTP stream #880
Conversation
|
💵 To receive payouts, sign up on Algora, link your Github account and connect with Stripe. |
|
This is looking promising! I tried implementing a Flow Subscriber some time ago but wasn't successful. Thank you for working on it |
| IO.Unsafe.apply { | ||
| val queue = Queue.Unsafe.init[A](capacity = bufferSize, access = Access.MultiProducerMultiConsumer) | ||
| val onSubscribePromise = Fiber.Promise.Unsafe.init[Nothing, (Subscription, Queue[A])]() | ||
| val kyoSubscriber = new KyoSubscriber[A]: |
There was a problem hiding this comment.
The mix of safe and unsafe code can be tricky as the codebase evolves. It'd be nice to refactor this to a separate class without the AllowUnsafe evidence from IO.Unsafe in scope. Otherwise there's a higher risk of accidentally omitting IO suspensions.
| await | ||
| } { isEmpty => | ||
| if isEmpty then | ||
| _isDone.fold { |
There was a problem hiding this comment.
Do you mind using pattern matching with Absent/Present instead of fold? It's the pattern across the codebase
| if isEmpty then | ||
| _isDone.fold { | ||
| // queue is empty, we parks until notified | ||
| awaitPromise.safe.useResult { |
There was a problem hiding this comment.
Isn't this the same as awaitPromise.safe.get?
|
I wonder if the new |
|
@HollandDM can we close this for now? I wanted to avoid having PRs open for long |
|
Yeah, lets close it for now, I’ll pick it up later |
|
💵 To receive payouts, sign up on Algora, link your Github account and connect with Stripe. |
|
@HollandDM I'm working on a project that requires streaming. Do you have plans to work on this? I could also collaborate on the implementation to finalize it if you like |
|
@fwbrasil I’m quite busy these days, so I’m not coming back to this for a while. You can pick up the work if you want |
/claim #764
Will be updated when #896 merged
"Description updating..."