diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java index f16fa2cf03..8413bbc756 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java @@ -44,7 +44,7 @@ * Reactive Streams implementations. *

* The Flowable hosts the default buffer size of 128 elements for operators, accessible via {@link #bufferSize()}, - * that can be overridden globally via the system parameter {@code rx2.buffer-size}. Most operators, however, have + * that can be overridden globally via the system parameter {@code rx3.buffer-size}. Most operators, however, have * overloads that allow setting their internal buffer size explicitly. *

* The documentation for this class makes use of marble diagrams. The following legend explains these diagrams: @@ -153,7 +153,7 @@ public abstract class Flowable implements Publisher { /** The default buffer size. */ static final int BUFFER_SIZE; static { - BUFFER_SIZE = Math.max(1, Integer.getInteger("rx2.buffer-size", 128)); + BUFFER_SIZE = Math.max(1, Integer.getInteger("rx3.buffer-size", 128)); } /** @@ -225,7 +225,7 @@ public static Flowable ambArray(Publisher... sources) { /** * Returns the default internal buffer size used by most async operators. - *

The value can be overridden via system parameter {@code rx2.buffer-size} + *

The value can be overridden via system parameter {@code rx3.buffer-size} * before the Flowable class is loaded. * @return the default internal buffer size. */ diff --git a/src/main/java/io/reactivex/rxjava3/core/Observable.java b/src/main/java/io/reactivex/rxjava3/core/Observable.java index bcc013a10f..018ead3ff8 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Observable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Observable.java @@ -43,7 +43,7 @@ * for such non-backpressured flows, which {@code Observable} itself implements as well. *

* The Observable's operators, by default, run with a buffer size of 128 elements (see {@link Flowable#bufferSize()}), - * that can be overridden globally via the system parameter {@code rx2.buffer-size}. Most operators, however, have + * that can be overridden globally via the system parameter {@code rx3.buffer-size}. Most operators, however, have * overloads that allow setting their internal buffer size explicitly. *

* The documentation for this class makes use of marble diagrams. The following legend explains these diagrams: @@ -160,7 +160,7 @@ public static Observable ambArray(ObservableSource... source /** * Returns the default 'island' size or capacity-increment hint for unbounded buffers. *

Delegates to {@link Flowable#bufferSize} but is public for convenience. - *

The value can be overridden via system parameter {@code rx2.buffer-size} + *

The value can be overridden via system parameter {@code rx3.buffer-size} * before the {@link Flowable} class is loaded. * @return the default 'island' size or capacity-increment hint */ diff --git a/src/main/java/io/reactivex/rxjava3/core/Scheduler.java b/src/main/java/io/reactivex/rxjava3/core/Scheduler.java index 952bbec74c..a300cb02ce 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Scheduler.java +++ b/src/main/java/io/reactivex/rxjava3/core/Scheduler.java @@ -74,7 +74,7 @@ * based on the relative time between it and {@link Worker#now(TimeUnit)}. However, drifts or changes in the * system clock could affect this calculation either by scheduling subsequent runs too frequently or too far apart. * Therefore, the default implementation uses the {@link #clockDriftTolerance()} value (set via - * {@code rx2.scheduler.drift-tolerance} in minutes) to detect a drift in {@link Worker#now(TimeUnit)} and + * {@code rx3.scheduler.drift-tolerance} in minutes) to detect a drift in {@link Worker#now(TimeUnit)} and * re-adjust the absolute/relative time calculation accordingly. *

* The default implementations of {@link #start()} and {@link #shutdown()} do nothing and should be overridden if the @@ -92,17 +92,17 @@ public abstract class Scheduler { /** * The tolerance for a clock drift in nanoseconds where the periodic scheduler will rebase. *

- * The associated system parameter, {@code rx2.scheduler.drift-tolerance}, expects its value in minutes. + * The associated system parameter, {@code rx3.scheduler.drift-tolerance}, expects its value in minutes. */ static final long CLOCK_DRIFT_TOLERANCE_NANOSECONDS; static { CLOCK_DRIFT_TOLERANCE_NANOSECONDS = TimeUnit.MINUTES.toNanos( - Long.getLong("rx2.scheduler.drift-tolerance", 15)); + Long.getLong("rx3.scheduler.drift-tolerance", 15)); } /** * Returns the clock drift tolerance in nanoseconds. - *

Related system property: {@code rx2.scheduler.drift-tolerance} in minutes. + *

Related system property: {@code rx3.scheduler.drift-tolerance} in minutes. * @return the tolerance in nanoseconds * @since 2.0 */ @@ -345,7 +345,7 @@ public S when(@NonNull Function * If the {@code Worker} is disposed, the {@code schedule} methods diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ComputationScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ComputationScheduler.java index 56fc837cff..eb686b17e0 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/ComputationScheduler.java +++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/ComputationScheduler.java @@ -38,7 +38,7 @@ public final class ComputationScheduler extends Scheduler implements SchedulerMu * Key to setting the maximum number of computation scheduler threads. * Zero or less is interpreted as use available. Capped by available. */ - static final String KEY_MAX_THREADS = "rx2.computation-threads"; + static final String KEY_MAX_THREADS = "rx3.computation-threads"; /** The maximum number of computation scheduler threads. */ static final int MAX_THREADS; @@ -47,7 +47,7 @@ public final class ComputationScheduler extends Scheduler implements SchedulerMu final ThreadFactory threadFactory; final AtomicReference pool; /** The name of the system property for setting the thread priority for this Scheduler. */ - private static final String KEY_COMPUTATION_PRIORITY = "rx2.computation-priority"; + private static final String KEY_COMPUTATION_PRIORITY = "rx3.computation-priority"; static { MAX_THREADS = cap(Runtime.getRuntime().availableProcessors(), Integer.getInteger(KEY_MAX_THREADS, 0)); diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java index 71fdfd3cd6..f4edcf0208 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java +++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java @@ -35,7 +35,7 @@ public final class IoScheduler extends Scheduler { static final RxThreadFactory EVICTOR_THREAD_FACTORY; /** The name of the system property for setting the keep-alive time (in seconds) for this Scheduler workers. */ - private static final String KEY_KEEP_ALIVE_TIME = "rx2.io-keep-alive-time"; + private static final String KEY_KEEP_ALIVE_TIME = "rx3.io-keep-alive-time"; public static final long KEEP_ALIVE_TIME_DEFAULT = 60; private static final long KEEP_ALIVE_TIME; @@ -46,7 +46,7 @@ public final class IoScheduler extends Scheduler { final AtomicReference pool; /** The name of the system property for setting the thread priority for this Scheduler. */ - private static final String KEY_IO_PRIORITY = "rx2.io-priority"; + private static final String KEY_IO_PRIORITY = "rx3.io-priority"; static final CachedWorkerPool NONE; diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadScheduler.java index 4322f0fe71..1cd48ce1e0 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadScheduler.java +++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/NewThreadScheduler.java @@ -32,7 +32,7 @@ public final class NewThreadScheduler extends Scheduler { private static final RxThreadFactory THREAD_FACTORY; /** The name of the system property for setting the thread priority for this Scheduler. */ - private static final String KEY_NEWTHREAD_PRIORITY = "rx2.newthread-priority"; + private static final String KEY_NEWTHREAD_PRIORITY = "rx3.newthread-priority"; static { int priority = Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY, diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java index 4e6d7844c7..cc82349cfc 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java +++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactory.java @@ -20,6 +20,8 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; +import io.reactivex.rxjava3.functions.Function; + /** * Manages the creating of ScheduledExecutorServices and sets up purging. */ @@ -29,14 +31,14 @@ private SchedulerPoolFactory() { throw new IllegalStateException("No instances!"); } - static final String PURGE_ENABLED_KEY = "rx2.purge-enabled"; + static final String PURGE_ENABLED_KEY = "rx3.purge-enabled"; /** * Indicates the periodic purging of the ScheduledExecutorService is enabled. */ public static final boolean PURGE_ENABLED; - static final String PURGE_PERIOD_SECONDS_KEY = "rx2.purge-period-seconds"; + static final String PURGE_PERIOD_SECONDS_KEY = "rx3.purge-period-seconds"; /** * Indicates the purge period of the ScheduledExecutorServices created by create(). @@ -90,40 +92,48 @@ public static void shutdown() { } static { - Properties properties = System.getProperties(); - - PurgeProperties pp = new PurgeProperties(); - pp.load(properties); - - PURGE_ENABLED = pp.purgeEnable; - PURGE_PERIOD_SECONDS = pp.purgePeriod; + SystemPropertyAccessor propertyAccessor = new SystemPropertyAccessor(); + PURGE_ENABLED = getBooleanProperty(true, PURGE_ENABLED_KEY, true, true, propertyAccessor); + PURGE_PERIOD_SECONDS = getIntProperty(PURGE_ENABLED, PURGE_PERIOD_SECONDS_KEY, 1, 1, propertyAccessor); start(); } - static final class PurgeProperties { - - boolean purgeEnable; - - int purgePeriod; - - void load(Properties properties) { - if (properties.containsKey(PURGE_ENABLED_KEY)) { - purgeEnable = Boolean.parseBoolean(properties.getProperty(PURGE_ENABLED_KEY)); - } else { - purgeEnable = true; + static int getIntProperty(boolean enabled, String key, int defaultNotFound, int defaultNotEnabled, Function propertyAccessor) { + if (enabled) { + try { + String value = propertyAccessor.apply(key); + if (value == null) { + return defaultNotFound; + } + return Integer.parseInt(value); + } catch (Throwable ex) { + return defaultNotFound; } + } + return defaultNotEnabled; + } - if (purgeEnable && properties.containsKey(PURGE_PERIOD_SECONDS_KEY)) { - try { - purgePeriod = Integer.parseInt(properties.getProperty(PURGE_PERIOD_SECONDS_KEY)); - } catch (NumberFormatException ex) { - purgePeriod = 1; + static boolean getBooleanProperty(boolean enabled, String key, boolean defaultNotFound, boolean defaultNotEnabled, Function propertyAccessor) { + if (enabled) { + try { + String value = propertyAccessor.apply(key); + if (value == null) { + return defaultNotFound; } - } else { - purgePeriod = 1; + return "true".equals(value); + } catch (Throwable ex) { + return defaultNotFound; } } + return defaultNotEnabled; + } + + static final class SystemPropertyAccessor implements Function { + @Override + public String apply(String t) throws Throwable { + return System.getProperty(t); + } } /** diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java index 35f60c8eb9..a1dba3036b 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java +++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/SingleScheduler.java @@ -31,7 +31,7 @@ public final class SingleScheduler extends Scheduler { final AtomicReference executor = new AtomicReference(); /** The name of the system property for setting the thread priority for this Scheduler. */ - private static final String KEY_SINGLE_PRIORITY = "rx2.single-priority"; + private static final String KEY_SINGLE_PRIORITY = "rx3.single-priority"; private static final String THREAD_NAME_PREFIX = "RxSingleScheduler"; diff --git a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java index 98189e8e6f..93a6e50c6c 100644 --- a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java +++ b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java @@ -30,14 +30,14 @@ *

* Supported system properties ({@code System.getProperty()}): *

*/ public final class Schedulers { @@ -112,8 +112,8 @@ private Schedulers() { * before the {@link Schedulers} class is referenced in your code. *

Supported system properties ({@code System.getProperty()}): *

*

* The default value of this scheduler can be overridden at initialization time via the @@ -157,8 +157,8 @@ public static Scheduler computation() { * before the {@link Schedulers} class is referenced in your code. *

Supported system properties ({@code System.getProperty()}): *

*

* The default value of this scheduler can be overridden at initialization time via the @@ -216,7 +216,7 @@ public static Scheduler trampoline() { * before the {@link Schedulers} class is referenced in your code. *

Supported system properties ({@code System.getProperty()}): *

*

* The default value of this scheduler can be overridden at initialization time via the @@ -265,7 +265,7 @@ public static Scheduler newThread() { * before the {@link Schedulers} class is referenced in your code. *

Supported system properties ({@code System.getProperty()}): *

*

* The default value of this scheduler can be overridden at initialization time via the diff --git a/src/test/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactoryTest.java b/src/test/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactoryTest.java index 9bf100fbbb..1fcbf105b4 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactoryTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/schedulers/SchedulerPoolFactoryTest.java @@ -18,12 +18,11 @@ import static org.junit.Assert.*; -import java.util.Properties; - import org.junit.Test; import io.reactivex.rxjava3.core.RxJavaTest; -import io.reactivex.rxjava3.internal.schedulers.SchedulerPoolFactory.PurgeProperties; +import io.reactivex.rxjava3.functions.Function; +import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.schedulers.Schedulers; import io.reactivex.rxjava3.testsupport.TestHelper; @@ -79,53 +78,66 @@ public void run() { } @Test - public void loadPurgeProperties() { - Properties props1 = new Properties(); - - PurgeProperties pp = new PurgeProperties(); - pp.load(props1); - - assertTrue(pp.purgeEnable); - assertEquals(pp.purgePeriod, 1); + public void boolPropertiesDisabledReturnsDefaultDisabled() throws Throwable { + assertTrue(SchedulerPoolFactory.getBooleanProperty(false, "key", false, true, failingPropertiesAccessor)); + assertFalse(SchedulerPoolFactory.getBooleanProperty(false, "key", true, false, failingPropertiesAccessor)); } @Test - public void loadPurgePropertiesDisabled() { - Properties props1 = new Properties(); - props1.setProperty(SchedulerPoolFactory.PURGE_ENABLED_KEY, "false"); + public void boolPropertiesEnabledMissingReturnsDefaultMissing() throws Throwable { + assertTrue(SchedulerPoolFactory.getBooleanProperty(true, "key", true, false, missingPropertiesAccessor)); + assertFalse(SchedulerPoolFactory.getBooleanProperty(true, "key", false, true, missingPropertiesAccessor)); + } - PurgeProperties pp = new PurgeProperties(); - pp.load(props1); + @Test + public void boolPropertiesFailureReturnsDefaultMissing() throws Throwable { + assertTrue(SchedulerPoolFactory.getBooleanProperty(true, "key", true, false, failingPropertiesAccessor)); + assertFalse(SchedulerPoolFactory.getBooleanProperty(true, "key", false, true, failingPropertiesAccessor)); + } - assertFalse(pp.purgeEnable); - assertEquals(pp.purgePeriod, 1); + @Test + public void boolPropertiesReturnsValue() throws Throwable { + assertTrue(SchedulerPoolFactory.getBooleanProperty(true, "true", true, false, Functions.identity())); + assertFalse(SchedulerPoolFactory.getBooleanProperty(true, "false", false, true, Functions.identity())); } @Test - public void loadPurgePropertiesEnabledCustomPeriod() { - Properties props1 = new Properties(); - props1.setProperty(SchedulerPoolFactory.PURGE_ENABLED_KEY, "true"); - props1.setProperty(SchedulerPoolFactory.PURGE_PERIOD_SECONDS_KEY, "2"); + public void intPropertiesDisabledReturnsDefaultDisabled() throws Throwable { + assertEquals(-1, SchedulerPoolFactory.getIntProperty(false, "key", 0, -1, failingPropertiesAccessor)); + assertEquals(-1, SchedulerPoolFactory.getIntProperty(false, "key", 1, -1, failingPropertiesAccessor)); + } - PurgeProperties pp = new PurgeProperties(); - pp.load(props1); + @Test + public void intPropertiesEnabledMissingReturnsDefaultMissing() throws Throwable { + assertEquals(-1, SchedulerPoolFactory.getIntProperty(true, "key", -1, 0, missingPropertiesAccessor)); + assertEquals(-1, SchedulerPoolFactory.getIntProperty(true, "key", -1, 1, missingPropertiesAccessor)); + } - assertTrue(pp.purgeEnable); - assertEquals(pp.purgePeriod, 2); + @Test + public void intPropertiesFailureReturnsDefaultMissing() throws Throwable { + assertEquals(-1, SchedulerPoolFactory.getIntProperty(true, "key", -1, 0, failingPropertiesAccessor)); + assertEquals(-1, SchedulerPoolFactory.getIntProperty(true, "key", -1, 1, failingPropertiesAccessor)); } @Test - public void loadPurgePropertiesEnabledCustomPeriodNaN() { - Properties props1 = new Properties(); - props1.setProperty(SchedulerPoolFactory.PURGE_ENABLED_KEY, "true"); - props1.setProperty(SchedulerPoolFactory.PURGE_PERIOD_SECONDS_KEY, "abc"); + public void intPropertiesReturnsValue() throws Throwable { + assertEquals(1, SchedulerPoolFactory.getIntProperty(true, "1", 0, 4, Functions.identity())); + assertEquals(2, SchedulerPoolFactory.getIntProperty(true, "2", 3, 5, Functions.identity())); + } - PurgeProperties pp = new PurgeProperties(); - pp.load(props1); + static final Function failingPropertiesAccessor = new Function() { + @Override + public String apply(String v) throws Throwable { + throw new SecurityException(); + } + }; - assertTrue(pp.purgeEnable); - assertEquals(pp.purgePeriod, 1); - } + static final Function missingPropertiesAccessor = new Function() { + @Override + public String apply(String v) throws Throwable { + return null; + } + }; @Test public void putIntoPoolNoPurge() {