From 79bbe090a2cbaaf9e3eb8c9cf8b5837a49ee2c92 Mon Sep 17 00:00:00 2001
From: Maxim Thomas <maxim.thomas@gmail.com>
Date: Sat, 28 Jun 2025 07:47:33 +0000
Subject: [PATCH] Fix OSGI bundle excluded package error for rxjava3 (#524)
---
opendj-grizzly/src/main/java/org/openidentityplatform/rxjava3/internal/util/BackpressureHelper.java | 153 +++++++++++++++++++++++++++++++++++++++++++++++++++
opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java | 2
pom.xml | 2
3 files changed, 155 insertions(+), 2 deletions(-)
diff --git a/opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java b/opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java
index 5dd6e09..d557f6c 100644
--- a/opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java
+++ b/opendj-grizzly/src/main/java/org/forgerock/opendj/grizzly/LDAPServerFilter.java
@@ -86,7 +86,7 @@
import com.forgerock.reactive.Stream;
import io.reactivex.rxjava3.exceptions.OnErrorNotImplementedException;
-import io.reactivex.rxjava3.internal.util.BackpressureHelper;
+import org.openidentityplatform.rxjava3.internal.util.BackpressureHelper;
/**
* Grizzly filter implementation for decoding LDAP requests and handling server side logic for SSL and SASL operations
diff --git a/opendj-grizzly/src/main/java/org/openidentityplatform/rxjava3/internal/util/BackpressureHelper.java b/opendj-grizzly/src/main/java/org/openidentityplatform/rxjava3/internal/util/BackpressureHelper.java
new file mode 100644
index 0000000..2d3c0d4
--- /dev/null
+++ b/opendj-grizzly/src/main/java/org/openidentityplatform/rxjava3/internal/util/BackpressureHelper.java
@@ -0,0 +1,153 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package org.openidentityplatform.rxjava3.internal.util;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.reactivex.rxjava3.annotations.NonNull;
+import io.reactivex.rxjava3.plugins.RxJavaPlugins;
+
+/**
+ * Utility class to help with backpressure-related operations such as request aggregation.
+ */
+public final class BackpressureHelper {
+ /** Utility class. */
+ private BackpressureHelper() {
+ throw new IllegalStateException("No instances!");
+ }
+
+ /**
+ * Adds two long values and caps the sum at {@link Long#MAX_VALUE}.
+ * @param a the first value
+ * @param b the second value
+ * @return the sum capped at {@link Long#MAX_VALUE}
+ */
+ public static long addCap(long a, long b) {
+ long u = a + b;
+ if (u < 0L) {
+ return Long.MAX_VALUE;
+ }
+ return u;
+ }
+
+ /**
+ * Multiplies two long values and caps the product at {@link Long#MAX_VALUE}.
+ * @param a the first value
+ * @param b the second value
+ * @return the product capped at {@link Long#MAX_VALUE}
+ */
+ public static long multiplyCap(long a, long b) {
+ long u = a * b;
+ if (((a | b) >>> 31) != 0) {
+ if (u / a != b) {
+ return Long.MAX_VALUE;
+ }
+ }
+ return u;
+ }
+
+ /**
+ * Atomically adds the positive value n to the requested value in the {@link AtomicLong} and
+ * caps the result at {@link Long#MAX_VALUE} and returns the previous value.
+ * @param requested the {@code AtomicLong} holding the current requested value
+ * @param n the value to add, must be positive (not verified)
+ * @return the original value before the add
+ */
+ public static long add(@NonNull AtomicLong requested, long n) {
+ for (;;) {
+ long r = requested.get();
+ if (r == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+ long u = addCap(r, n);
+ if (requested.compareAndSet(r, u)) {
+ return r;
+ }
+ }
+ }
+
+ /**
+ * Atomically adds the positive value n to the requested value in the {@link AtomicLong} and
+ * caps the result at {@link Long#MAX_VALUE} and returns the previous value and
+ * considers {@link Long#MIN_VALUE} as a cancel indication (no addition then).
+ * @param requested the {@code AtomicLong} holding the current requested value
+ * @param n the value to add, must be positive (not verified)
+ * @return the original value before the add
+ */
+ public static long addCancel(@NonNull AtomicLong requested, long n) {
+ for (;;) {
+ long r = requested.get();
+ if (r == Long.MIN_VALUE) {
+ return Long.MIN_VALUE;
+ }
+ if (r == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+ long u = addCap(r, n);
+ if (requested.compareAndSet(r, u)) {
+ return r;
+ }
+ }
+ }
+
+ /**
+ * Atomically subtract the given number (positive, not validated) from the target field unless it contains {@link Long#MAX_VALUE}.
+ * @param requested the target field holding the current requested amount
+ * @param n the produced element count, positive (not validated)
+ * @return the new amount
+ */
+ public static long produced(@NonNull AtomicLong requested, long n) {
+ for (;;) {
+ long current = requested.get();
+ if (current == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+ long update = current - n;
+ if (update < 0L) {
+ RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
+ update = 0L;
+ }
+ if (requested.compareAndSet(current, update)) {
+ return update;
+ }
+ }
+ }
+
+ /**
+ * Atomically subtract the given number (positive, not validated) from the target field if
+ * it doesn't contain {@link Long#MIN_VALUE} (indicating some cancelled state) or {@link Long#MAX_VALUE} (unbounded mode).
+ * @param requested the target field holding the current requested amount
+ * @param n the produced element count, positive (not validated)
+ * @return the new amount
+ */
+ public static long producedCancel(@NonNull AtomicLong requested, long n) {
+ for (;;) {
+ long current = requested.get();
+ if (current == Long.MIN_VALUE) {
+ return Long.MIN_VALUE;
+ }
+ if (current == Long.MAX_VALUE) {
+ return Long.MAX_VALUE;
+ }
+ long update = current - n;
+ if (update < 0L) {
+ RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
+ update = 0L;
+ }
+ if (requested.compareAndSet(current, update)) {
+ return update;
+ }
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 5053002..98910b0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -35,7 +35,7 @@
<product.locales>ca_ES,es,de,fr,ja,ko,pl,zh_CN,zh_TW</product.locales>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<localized.jars.classifier>i18n</localized.jars.classifier>
- <commons.version>2.3.0</commons.version>
+ <commons.version>2.3.1-SNAPSHOT</commons.version>
<freemarker.version>2.3.34</freemarker.version>
<metrics-core.version>4.2.30</metrics-core.version>
<maven.compiler.target>8</maven.compiler.target>
--
Gitblit v1.10.0