Skip to content

Commit bf9a541

Browse files
akarnokdakarnokd
committed
Copy internal helper methods from RxJava to avoid OSGi problems
1 parent b8a5acf commit bf9a541

33 files changed

+4207
-56
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ Check out the [https://github.com/akarnokd/RxJavaBridge](https://github.com/akar
2121

2222
```
2323
dependencies {
24-
compile "com.github.akarnokd:rxjava3-interop:3.0.0"
24+
compile "com.github.akarnokd:rxjava3-interop:3.0.1"
2525
}
2626
```
2727

‎gradle.properties‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
GROUP=com.github.akarnokd
2-
VERSION_NAME=3.0.0
3-
version=3.0.0
2+
VERSION_NAME=3.0.1
3+
version=3.0.1
44

55
POM_ARTIFACT_ID=rxjava3-interop
66
POM_NAME=Interop library between RxJava 1.x and 3.x

‎src/main/java/hu/akarnokd/rxjava3/interop/BackpressureHelper.java‎

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
/*
2+
* Copyright 2016-2020 David Karnok
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package hu.akarnokd.rxjava3.interop;
18+
19+
import java.util.concurrent.atomic.AtomicLong;
20+
21+
import io.reactivex.rxjava3.annotations.NonNull;
22+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
23+
24+
/**
25+
* Utility class to help with backpressure-related operations such as request aggregation.
26+
*/
27+
final class BackpressureHelper {
28+
/** Utility class. */
29+
private BackpressureHelper() {
30+
throw new IllegalStateException("No instances!");
31+
}
32+
33+
/**
34+
* Adds two long values and caps the sum at {@link Long#MAX_VALUE}.
35+
* @param a the first value
36+
* @param b the second value
37+
* @return the sum capped at {@link Long#MAX_VALUE}
38+
*/
39+
public static long addCap(long a, long b) {
40+
long u = a + b;
41+
if (u < 0L) {
42+
return Long.MAX_VALUE;
43+
}
44+
return u;
45+
}
46+
47+
/**
48+
* Multiplies two long values and caps the product at {@link Long#MAX_VALUE}.
49+
* @param a the first value
50+
* @param b the second value
51+
* @return the product capped at {@link Long#MAX_VALUE}
52+
*/
53+
public static long multiplyCap(long a, long b) {
54+
long u = a * b;
55+
if (((a | b) >>> 31) != 0) {
56+
if (u / a != b) {
57+
return Long.MAX_VALUE;
58+
}
59+
}
60+
return u;
61+
}
62+
63+
/**
64+
* Atomically adds the positive value n to the requested value in the {@link AtomicLong} and
65+
* caps the result at {@link Long#MAX_VALUE} and returns the previous value.
66+
* @param requested the {@code AtomicLong} holding the current requested value
67+
* @param n the value to add, must be positive (not verified)
68+
* @return the original value before the add
69+
*/
70+
public static long add(@NonNull AtomicLong requested, long n) {
71+
for (;;) {
72+
long r = requested.get();
73+
if (r == Long.MAX_VALUE) {
74+
return Long.MAX_VALUE;
75+
}
76+
long u = addCap(r, n);
77+
if (requested.compareAndSet(r, u)) {
78+
return r;
79+
}
80+
}
81+
}
82+
83+
/**
84+
* Atomically adds the positive value n to the requested value in the {@link AtomicLong} and
85+
* caps the result at {@link Long#MAX_VALUE} and returns the previous value and
86+
* considers {@link Long#MIN_VALUE} as a cancel indication (no addition then).
87+
* @param requested the {@code AtomicLong} holding the current requested value
88+
* @param n the value to add, must be positive (not verified)
89+
* @return the original value before the add
90+
*/
91+
public static long addCancel(@NonNull AtomicLong requested, long n) {
92+
for (;;) {
93+
long r = requested.get();
94+
if (r == Long.MIN_VALUE) {
95+
return Long.MIN_VALUE;
96+
}
97+
if (r == Long.MAX_VALUE) {
98+
return Long.MAX_VALUE;
99+
}
100+
long u = addCap(r, n);
101+
if (requested.compareAndSet(r, u)) {
102+
return r;
103+
}
104+
}
105+
}
106+
107+
/**
108+
* Atomically subtract the given number (positive, not validated) from the target field unless it contains {@link Long#MAX_VALUE}.
109+
* @param requested the target field holding the current requested amount
110+
* @param n the produced element count, positive (not validated)
111+
* @return the new amount
112+
*/
113+
public static long produced(@NonNull AtomicLong requested, long n) {
114+
for (;;) {
115+
long current = requested.get();
116+
if (current == Long.MAX_VALUE) {
117+
return Long.MAX_VALUE;
118+
}
119+
long update = current - n;
120+
if (update < 0L) {
121+
RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
122+
update = 0L;
123+
}
124+
if (requested.compareAndSet(current, update)) {
125+
return update;
126+
}
127+
}
128+
}
129+
130+
/**
131+
* Atomically subtract the given number (positive, not validated) from the target field if
132+
* it doesn't contain {@link Long#MIN_VALUE} (indicating some cancelled state) or {@link Long#MAX_VALUE} (unbounded mode).
133+
* @param requested the target field holding the current requested amount
134+
* @param n the produced element count, positive (not validated)
135+
* @return the new amount
136+
*/
137+
public static long producedCancel(@NonNull AtomicLong requested, long n) {
138+
for (;;) {
139+
long current = requested.get();
140+
if (current == Long.MIN_VALUE) {
141+
return Long.MIN_VALUE;
142+
}
143+
if (current == Long.MAX_VALUE) {
144+
return Long.MAX_VALUE;
145+
}
146+
long update = current - n;
147+
if (update < 0L) {
148+
RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
149+
update = 0L;
150+
}
151+
if (requested.compareAndSet(current, update)) {
152+
return update;
153+
}
154+
}
155+
}
156+
}

‎src/main/java/hu/akarnokd/rxjava3/interop/CompletableV1ToCompletableV3.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 David Karnok
2+
* Copyright 2016-2020 David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/interop/CompletableV1ToMaybeV3.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 David Karnok
2+
* Copyright 2016-2020 David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/interop/CompletableV3ToCompletableV1.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 David Karnok
2+
* Copyright 2016-2020 David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/interop/DisposableHelper.java‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
/*
2+
* Copyright 2016-2020 David Karnok
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package hu.akarnokd.rxjava3.interop;
18+
19+
import java.util.Objects;
20+
import java.util.concurrent.atomic.AtomicReference;
21+
22+
import io.reactivex.rxjava3.disposables.Disposable;
23+
import io.reactivex.rxjava3.exceptions.ProtocolViolationException;
24+
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
25+
26+
/**
27+
* Utility methods for working with Disposables atomically.
28+
*/
29+
enum DisposableHelper implements Disposable {
30+
/**
31+
* The singleton instance representing a terminal, disposed state, don't leak it.
32+
*/
33+
DISPOSED
34+
;
35+
36+
/**
37+
* Checks if the given Disposable is the common {@link #DISPOSED} enum value.
38+
* @param d the disposable to check
39+
* @return true if d is {@link #DISPOSED}
40+
*/
41+
public static boolean isDisposed(Disposable d) {
42+
return d == DISPOSED;
43+
}
44+
45+
/**
46+
* Atomically sets the field to the given non-null Disposable and returns true
47+
* or returns false if the field is non-null.
48+
* If the target field contains the common DISPOSED instance, the supplied disposable
49+
* is disposed. If the field contains other non-null Disposable, an IllegalStateException
50+
* is signalled to the RxJavaPlugins.onError hook.
51+
*
52+
* @param field the target field
53+
* @param d the disposable to set, not null
54+
* @return true if the operation succeeded, false
55+
*/
56+
public static boolean setOnce(AtomicReference<Disposable> field, Disposable d) {
57+
Objects.requireNonNull(d, "d is null");
58+
if (!field.compareAndSet(null, d)) {
59+
d.dispose();
60+
if (field.get() != DISPOSED) {
61+
reportDisposableSet();
62+
}
63+
return false;
64+
}
65+
return true;
66+
}
67+
68+
/**
69+
* Atomically disposes the Disposable in the field if not already disposed.
70+
* @param field the target field
71+
* @return true if the current thread managed to dispose the Disposable
72+
*/
73+
public static boolean dispose(AtomicReference<Disposable> field) {
74+
Disposable current = field.get();
75+
Disposable d = DISPOSED;
76+
if (current != d) {
77+
current = field.getAndSet(d);
78+
if (current != d) {
79+
if (current != null) {
80+
current.dispose();
81+
}
82+
return true;
83+
}
84+
}
85+
return false;
86+
}
87+
88+
/**
89+
* Verifies that current is null, next is not null, otherwise signals errors
90+
* to the RxJavaPlugins and returns false.
91+
* @param current the current Disposable, expected to be null
92+
* @param next the next Disposable, expected to be non-null
93+
* @return true if the validation succeeded
94+
*/
95+
public static boolean validate(Disposable current, Disposable next) {
96+
if (next == null) {
97+
RxJavaPlugins.onError(new NullPointerException("next is null"));
98+
return false;
99+
}
100+
if (current != null) {
101+
next.dispose();
102+
reportDisposableSet();
103+
return false;
104+
}
105+
return true;
106+
}
107+
108+
/**
109+
* Reports that the disposable is already set to the RxJavaPlugins error handler.
110+
*/
111+
public static void reportDisposableSet() {
112+
RxJavaPlugins.onError(new ProtocolViolationException("Disposable already set!"));
113+
}
114+
115+
/**
116+
* Atomically tries to set the given Disposable on the field if it is null or disposes it if
117+
* the field contains {@link #DISPOSED}.
118+
* @param field the target field
119+
* @param d the disposable to set
120+
* @return true if successful, false otherwise
121+
*/
122+
public static boolean trySet(AtomicReference<Disposable> field, Disposable d) {
123+
if (!field.compareAndSet(null, d)) {
124+
if (field.get() == DISPOSED) {
125+
d.dispose();
126+
}
127+
return false;
128+
}
129+
return true;
130+
}
131+
132+
@Override
133+
public void dispose() {
134+
// deliberately no-op
135+
}
136+
137+
@Override
138+
public boolean isDisposed() {
139+
return true;
140+
}
141+
}

‎src/main/java/hu/akarnokd/rxjava3/interop/DisposableV3ToSubscriptionV1.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 David Karnok
2+
* Copyright 2016-2020 David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.

‎src/main/java/hu/akarnokd/rxjava3/interop/FlowableV3ToObservableV1.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 David Karnok
2+
* Copyright 2016-2020 David Karnok
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -58,23 +58,23 @@ static final class SourceSubscriber<T>
5858
@Override
5959
public void request(long n) {
6060
if (n != 0L) {
61-
io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper.deferredRequest(this, requested, n);
61+
SubscriptionHelper.deferredRequest(this, requested, n);
6262
}
6363
}
6464

6565
@Override
6666
public void unsubscribe() {
67-
io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper.cancel(this);
67+
SubscriptionHelper.cancel(this);
6868
}
6969

7070
@Override
7171
public boolean isUnsubscribed() {
72-
return io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper.CANCELLED == get();
72+
return SubscriptionHelper.CANCELLED == get();
7373
}
7474

7575
@Override
7676
public void onSubscribe(org.reactivestreams.Subscription s) {
77-
io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper.deferredSetOnce(this, requested, s);
77+
SubscriptionHelper.deferredSetOnce(this, requested, s);
7878
}
7979

8080
@Override

0 commit comments

Comments
 (0)