Fluxはnull値を受け入れませんか？

Question

Fluxのセットがあるとします。これは、二重機能と一緒に圧縮したいと考えています。オブジェクトが特定の制約を満たしている場合

Flux<String> flux1 = Flux.just("A", "B", "C"); 
Flux<String> flux2 = Flux.just("D", "E", "F"); 
Flux.zip(flux1, flux2, this::zipString).subscribe(System.out::println); 

このbifunctionは以下nullを返します。 Fluxを一緒に圧縮した後にフィルタを適用する可能性があります。

public String zipString(String a, String b) { 
    if (a.equals("B")) 
     return null; 
    return a + b; 
} 

この戦略はNullPointerExceptionを投げてしまいます。

Exception in thread "main" 
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.NullPointerException: The zipper returned a null value 
Caused by: java.lang.NullPointerException: The zipper returned a null value 
    at java.util.Objects.requireNonNull(Objects.java:228) 
    at reactor.core.publisher.FluxZip$ZipCoordinator.drain(FluxZip.java:711) 
    at reactor.core.publisher.FluxZip$ZipInner.onSubscribe(FluxZip.java:861) 
    at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:53) 
    at reactor.core.publisher.FluxArray.subscribe(FluxArray.java:59) 

EDIT：nullを含むFluxがある場合にも発生します。

Flux<String> flux2 = Flux.just(null, "B", "C"); 

これは私に質問します。 nullの値がFluxに入ることを禁止されているのはなぜですか？

下にユースケースを軽減するためのいくつかの潜在的な戦略は何です：

がflux2したがって、セットが破棄されなければならない、我々は捨てるべき値が含まれています。

Answer 1

反応性仕様では、ストリーム内でnullを使用できません。 "n/a"のような定数を選び、後でそれをフィルタリングしてください。 onSubscribe、onNext、のonError又はonCompleteの任意の提供パラメータは、それが れるjava.langをスローする必要があり、その場合にヌルである場合を除いて、通常 を返さなければならないの呼び出し

https://github.com/reactive-streams/reactive-streams-jvm#2-subscriber-code

1. .NullPointerExceptionが呼び出し元に返されます。