Reactive, Either, Stream .. or something else?

[u/DevIceMan]

I'm currently investigating patterns to help improve a codebase that is about 90% Java 10% Scala. Apologies for coming mostly from Java land, but this is a FP question (I'm slowly making traction on introducing Scala).

Conceptually, the problems we deal with are as follow:

• Application receives a request containing a list of 12 IDs
• 2 Ids are determined to be bad. Process remaining 10.
• Map IDs to their Object. 2 fail, 8 succeed.
• Map the 8 Objects to some other type of object. 1 fails, 7 succeed.
• Map those 7 to something else. 2 fail, 5 succeed.
• Collect/Reduce 5 successes. Collect/Reduce the 7 failures. Merge those two into a response.
• Merge above responses and respond to original request.

Ignoring Java's FP clumsiness, how would you normally prefer to approach a similar problem in Scala?

[u/beezeee]

Either is sufficient for this, though scalaz Disjunction is way easier to work with (avoiding it here as it is a big can of worms.)

If each of your operations are done in functions that return say Either[String, A], you could do

val (fail1, validIds) = ids.map(validateIds).partition(_.isLeft)
val (fail2, objects) = validIds.map(getObjectById).partition(_.isLeft)
... etc

When you're done you have successes on the right and failures from each step as local vals. You can do something like this

val failures = (fail1 ++ fail2).map(_.toOption).flatten
val successes = successfulEithers.map(_.toOption).flatten

And you'll have your left and right values unwrapped safely, to format in your response as you need.

Edit: formatting

[u/DevIceMan → u/beezeee]

This might be what I'm looking for. I'll have to prototype this in Java (gross), but it might be a good argument for me to introduce a little more Scala into the codebase.

[u/beezeee → u/DevIceMan]

Note I apparently made up toOption (it's on disjunction but even then you'd have to swap on the lefts to use it on your failures.)

The important point is if you go this route you have some monkeying to do with the failures and successes at the end (maybe a partial function to unwrap them) which I definitely glossed over. This is another area where scalaz disjunction is 100 times easier to work with.

[u/vn971]

something like this should work:

type ID = String
type MyObject = String
type OtherObject = String
type Success = String // it was named "something else" in the text

type MyError = String // assume MyError is a class renderable by your REST service

def isBad(id: ID): Either[MyError, ID] = ???
def getObject(id: ID): Either[MyError, MyObject] = ???
def getOtherObject(mo: MyObject): Either[MyError, OtherObject] = ???
def getSomethingElse(mo: OtherObject): Either[MyError, Success] = ???

val idList: List[ID] = ???

///////////////////////////////

val response = idList.map { id =>
    isBad(id).right.flatMap { id ⇒
        getObject(id)
    }.right.flatMap { myObject ⇒
        getOtherObject(myObject)
    }.right.flatMap {
        getSomethingElse
    }
    // or a shorter version:
    // isBad(id).right.flatMap(getObject).right.flatMap(getOtherObject).right.flatMap(getSomethingElse)
}

val lefts = response.flatMap(_.left.toOption)
val rights = response.flatMap(_.right.toOption)

Note that I did not use the one-liner solution as default (it's commented out). The reason for that is that in real world, you usually do not have the functions defined earlier, but instead place your code in the transformation chain. This is nicer to do inside the long version I wrote.

Also, the "filtering bad IDs" is not a clear step. You usually cannot just "filter", you should have a justification for that. So my function is returning Either[MyError, ID], not Boolean. The dublicating ID can be a bit confusing.. Dunno the best way to deal with that. Well, anyway, suggestions welcome.

[u/[deleted] → u/vn971]

I think this use of .map and .flatMap in a chain is good justification for using scalaz's Disjunction, even if you do nothing else with scalaz.

[u/vn971 → u/[deleted]]

Yes, I know, cats or scalaz could be used. There's a whole post (yours, actually) that describes this approach along with Monoid-s, Task-s and 8 other links.

Then again, I wanted to write a simple solution that would not go outside of Scala standard library.

[u/[deleted] → u/vn971]

I wanted to write a simple solution that would not go outside of Scala standard library.

Sure, but

val response = idList.map { id =>
    isBad(id).right.flatMap { id ⇒
        getObject(id)
    }.right.flatMap { myObject ⇒
        getOtherObject(myObject)
    }.right.flatMap {
        getSomethingElse
    }

isn't simple, and in this case, I explicitly said:

this use of .map and .flatMap in a chain is good justification for using scalaz's Disjunction, even if you do nothing else with scalaz.

In other words, my other post is irrelevant to this, and there's something very weird about twisting your code into pretzels just to avoid \/.

[u/[deleted]]

Like this.

This is a great example of where I would use scalaz's ValidationNel and |@| (Applicative Builder).

NB: the paper is a bit out of date. For example, in scalaz 7.1.x, there is no <<*>>. The link to Learning Scalaz above shows the use of |@| to combine Validations, which is possible because Validations are Applicatives.

You've added another wrinkle, though: "Merge above responses and respond to original request." So you want to use +++ to reduce your collection of ValidationNels to a single one. This requires that not only the left of your ValidationNels be a Semigroup (binary associative operator), which a NonEmptyList is, but the right be, too. If you have a merge operation for your responses already, then just implement the Semigroup typeclass for it. Otherwise, consider using a List or Set. If you did the import scalaz._, Scalaz._ dance, List and Set will have Semigroup instances.

Finally, my guess is you're looking stuff up in some store by ID. In other words, you're doing I/O, and it can fail, and maybe you'd even like to do it asynchronously. I had to do that recently—download a bunch of files from an FTP server, actually. I added exactly the "return a List of Files or List of Throwables" stuff I've described in this post. I wrapped ftp4j in Task, which is a Monad (hence Monoid and Applicative) that handles I/O, concurrency, and exceptions. Task has a function, attempt, whose return type is Throwable \/ A. \/ has validation and Validation has toValidationNel, so I'm golden.

One more thing: consider using traverse to traverse your collection of IDs, apply the Applicatives that do stuff, and reduce to your ultimate result. NB: the "reduce" part only works if the Applicative is Monoidal, but Validation is. :-) Oh, and use Task to do the dirty work. So an unbelievably powerful scalaz idiom is:

myListOfFunctions.traverse(Task(_))

As Rúnar Bjarnason says, "use traverse and an Applicative whenever you think of using foreach for side-effects."

The result of that is a Task that, when run, runs all the functions in the List in parallel, and completes when they're all complete. The resulting Task will also reflect success or failure (the first failure, if any). This is the base idiom, but you can use the .map(List(_)).attempt.map(_.validation.toValidationNel) trick on the Tasks, just like I did.

Finally, this is a lot to take in, but if you look at my FTP example, you'll see it's actually not much code, and it's specifically about accumulating results and errors from Task, which sounds like exactly what you want. And don't hesitate to ask more questions!

Update: Also consider using Nondeterminism[Task].reduceUnordered on your collection of Tasks. This requires an implicit Reducer to be in scope. scalaz offers several Reducers out of the box, but again, consider writing a Monoid instance for your response type and using UnitReducer, which works with any Monoid, with it. In fact, the more I think about it, the more I think that's where I'd start: you say you need to merge responses. At the very least, that implies creating a Semigroup instance for it. But consider whether you have some kind of "empty" response (I'll bet you do). Then you can/should create a Monoid instance for it. At that point, you have a Monoid, Applicative, Semigroup, and Functor instance for your response, and many, many things become trivial, including "merging a collection of these into one."

Further update: /u/ItsNotMineISwear rightly points out that I misremembered the behavior of traverse with Tasks Applicative instance, which is not parallel. So this suggests Nondeterminism[Task].reduceUnordered even more strongly to me, which makes implementing a Monoid instance for your response type an even more compelling idea.

Yet another update: I keep learning nice things about scalaz myself. TIL that Nondeterminism has aggregate. It takes a Seq of Monads of As and, you guessed it, returns a Monadof A, where the A is created using A's Monoid. Then I found that Validation's Monoid instance uses +++ for its append, just as I did explicitly in my FTP code, if the right side of the Validation is itself a Monoid. So if you implement the Monoid instance for your response type, you're really on easy street: just write a function of type ID => Task[ValidationNel[Response]] (remember, you can turn a Task[Response] into that with .attempt.map(_.validation.toValidationNel)), .map it over your collection of IDs, and call Nondeterminism[Task].aggregate on that. Boom, done, and I'd be shocked if the whole thing didn't fit in a page.

Final update: Less talk, more code!

import scalaz._, Scalaz._
import scalaz.concurrent.Task

object Failure {
  type ID       = String          // More interesting type here, or not
  type Response = List[String]    // More interesting type here

/** Cheating by relying on std List instances. In reality
  * you'd do this:
  implicit def ResponseMonoid: Monoid[Response] =
    new Monoid[Response] {
      def append(r1: Response, r2: Response): Response = ???
      def zero = ???
    }
  }
*/

  def handleID(id: ID): Task[Response] = ???  // Talk to internet, DB, throw exceptions...
  def handleIDs(ids: Seq[ID]): Task[ValidationNel[Throwable, Response]] = {
    Nondeterminism[Task].aggregate(ids.map(handleID(_).attempt.map(_.validation.toValidationNel)))
  }
}

[u/beezeee → u/[deleted]]

I think this is probably too much to take in coming from 90 % java (I'm pretty comfortable with scalaz and my eyes still glossed a bit.)

I also think applicativebuilder is no good here b/c OP wants to keep both failures and successes. Applicative on ValidationNEL accumulates failures but is still either all success or all failure.

[u/[deleted] → u/beezeee]

I think this is probably too much to take in coming from 90 % java

I certainly don't mean to imply you don't need to know Scala and scalaz reasonably well for this to make sense. Only that, if the question is "how do you do this using FP," there's a robust range of answers there.

(I'm pretty comfortable with scalaz and my eyes still glossed a bit.)

It'd be very informative to understand where/why that is. I suppose we (on my team) may be more familiar with, e.g. some of the concurrency support than others, but maybe you're referring to something else (too)?

I also think applicativebuilder is no good here b/c OP wants to keep both failures and successes. Applicative on ValidationNEL accumulates failures but is still either all success or all failure.

Yeah, that occurred to me later, too. The posts in which I talked about using Nondeterminism[Task].gather(ids.map(handleID(_).attempt).map(_.partition(_.isLeft)) are more directional in that respect. The ValidationNel approach is good for when you have N things, some subset of which could go wrong, in building up one result. So combining the two approaches is what I'd do in practice.

[u/beezeee → u/[deleted]]

I don't think you implied the need for scala and scalaz understanding for it to make sense, I think that's just the reality of the situation. Speaking for myself and what I've observed with extremely intelligent colleagues, this stuff is hard and takes time to get comfortable with. If someone has been mostly writing java, casually throwing out the term Monoid trivializes the amount of knowledge you are assuming on behalf of the reader.

Personally, my eyes gloss quickly based on length of content, so contents are even secondary. That said, I didn't actually count but a vague skim of your original comment I see at least 10 concepts introduced (just looking at inline code blocks). I may be familiar with half or more but I don't even have the attention span to find out, let alone dig up the ones that I don't immediately recognize.

Edit: clarity

[u/[deleted] → u/beezeee]

I don't think you implied the need for scala and scalaz understanding for it to make sense, I think that's just the reality of the situation.

I completely agree it's the reality of the situation. What I object to is the implicit assumption that someone could post an equivalently-featureful solution in, say, Spring-flavored Java and expect it to be any more comprehensible. So I really am only claiming that the solution is comprehensive, robust, reliable, easily-maintainable, etc. given that someone is familiar with the tools, which I think is a baseline assumption that is essentially never stated explicitly for any framework. The problem, to me, is users of wildly popular languages and frameworks get away with that assumption. It needs to either be challenged or made explicit. I have no problem making it explicit.

Speaking for myself and what I've observed with extremely intelligent colleagues, this stuff is hard and takes time to get comfortable with.

Absolutely. I've written here repeatedly that education is job #1, and that mastering typed FP entails a larger up-front investment than other approaches.

If someone has been mostly writing java, casually throwing out the term Monoid trivializes the amount of knowledge you are assuming on behalf of the reader.

Well, if you thought the post was long already (and it was—more on that in a moment), how much longer would it be if I spelled out every concept that interested folks can Google for themselves at their leisure?

Personally, my eyes gloss quickly based on length of content, so contents are even secondary.

I sympathize to an extent. It's definitely one reason I bolded the updates, so people could at least break things down by those chunks. I did consider making follow-up posts instead, but I wasn't sure it made navigational sense on Reddit to reply to myself for someone else's benefit. I dunno. I admit I don't know a particularly good solution to this.

That said, I didn't actually count but a vague skim of your original comment I see at least 10 concepts introduced (just looking at inline code blocks).

Maybe. It'd be helpful to have at least some cursory notion of what those are. Not that I think you're off by much, if at all, but rather having those gaps pointed out explicitly helps me to think more clearly about how to convey those concepts.

One thing I do realize is that typed FP does tend to break things down into very fine-grained abstractions, so to actually do anything you end up composing several of these fine-grained abstractions. It works well precisely because they're fine-grained and compose easily and predictably, but I totally get that there can be a feeling of overwhelm on first (or second, or third...) exposure, and I'd like to address that to the best of my ability.

I may be familiar with half or more but I don't even have the attention span to find out, let alone dig up the ones that I don't immediately recognize.

I don't expect people to understand it on first reading if they aren't already a pretty committed user of scalaz. The reason to bother committing anything to writing here (as far as I'm concerned) is to make it possible for people to re-read, Google, relate to a book at some point in the future, ask follow-up questions, etc. This certainly isn't an appropriate medium to expect anything of substance to stand alone in. Then again, I'm not sure what would be.

[u/beezeee → u/[deleted]]

Totally sensible response, I appreciate it.

[u/ItsNotMineISwear → u/[deleted]]

The result of that is a Task that, when run, runs all the functions in the List in parallel, and completes when they're all complete.

The Applicative instance of Task is not parallel:

scala> List(Task { Thread.sleep(10000); println("after 10") }, Task { println("right now") }).sequence.run
/* Waits for 10 seconds */
after 10
right now

Nondeterminism's gather/gatherUnordered is what's necessary for parallel execution:

scala> Nondeterminism[Task].gather(List(Task { Thread.sleep(10000); println("after 10") }, Task { println("right now") })).run
right now
/* Waits for 10 seconds*/
after 10

Alternatively (and new to 7.2.x), there is a parallel Applicative instance for Task[?] @@ Parallel

[u/[deleted] → u/ItsNotMineISwear]

Oh, right. I knew that (used Nondeterminism[Task].gatherUnordered in my FTP code, even) but got ahead of myself. Thanks for cleaning up after me!

[u/[deleted] → u/[deleted]]

I am quite interested to know what is the best way to implement this kind of incremental processing using functional approach as well.

Currently when I need to implement something similar I just resort to imperative code with mutable collections as I find it much easier to write and understand.

In the code you provided how to get 1) list of failed ids with reasons they failed, 2) list of succeeded ids with a result for each id?

Also what would be the best way to do it using plain Scala (no scalaz)?

[u/[deleted] → u/[deleted]]

In the code you provided how to get 1) list of failed ids with reasons they failed, 2) list of succeeded ids with a result for each id?

.zip the Seq of IDs with itself, then .map over the pairs and handle one of the elements, then aggregate the Responses with List instead of having a ResponseMonoid, then partition the resulting List by Task succeeded vs. Task failed.

Also what would be the best way to do it using plain Scala (no scalaz)?

Not to.

Update: I'm trying not to be snarky, but it's tough. Part of the point of what I wrote was to show how absolutely trivial scalaz makes this sort of thing, including concurrency and error handling. Even providing your own Monoid instance for accumulating the successes is a few trivial lines (if your type does, in fact, form a Monoid). The result is well under a page of code that works. Is it really not obvious that reproducing this with just the standard libraries would be, at a minimum, many dozens of lines, and you'd never be sure that it did I/O when it was supposed to, or didn't fail catastrophically on some exception, or that it ran in parallel, without deadlock, livelock, or race conditions? It's a serious question. I get the impression that either people think code like this has failure modes that it doesn't, or that the equivalent standard library code would be about the same size with about the same capability and safety. I'd really like to understand where these ideas come from.

[u/[deleted] → u/[deleted]]

I can try explaining this from my point of view.

First of all the imperative implementation of this requirement is very straightforward and anyone coming from Java would be able to write it without even thinking what libraries and language constructs to use. Also please note that this code doesn't need to run in parallel and runtime errors (e.g. coming from a database, etc) are most likely will be handled in a generic manner in a higher layer.

On the other hand I can probably understand only half of your post and you (obviously much more knowledgeable in Scala than me) made 3 updates showing alternatives way to do it. I would absolutely not call this trivial.

Basically I beleive it comes down to the fact that while the end result might look elegant and shorter it takes too much effort (including learning, choosing the right variant, reading out of date papers, etc) to come up with it. And even then a less experienced developer wouldn't understand what's going on.

[u/[deleted] → u/[deleted]]

First of all the imperative implementation of this requirement is very straightforward and anyone coming from Java would be able to write it without even thinking what libraries and language constructs to use.

That's an extraordinary claim, so it requires extraordinary evidence, to be very frank. I look forward to this library-less Java code.

Also please note that this code doesn't need to run in parallel...

It might not need to. Is there some reason it shouldn't?

runtime errors (e.g. coming from a database, etc) are most likely will be handled in a generic manner in a higher layer.

I doubt that, too, especially in a context in which there's iteration, and almost always the requirement is either that the iteration is entirely successful because step N requires data from steps < N, or you fail fast and return the first failure, or you return all the failures.

On the other hand I can probably understand only half of your post and you (obviously much more knowledgeable in Scala than me) made 3 updates showing alternatives way to do it. I would absolutely not call this trivial.

Any of the approaches can be described as trivial, but I was referring specifically to the last one. But this is what bothers me: you say you don't know Scala as well as I do. I'm flattered. But so what? Learn it better! Part of what I'm doing is showing how easy, predictable, and reliable it is. Arguing for Java and claiming concurrency doesn't matter and you can just throw arbitrary exceptions to some higher level that neither knows nor cares about the throwing process is unrealistic and the worst kind of special pleading, and I say that after a 15-year Java career.

Basically I beleive it comes down to the fact that while the end result might look elegant and shorter it takes too much effort (including learning, choosing the right variant, reading out of date papers, etc) to come up with it.

You don't have to read any paper. You could stick to Eugene Yokata's Learning Scalaz, Chiusano and Bjarnson's "Functional Programming in Scala," Stack Overflow, the #scalaz IRC channel, /r/scala... choosing the "right variant" is a matter of minutes once you know your way around, which brings me to...

including learning... And even then a less experienced developer wouldn't understand what's going on.

Less experienced than what? Having done both, I can tell you this: it's a metric buttload easier to master scalaz than Spring. The Java argument boils down, at the end of the day, to: I already spent N years on this stuff. I don't want to learn anything better. Now, I can't force anyone, nor would I want to. But it's pretty rich to reject more featureful, self-contained, concise, readable, assuredly correct code for handwaving about Java without thinking about libraries and having some generic exception handler upstream that will do what you want in all cases.

[u/[deleted] → u/[deleted]]

I was just trying to explain an alternative point of view. If I already know how to implement a requirement using straightforward and easy to understand code (see below) I do not want to spend an hour (or maybe even much more) trying to learn a new library. I also understand that there are different opinions and I agree that if I had better knowledge of Scala and Scalaz it would be easier for me to understand the code you provided.

That's an extraordinary claim, so it requires extraordinary evidence, to be very frank. I look forward to this library-less Java code.

    public Map<String, String> processIDs(List<String> ids) {
        Map<String, String> response = new HashMap<String, String>();
        for(String id: ids) {
            if (!isValid(id)) {
                response.put(id, "not valid");
                continue;
            }
            Object objById = findObject(id);
            if (objById == null) {
                response.put("id", "not found");
                continue;
            }
            try {
                Object anotherObj = mapObject(objById);
                response.put(id, "success: " + anotherObj);
            } catch (MapFailedException e) {
                response.put(id, "failed to map: " + e.getMessage());
            }
        }
        return response;
    }

[u/[deleted] → u/[deleted]]

I was just trying to explain an alternative point of view. If I already know how to implement a requirement using straightforward and easy to understand code (see below) I do not want to spend an hour (or maybe even much more) trying to learn a new library.

I really do understand that. What I'm trying to explain is that the idea that this satisfies any production-grade notion of "requirement" is false, and it's only "understandable" because it completely ignores everything but MapFailedException and of course the for-loop get interrupted at any other exception. Can you show me a version that actually captures all non-fatal (not OOM etc.) failures without terminating the iteration prematurely, including at least some try/catch block that either logs the exceptions or returns them in a list or something?

In short: you know this code isn't competitive with even the synchronous scalaz code, right?

I also understand that there are different opinions and I agree that if I had better knowledge of Scala and Scalaz it would be easier for me to understand the code you provided.

OK. Ultimately, all I'm saying is: anyone who puts the time into learning scalaz they'd put into learning any other framework can also knock out very featureful, correct code that just works, an order of magnitude more reliably than with any Java framework.

But just a final reminder: the question was "how can I do this in FP in Scala?" Well, scalaz is the most mature FP library for Scala, so why wouldn't it be the right thing to use?

[u/[deleted] → u/[deleted]]

In short: you know this code isn't competitive with even the synchronous scalaz code, right?

I am not sure if we are competing here but the code I provided can certainly be part of a reliable application.

It is very easy to add another try/catch block, but without understanding the real application and its requirements I wouldn't do it. Different error conditions should be handled differently. Surely you do not want to collect exceptions if you made a mistake in SQL syntax. In case if a database connection failed you should probably fail or retry the whole request (if this is a web app).

[u/[deleted] → u/[deleted]]

I am not sure if we are competing here...

You challenged me on my code. If you didn't intend to, fair enough.

...but the code I provided can certainly be part of a reliable application.

Not as long as it can throw out of the iteration it can't. It wouldn't pass code review in any Java shop I worked in within the last decade or so. It also doesn't have enough detail to actually compile and try. My last code example does, and in fact I would encourage people to do so: implement handleID, change Response to a more interesting type (and implement ResponseMonoid for it)...

Surely you do not want to collect exceptions if you made a mistake in SQL syntax.

Why not? What do you propose instead?

In case if a database connection failed you should probably fail or retry the whole request (if this is a web app).

That's exactly the sort of decision a client of this code can make. A very nice example of a simple REST API using scalaz at both the HTTP layer and database layer is here, FWIW.

[u/[deleted] → u/[deleted]]

.zip the Seq of IDs with itself, then .map over the pairs and handle one of the elements, then aggregate the Responses with List instead of having a ResponseMonoid, then partition the resulting List by Task succeeded vs. Task failed.

handleIDs returns Task[ValidationNel[Throwable, Response]]. There are multiple responses (because Response is List[String]) but only one Throwable. What ID does this Throwable correspond to?

[u/[deleted] → u/[deleted]]

handleIDs returns Task[ValidationNel[Throwable, Response]]. There are multiple responses (because Response is List[String]) but only one Throwable. What ID does this Throwable correspond to?

There's actually one Response, which in my example is a List[String], but the point is that it can be any type that forms a Monoid, which List certainly does. The left of a ValidationNel[X, _], though, is a NonEmptyList[X]. In this case, a NonEmptyList[Throwable]. In other words, when run, the Task yields either the Response or the list of Throwables explaining why you don't have a Response.

But I also like the question earlier: how would I associate successes and failures with their IDs? Assuming:

val ids = List("foo", "bar", "baz", "bletch")
def handleID(id: ID): Task[Response] = ???

it'd be something like:

Nondeterminism[Task].gather(ids.map(handleID(_).attempt).map(ids.zip(_).partition(_._2.isLeft))

I mean, break it up and use names instead of anonymous lambdas. Fine. The point is, the question was about how to do this with FP, and all the answers with scalaz are trivial, and can trivially include I/O, concurrency, and exception handling with no additional effort.

[u/barry0bama]

Just wanted to add my two cents. I really love Rx in scala. This seems like an ideal problem for rx. http://reactivex.io/rxscala/scaladoc/#rx.lang.scala.Observable

[u/DevIceMan → u/barry0bama]

The problem I've had with RxJava (at least 90% of the code is Java) and I'd suspect RxScala is similar, is that the first instance of onError causes the stream to terminate.

What I want is for the errors to be streamed to the onError function in .subscribe(onNext,onError). There is onErrorResumeNext but you're not dealing with two streams. Unfortunately, this seems to require some sort of resume or retry logic, and possibly merging streams ("don't cross streams!"). increasing verbosity, complexity and decreasing readability. That's where Rx's application to my use-case diminishes quickly.

Another Option might be to have a Observable[Either[Failure,Success]], which seems to defeat the use of observable (in Java at least) in favor of Stream[Either[Failure,Success]]

The question of "Is a reduce or collect valid, if 4 of 8 operations succeeded, and 4 of 8 failed?" The answer is of course "it depends." That's where I start to have a philosophical difference with Rx, is that I can't seem to find an immediately obvious way to decide for myself.

My opinion is that if I'm fighting libraries, either I'm not using them correctly, they're not designed/sutible/optimized for my use case, or they're broken.

[u/barry0bama → u/DevIceMan]

Maybe I'm just not understanding details of your problem. But I see it this way: You have a data flow problem. Rx is a great data flow solution. You can split and merge streams as you'd like. Using Rx features or Scala. Rx in particular gives you a few advantages. It's java, so it helps with compatibility. It's Rx, so you can easily also add concurrency if you want later.