@W-17808637 Improving code coverage for core/internal module

core-tests/src/test/java/org/mule/runtime/core/internal/util/java/JdkVersionUtilsTestCase.java → core-tests/src/test/java/org/mule/runtime/core/internal/util/version/JdkVersionUtilsTestCase.java

@@ -4,10 +4,10 @@
  * license, a copy of which has been included with this distribution in the
  * LICENSE.txt file.
  */
-package org.mule.runtime.core.internal.util.java;
+package org.mule.runtime.core.internal.util.version;
 
-import static org.mule.runtime.core.internal.util.java.JdkVersionUtils.isRecommendedJdkVersion;
-import static org.mule.runtime.core.internal.util.java.JdkVersionUtils.isSupportedJdkVersion;
+import static org.mule.runtime.core.internal.util.version.JdkVersionUtils.isRecommendedJdkVersion;
+import static org.mule.runtime.core.internal.util.version.JdkVersionUtils.isSupportedJdkVersion;
 import static org.mule.runtime.manifest.api.MuleManifest.getMuleManifest;
 import static org.mule.test.allure.AllureConstants.SupportedEnvironmentsFeature.SUPPORTED_ENVIRONMENTS;
 import static org.mule.test.allure.AllureConstants.SupportedEnvironmentsFeature.JdkVersionStory.JDK_VERSION;
@@ -26,8 +26,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import org.mule.runtime.core.internal.util.java.JdkVersionUtils.JdkVersion;
-import org.mule.runtime.core.internal.util.version.JdkVersionUtils;
+import org.mule.runtime.core.internal.util.version.JdkVersionUtils.JdkVersion;
 import org.mule.runtime.manifest.internal.DefaultMuleManifest;
 import org.mule.tck.junit4.AbstractMuleTestCase;
 

core-tests/src/test/java/org/mule/runtime/core/internal/util/version/VersionRangeTestCase.java

@@ -8,12 +8,18 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
 
 import org.mule.tck.junit4.AbstractMuleTestCase;
 
 import org.junit.Test;
 
+import java.util.List;
+
 public class VersionRangeTestCase extends AbstractMuleTestCase {
 
   @Test
@@ -71,6 +77,79 @@ public void invalidRange() {
     assertThrows(IllegalArgumentException.class, () -> new VersionRange("saraza"));
   }
 
+  @Test
+  public void testSingleVersionRange() {
+    VersionRange vr = new VersionRange("[1.5.0_11,1.6)");
+    assertNotNull(vr);
+    assertTrue(vr.isLowerBoundInclusive());
+    assertFalse(vr.isUpperBoundInclusive());
+    assertEquals("1.5.0_11", vr.getLowerVersion());
+    assertEquals("1.6", vr.getUpperVersion());
+
+    vr = new VersionRange("(1.5.0_11-b05,2.7.12]");
+    assertNotNull(vr);
+    assertFalse(vr.isLowerBoundInclusive());
+    assertTrue(vr.isUpperBoundInclusive());
+    assertEquals("1.5.0_11-b05", vr.getLowerVersion());
+    assertEquals("2.7.12", vr.getUpperVersion());
+  }
+
+  @Test
+  public void testCreateVersionRanges() {
+    List<VersionRange> ranges = VersionRange.createVersionRanges("(,1.4.2),[1.5.0_11,1.6),[1.7,]");
+    assertNotNull(ranges);
+    assertEquals(3, ranges.size());
+
+    VersionRange vr = ranges.get(0);
+    assertFalse(vr.isLowerBoundInclusive());
+    assertFalse(vr.isUpperBoundInclusive());
+    assertEquals("", vr.getLowerVersion());
+    assertEquals("1.4.2", vr.getUpperVersion());
+
+    vr = ranges.get(1);
+    assertTrue(vr.isLowerBoundInclusive());
+    assertFalse(vr.isUpperBoundInclusive());
+    assertEquals("1.5.0_11", vr.getLowerVersion());
+    assertEquals("1.6", vr.getUpperVersion());
+
+    vr = ranges.get(2);
+    assertTrue(vr.isLowerBoundInclusive());
+    assertTrue(vr.isUpperBoundInclusive());
+    assertEquals("1.7", vr.getLowerVersion());
+    assertEquals("", vr.getUpperVersion());
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testInvalidDelimiter() {
+    VersionRange range = new VersionRange("{1.3,1.4.2)");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testMissingDelimiter() {
+    VersionRange range = new VersionRange("1.3,1.4.2)");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testInvalidVersion() {
+    VersionRange range = new VersionRange("[1.3,0,1.4.2)");
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testInvalidCreateVersionRanges() {
+    List<VersionRange> ranges = VersionRange.createVersionRanges("(,1.4.2),1.5.0_11,1.6),[1.7,]");
+    for (VersionRange vr : ranges) {
+      System.out.println(vr);
+    }
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testInvalidDelimiterCreateVersionRanges() {
+    List<VersionRange> ranges = VersionRange.createVersionRanges("(,1.4.2)|[1.5.0_11,1.6)|[1.7,]");
+    for (VersionRange vr : ranges) {
+      System.out.println(vr);
+    }
+  }
+
   private void assertRangeContainsVersion(final VersionRange range, String version, boolean expected) {
     assertThat("`" + version + "` " + (expected ? "not " : " ") + "contained in " + range.toString(),
                range.contains(version), is(expected));

core/pom.xml

@@ -216,5 +216,38 @@
             <groupId>net.bytebuddy</groupId>
             <artifactId>byte-buddy</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-engine</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.hamcrest</groupId>
+            <artifactId>hamcrest</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

core/src/main/java/org/mule/runtime/core/internal/processor/strategy/AbstractProcessingStrategy.java

@@ -81,29 +81,6 @@ public void setSchedulerDecorator(Function<ScheduledExecutorService, ScheduledEx
     this.schedulerDecorator = schedulerDecorator;
   }
 
-  /**
-   * Checks whether an error indicates that a thread pool is full.
-   *
-   * @param t the thrown error to analyze
-   * @return {@code true} if {@code t} indicates that a thread pool needed for the processing strategy owner rejected a task.
-   */
-  protected boolean isSchedulerBusy(Throwable t) {
-    final Throwable cause = unwrap(t);
-    return RejectedExecutionException.class.isAssignableFrom(cause.getClass()) || isOverloadError(cause);
-  }
-
-  private boolean isOverloadError(final Throwable cause) {
-    if (cause instanceof MessagingException) {
-      return ((MessagingException) cause).getEvent().getError()
-          .map(e -> e.getErrorType())
-          .filter(errorType -> OVERLOAD.getName().equals(errorType.getIdentifier())
-              && OVERLOAD.getNamespace().equals(errorType.getNamespace()))
-          .isPresent();
-    } else {
-      return false;
-    }
-  }
-
   /**
    * Extension of {@link Sink} using Reactor's {@link FluxSink} to accept events.
    */

core/src/main/java/org/mule/runtime/core/internal/processor/strategy/AbstractReactorStreamProcessingStrategy.java

@@ -6,6 +6,7 @@
  */
 package org.mule.runtime.core.internal.processor.strategy;
 
+import static java.lang.Thread.currentThread;
 import static org.mule.runtime.core.api.construct.BackPressureReason.MAX_CONCURRENCY_EXCEEDED;
 import static org.mule.runtime.core.api.processor.ReactiveProcessor.ProcessingType.CPU_LITE;
 import static org.mule.runtime.core.api.processor.ReactiveProcessor.ProcessingType.CPU_LITE_ASYNC;
@@ -70,7 +71,7 @@ public abstract class AbstractReactorStreamProcessingStrategy extends AbstractSt
                                           int parallelism,
                                           int maxConcurrency,
                                           boolean maxConcurrencyEagerCheck) {
-    super(subscribers, maxConcurrency, maxConcurrencyEagerCheck);
+    super(subscribers, maxConcurrency, maxConcurrencyEagerCheck, () -> currentThread());
     this.cpuLightSchedulerSupplier = cpuLightSchedulerSupplier;
     this.parallelism = parallelism;
   }

core/src/main/java/org/mule/runtime/core/internal/processor/strategy/AbstractStreamProcessingStrategyFactory.java

@@ -125,12 +125,15 @@ abstract static class AbstractStreamProcessingStrategy extends AbstractProcessin
     final protected int maxConcurrency;
     final protected boolean maxConcurrencyEagerCheck;
     final protected ClassLoader executionClassloader;
+    private final Supplier<Thread> threadSupplier;
 
-    protected AbstractStreamProcessingStrategy(int subscribers, int maxConcurrency, boolean maxConcurrencyEagerCheck) {
+    protected AbstractStreamProcessingStrategy(int subscribers, int maxConcurrency, boolean maxConcurrencyEagerCheck,
+                                               Supplier<Thread> threadSupplier) {
       this.subscribers = requireNonNull(subscribers);
       this.maxConcurrency = requireNonNull(maxConcurrency);
       this.maxConcurrencyEagerCheck = maxConcurrency < MAX_VALUE && maxConcurrencyEagerCheck;
-      this.executionClassloader = currentThread().getContextClassLoader();
+      this.executionClassloader = threadSupplier.get().getContextClassLoader();
+      this.threadSupplier = threadSupplier;
     }
 
     protected void awaitSubscribersCompletion(FlowConstruct flowConstruct, final long shutdownTimeout,
@@ -147,7 +150,7 @@ protected void awaitSubscribersCompletion(FlowConstruct flowConstruct, final lon
           }
         }
       } catch (InterruptedException e) {
-        currentThread().interrupt();
+        threadSupplier.get().interrupt();
         if (getProperty(MULE_LIFECYCLE_FAIL_ON_FIRST_DISPOSE_ERROR) != null) {
           throw new IllegalStateException(format("Subscribers of ProcessingStrategy for flow '%s' not completed before thread interruption",
                                                  flowConstruct.getName()));

core/src/main/java/org/mule/runtime/core/internal/processor/strategy/DirectProcessingStrategyFactory.java

@@ -15,8 +15,6 @@
 import org.mule.runtime.core.api.processor.Sink;
 import org.mule.runtime.core.api.processor.strategy.ProcessingStrategy;
 import org.mule.runtime.core.api.processor.strategy.ProcessingStrategyFactory;
-import org.mule.runtime.core.internal.processor.strategy.AbstractProcessingStrategy;
-import org.mule.runtime.core.internal.processor.strategy.StreamPerEventSink;
 
 /**
  * Processing strategy that processes the {@link Pipeline} in the caller thread and does not schedule the processing of any

core/src/main/java/org/mule/runtime/core/internal/processor/strategy/DirectSink.java

@@ -40,8 +40,11 @@ class DirectSink implements Sink, Disposable {
   public DirectSink(Function<Publisher<CoreEvent>, Publisher<CoreEvent>> function,
                     Consumer<CoreEvent> eventConsumer, int bufferSize) {
     EmitterProcessor<CoreEvent> emitterProcessor = EmitterProcessor.create(bufferSize);
-    final reactor.core.Disposable disposable = emitterProcessor.transform(function).doOnError(throwable -> {
-    }).subscribe();
+    final reactor.core.Disposable disposable = emitterProcessor
+        .transform(function)
+        .doOnError(throwable -> {
+        })
+        .subscribe();
     reactorSink = new DefaultReactorSink(emitterProcessor.sink(), millis -> disposable.dispose(), eventConsumer, bufferSize);
   }