<settings>
<servers>
<server>
<id>oss.sonatype</id>
<username>CHANGE_ME</username>
<password>CHANGE_ME</password>
</server>
</servers>
<profiles>
<profile>
<id>ossrh</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<properties>
<gpg.executable>gpg</gpg.executable>
<gpg.passphrase>CHANGE_ME</gpg.passphrase>
<gpg.keyname>CHANGE_ME</gpg.keyname>
</properties>
</profile>
</profiles>
</settings>
Hi,
I am creating a Multi from a publisher and I am getting an error.
Mutiny Code:
final Multi<Object> multiPublisher = Multi.createFrom().publisher(s -> {
s.onNext("Hello");
s.onNext("Mutiny Publisher");
});
multiPublisher.subscribe().with(System.out::println);Exception:Caused by: java.lang.NullPointerException: Subscription not set!
This is similar to what I would do with RxJava:
Flowable<String> flowable = Flowable.fromPublisher(s -> {
s.onNext("Hello");
s.onNext("RxJava Flowable");
});
flowable.subscribe(System.out::println);Any hints?
Hello, your publisher is not correct (violate the reactive streams protocol). You need to send a subscription before calling onNext and wait for downstream requests before calling onNext. I recommend you to use: Multi.createFrom().emitter(e -> …) which will makes shure this is done.
Hi,
I am mapping some RX Java code into Mutiny and I'm having some problems with grouping. In this example I am grouping employee names by rating into a multi map.
RX Java
Observable.fromArray(employees)
.groupBy(employee -> employee.rating)
.flatMapSingle(e -> e.toMultimap(key -> e.getKey(), employee -> employee.name))
.subscribe(System.out::println);Mutiny
Multi.createFrom().items(employees)
.groupItems().by(employee -> employee.rating)
.map(e -> e.collectItems().asMultiMap(key -> e.key(), employee -> employee.name))
.flatMap(Uni::toMulti)
.subscribe().with(System.out::println);The Mutiny version it's a bit weird. Any suggestions on how to improve this?
Hi,
Sorry for being such a nag, but I was looking at the concepts of caching and replaying events in RX Java and although the caching is very similar in Mutiny, I couldn't find anything for the replay.
Examples of replay in RX Java:
final Observable<Long> source = Observable.interval(1, TimeUnit.SECONDS).replay().autoConnect();
final Observable<Long> source = Observable.interval(1, TimeUnit.SECONDS).replay(5).autoConnect();
final Observable<Long> source = Observable.interval(1, TimeUnit.SECONDS)
.replay(2, 1, TimeUnit.SECONDS).autoConnect();Regarding hotstreams I also couldn't find a BehaviourProcessor where the subscriber can receive the last event.