let single: IO<Never, [Void]> = effects.traverse { effect in
let action = IO<Never, Action>.var()
return binding(
continueOn(.global(qos: .userInitiated)),
action <- effect,
continueOn(.main),
|<-IO.invoke { self.send(action.get) },
yield: ())
}^
single.unsafeRunAsync(on: .global(qos: .userInitiated)) { _ in }traverse will get you a single IO describing the effects of your array of IO, and collecting all results in an array. You can also use parTraverse if you'd like the execution of the effects to be run in parallel.
send could be something like:public func send(_ action: Action) {
let effects = self.reducer(&self.value, action)
let single: UIO<[Void]> = effects.traverse { effect in
let action = UIO<Action>.var()
return binding(
continueOn(.global(qos: .userInitiated)),
action <- effect,
continueOn(.main),
|<-UIO.invoke { self.send(action.get) },
yield: ())
}^
single.unsafeRunAsync(on: .global(qos: .userInitiated)) { _ in }
}
oh great, I will try that, thank you for your help 🙂.
an other question on the API, wouldn't it be great to have this possibility :
effect.execute(on: .global(qos: .background)) // execute the IO on the background thread
.unsafeRunAsync(on: .main, // run the callback execution on the main thread
{ resultAction in
print(resultAction)
self.send(resultAction.rightValue) //Should never error out
})I find it hard to resort to Monad comprehension for a simple case like this. And the empty unsafeRunAsync is bothering me :). But there is maybe something like that already and I may have missed it.
- Once you call
unsafeRunSyncorunsafeRunAsync, you shouldn't make any other operation. We could add to the API a default empty closure, so that you don't have to pass it yourself. - You don't need to resort to monad comprehensions to switch the queue. IO also has a method
continueOnthat you can invoke to switch the queue of whatever comes next. In any case, you can still do what you are suggesting by callingattempt(yourexecutein the example above) and then theunsafeRun. - The problem was that you were wrapping everything on a
DispatchQueue.main.asyncand then launching theunsafeRunAsyncalso in the main queue, and causing a deadlock.
I added the DispatchQueue.main.async just before pushing the code, because it was working as I wanted to with it. I didn't have that when I was asking initially.
Thank you for your help on understanding that, it is really nice of you. So don't hesitate to tell me if it's bothering you, I don't want to take too much of your time. :)
so by doing something like that :
public func send(_ action: Action) {
print("send method called on \(DispatchQueue.currentLabel)")
let effects = self.reducer(&self.value, action)
print(effects)
effects.forEach { effect in
effect.attempt(on: .global(qos: .userInitiated))
.unsafeRunAsync(on: .main,
{ resultAction in
print("Resulting action : \(resultAction) on \(DispatchQueue.currentLabel)")
self.send(resultAction.rightValue) //Should never error out
})
}
}(I don't have the Traverse, but I want to understand with my initial code 🙂 )
I should have the same result as with your propose solution ? But it's not the case, I still have the deadlock . So what would be the right way to do it this way ?
Ok I think I put the finger on what is bothering me, the unsafeRunAsync run the whole computation of the IO, not just the ending callback on the thread that we pass, meaning that it will wait (because it wrap a unsafeRunSync call) for my IO computation to be executed before executing the callback on the given thread
Am I right with that ?
I assumed the IO would be executed asynchronously on any previously given Queue, then the callback would be called when the computation finished and this same callback would be called on the queue given through the .unsafeRunAsync method
extension IO {
public func attemptAsync(on queue: DispatchQueue = .main) -> IO<E, A> {
queue.shift().followedBy(self)^
}
public func myUnsafeRunAsync(on queue: DispatchQueue = .main, _ callback: @escaping Callback<E, A>) {
self.unsafeRunAsync(on: .global(qos: .background)) { (result) in // I don't know on which queue it should run by default, it's only to show what I was assuming
queue.async {
callback(result)
}
}
}
}
// using this extension I can now define the send method like that :
public func send(_ action: Action) {
print("send method called on \(DispatchQueue.currentLabel)")
let effects = self.reducer(&self.value, action)
print(effects)
effects.forEach { effect in
effect
.attemptAsync(on: .global(qos: .userInitiated))
.myUnsafeRunAsync({ resultAction in
print("Resulting action : \(resultAction) on \(DispatchQueue.currentLabel)")
self.send(resultAction.rightValue) //Should never error out
})
}
}.unsafeRunAsync as the finality of my IO, where I collect my data kind of like the .sinkin combine. but maybe it's not the right way of seeing things and I should see the final computation (getting the action back and feeding it to the .sendmethod) as part of my whole IO computation (the solution you gave earlier)
Hi,
I have a question regarding monad comprehension. Is it possible to run an IO using binding without waiting for the result of the execution of the IO?
My use case is, I want to trigger loading data from some source (API, DB,..). Once the data is loaded it will be made available using a Publisher. So I want to have a program that starts loading the data and then immediately returns without waiting for the loading to complete.
import Foundation
import Bow
import BowEffects
import Combine
func myProgram() -> IO<Never, ViewModel> {
let publisher = CurrentValueSubject<Data, Never>(Data())
return binding(
// is it possible to run this IO without waiting for it to return?
|<-loadData(publisher: publisher),
yield: ViewModel(data: publisher.eraseToAnyPublisher())
)^
}
func loadData(publisher: CurrentValueSubject<Data, Never>) -> UIO<Void> {
UIO.invoke {
// asume this takes a couple of seconds
let data = Api.getMyData()
publisher.send(data)
}
}
// stubs
struct Data {}
struct Api {
static func getMyData() -> Data {
Data()
}
}
struct ViewModel {
let data: AnyPublisher<Data, Never>
}binding, IO is not run; it is just a way of composing values, like you'd do using flatMap. IO will be run when you invoke any of the unsafeRun methods, so depending on what you choose, that operation will be synchronous or asynchronous. That is, if you invoke myProgram().unsafeRunAsync(on: queue) { ... } and pass the appropriate queue, it will return immediately while doing the operation on the queue that you passed. Be aware that, if you don't send a queue, the main one will be used, and it could lead to a deadlock.
I understand that
binding() is syntactic sugar for a bunch of flatMap()s and the resulting IO needs to be executed using one of the unsafeRun functions.So I guess, if I want to 'fire and forget' some side effect I just need to call
unsafeRunAsync() on it. And if I want to do it within a binding I need to wrap it in an IO. E.g. UIO.invoke { loadData.unsafeRunAsync() }.In Arrow there is syntactic sugar for this:
!effect { }. That's why I was under the impression the same thing could be achieved in Bow using less code.
handle: (Input) -> Void function to notify actions to Bow Arch whenever an event is triggered in the observer. If you have some sample code we can discuss around a concrete example.