1.X: UnicastSubject fail-fast and delay-error behavior

[mostroverkhov]

This PR adds support for delay-error behavior to UnicastSubject with methods UnicastSubject<T> create(boolean delayError), UnicastSubject<T> create(int capacityHint, Action0 onTerminated, boolean delayError). Behavior of existing factory methods was not changed, and is fail-fast.

Relates to #5165

[codecov]

Codecov Report

Merging #5195 into 1.x will increase coverage by 0.02%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##                1.x    #5195      +/-   ##
============================================
+ Coverage     84.36%   84.38%   +0.02%     
- Complexity     2880     2883       +3     
============================================
  Files           290      290              
  Lines         18115    18123       +8     
  Branches       2478     2479       +1     
============================================
+ Hits          15282    15293      +11     
+ Misses         1964     1961       -3     
  Partials        869      869

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/rx/subjects/UnicastSubject.java	84.56% <100%> (+0.8%)	9 <4> (+2)	⬆️
...ain/java/rx/internal/schedulers/SchedulerWhen.java	83.78% <0%> (-4.06%)	4% <0%> (ø)
...n/java/rx/subjects/SubjectSubscriptionManager.java	80.71% <0%> (-1.43%)	15% <0%> (-1%)
...main/java/rx/internal/operators/OperatorMerge.java	86.83% <0%> (+0.23%)	7% <0%> (ø)	⬇️
...in/java/rx/internal/operators/OperatorPublish.java	79.32% <0%> (+0.84%)	8% <0%> (ø)	⬇️
...n/java/rx/internal/operators/CachedObservable.java	84.78% <0%> (+1.08%)	6% <0%> (ø)	⬇️
.../java/rx/internal/operators/BackpressureUtils.java	70.45% <0%> (+2.27%)	28% <0%> (+1%)	⬆️
...in/java/rx/internal/schedulers/SleepingAction.java	61.11% <0%> (+5.55%)	3% <0%> (+1%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 0f1542d...d3e9cd1. Read the comment docs.

[akarnokd]

You should have a local delayError field for this, otherwise you are re-reading the flag on each emission which can cause unnecessary false sharing.

Queue<Object> q = queue;
boolean delayError = this.delayError;
// ...

[mostroverkhov]

done

[akarnokd]

Thanks!