Where communities thrive


  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
People
Repo info
Activity
    Dorus
    @Dorus
    flatMap(e => asyncCall()) certainly would run in parallel.
    Dmitry V. Roenko
    @Eternity-Yarr
    Hey all! Can anyone explain to me this behavior: https://gist.github.com/Eternity-Yarr/93dd29aa90824cb92c53494c8330a601?ts=2 ?
    Basically I'm trying to create observable from infinite iterable
    with some queue backing it. And it kinda works, except queue is always empty for some reason, even though producer supplies 1 item per 1ms and consumers is only 4 threads with 1 sec delay each
    Dmitry V. Roenko
    @Eternity-Yarr
    I suspect that there is some unbounded queue somewhere deep inside of OnSubscribeFromIterable
    MergeSubscriber.queue ?
    looks like it..
    Dmitry V. Roenko
    @Eternity-Yarr
    okay! found it :)) it's because of flatMap
    thanks!
    Toon Sevrin
    @toonsevrin

    Hey folks, I am just starting out with RxJava and am trying to implement it as an event bus.
    Is is very bad practice to pass Cancellable events through the bus (these events are mutable, and after being processed the behaviour is result or not)?
    If it's ok how can I subscribe with a priority in the call iteration?

    I would really appreciate any help that can be given! Thanks guys!

    The bus is currently a PublishSubject btw ;)
    Dorus
    @Dorus
    @stink123456 Observables are always cancellable. You can just push observables trough.
    cancel is just unsubscribe.
    Darius Lapūnas
    @DariusL
    Well, if your bus can have many listeners and the items are mutable, you might run into trouble
    Toon Sevrin
    @toonsevrin
    @Dorus Oh lol, I meant cancelling the passed individual items (Setting a boolean on then that will be processed by the emitter)
    @DariusL What would be the recommended approach?
    @DariusL The API I used before this used annotations with an EventPriority enum.
    Darius Lapūnas
    @DariusL
    Depends on what you're trying to accomplish. IMO, if it's cancellable, it's not an event, it's a request. Then you need a queue, which is what most Schedulers are. If you want to use a bus, your events should be immutable, so that any number of listeners could handle them in their own ways without impacting each other.
    Toon Sevrin
    @toonsevrin
    Aah I see, that's the answer I was looking for, thanks a lot. An example could be a webApp with a request to display a page. The security extension/module could catch this request and if the user is not authorized, deny the request.
    Darius Lapūnas
    @DariusL
    yeah, if the order is important, there's no place for an even bus
    Toon Sevrin
    @toonsevrin
    So what would the recommended approach be if there are multiple actors on the request result (say if the security is allowed, the request will only pass after another module passes it)
    You say some kind of queue right
    There's probably a fundamental design flaw in this because actors have to know of each other to sort themselves in the right order :(
    Alexander Lin
    @a2lin
    Am I using Rxjava wrong if I never have a subscriber? I want to use it to represent a pipeline of transformations from request -> response on the server side, but I don't really have side effects in this pipeline so subscribers don't really fit the model.
    Darius Lapūnas
    @DariusL
    @a2lin yes, the point is that the subscriber receives the processed data. You would return your response to the caller by receiving it in the subscriber. In fact, nothing should be executed until you (or some operator) call subscribe. If you're okay with waiting, you can use a blocking observable to wait for data.
    James
    @jemmrich
    I have services which return observables (like http services), the services accept params for the http call, how can I use Zip to execute multiple observables which require params that depend on the previous observables response? Is that possible or do I just have to nest subscribes within subscribes within subscribes?
    Dorus
    @Dorus
    you can zip as long as all sources emit the same number of values
    pretty common pattern
    Ben Christensen
    @benjchristensen
    Don't nest subscribes. That's an anti-pattern
    Dorus
    @Dorus
    Bit old, but for example it's used here: http://techblog.netflix.com/2013/02/rxjava-netflix-api.html (the zip trick, not nesting subscribers)
    there is also a nice precentation but i cant find it quickly.
    Ben Christensen
    @benjchristensen
    It causes grief with error handling, backpressure, etc
    Yes, that example is good. Here is code as well related to that blog and similar presentations: https://github.com/Netflix/ReactiveLab/blob/master/reactive-lab-gateway/src/main/java/io/reactivex/lab/gateway/routes/RouteForDeviceHome.java
    James
    @jemmrich
    I just have a couple (not a lot) of apis that I need to call sequentially and for now I am having to nest the subscribes which is something that makes me cringe. But the apis I need to call supply data that the next api needs for its call
    Ben Christensen
    @benjchristensen
    Use flatMap in almost all cases where nested subscribes occur.
    James
    @jemmrich
    @Dorus Thanks I am going to read that link you provided
    Ben Christensen
    @benjchristensen
    It is almost never needed or a good idea to
    nest subscribes.
    James
    @jemmrich
    @benjchristensen absolutely agree
    Dorus
    @Dorus
    Ben Christensen
    @benjchristensen
    FlatMap allows you to get the data from a previous call.
    Hi :-)
    James
    @jemmrich
    @benjchristensen Ill have to look for some examples on flatmap, I saw it mentioned a few times, but didn't see or notice how to retrieve the results from the previous observable
    @Dorus Thanks going to have a look at it
    Ben Christensen
    @benjchristensen
    The link I gave before shows how to use it. It's the most used operator for composition.
    stream.flatmap(value -> return something(value.arg))
    James
    @jemmrich
    @benjchristensen Thanks, you all have been very helpful
    Włodzimierz Rożkow
    @rozhok
    hey guys, need a quick advice about rx.
    background: we have microservices-based application, where some of services make calls to each other. we need to do this in parallel/async manner. values returned from queries are heterogeneous so they couldn't be zipped properly afaik
    does rx will help me here or I may stick with ol' good CompletableFuture?
    Thomas Wilgenbus
    @regnarock

    Hi there! Question about a combination of operators to solve a specific need of mine:
    Prelude: I have two async Observables A and B. A emits 1+ elements and B always 1. I want to combine the two such as the ouput of A: [A1, A2, A3] and B: [B1] -> [A1B1, A2B1, A3B1].

    my solution:

    A.toList().zipWith(B, (list_of_A, B1) -> Lists.transform(list_of_A, An -> combine(An, B1)))

    it does work but I am unsure if this is the right way to do it?
    (note: I used Lists.tranform from guava for readability purpose)

    Dorus
    @Dorus
    @regnarock Flatmap?
    B.flatMap(() -> A, (b, a) -> {...})