Merge pull request wildfly#17096 from pferraro/WFLY-18315

WFLY-18315 Optimize payload of distributed session metadata

clustering/ee/cache/pom.xml

@@ -84,6 +84,36 @@
             <artifactId>jboss-threads</artifactId>
             <scope>runtime</scope>
         </dependency>
+
+        <!-- Internal test dependencies -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>wildfly-clustering-marshalling-api</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>wildfly-clustering-marshalling-protostream</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>wildfly-clustering-marshalling-spi</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
+
+        <!-- External test dependencies -->
+        <dependency>
+            <groupId>com.squareup</groupId>
+            <artifactId>protoparser</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/function/FunctionSerializationContextInitializer.java

@@ -9,6 +9,8 @@
 import org.infinispan.protostream.SerializationContextInitializer;
 import org.kohsuke.MetaInfServices;
 import org.wildfly.clustering.marshalling.protostream.AbstractSerializationContextInitializer;
+import org.wildfly.clustering.marshalling.protostream.FunctionalScalarMarshaller;
+import org.wildfly.clustering.marshalling.protostream.Scalar;
 
 /**
  * @author Paul Ferraro
@@ -21,5 +23,6 @@ public void registerMarshallers(SerializationContext context) {
         context.registerMarshaller(new MapComputeFunctionMarshaller());
         context.registerMarshaller(new CollectionFunctionMarshaller<>(SetAddFunction.class, SetAddFunction::new));
         context.registerMarshaller(new CollectionFunctionMarshaller<>(SetRemoveFunction.class, SetRemoveFunction::new));
+        context.registerMarshaller(new FunctionalScalarMarshaller<>(RemappingFunction.class, Scalar.ANY, RemappingFunction::getOperand, RemappingFunction::new));
     }
 }

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/function/Remappable.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.clustering.ee.cache.function;
+
+/**
+ * Implemented by cache value types that support copy-on-write operations.
+ * @author Paul Ferraro
+ * @param <V> the cache value type
+ * @param <O> the operand type
+ */
+public interface Remappable<V extends Remappable<V, O>, O> {
+
+    /**
+     * Returns a new instance of this object with the specified operand applied.
+     * @param operand a value applied to the replacement value
+     * @return a replacement value
+     */
+    V remap(O operand);
+}

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/function/RemappingFunction.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.clustering.ee.cache.function;
+
+import java.util.function.BiFunction;
+
+/**
+ * Generic function for use with {@link java.util.Map#compute(Object, BiFunction)} operations using {@link Remappable} values.
+ * @author Paul Ferraro
+ * @param <V> the cache value type
+ * @param <O> the operand type
+ */
+public class RemappingFunction<V extends Remappable<V, O>, O> implements BiFunction<Object, V, V> {
+
+    private final O operand;
+
+    public RemappingFunction(O operand) {
+        this.operand = operand;
+    }
+
+    public O getOperand() {
+        return this.operand;
+    }
+
+    @Override
+    public V apply(Object key, V value) {
+        return (value != null) ? value.remap(this.operand) : null;
+    }
+}

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/offset/Offset.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.clustering.ee.cache.offset;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.function.BiFunction;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
+
+/**
+ * Encapsulates an offset that can be applied to a value.
+ * @author Paul Ferraro
+ * @param <V> the value to which the offset can be applied
+ */
+public interface Offset<V> extends UnaryOperator<V> {
+    /**
+     * Returns true if this offset is zero, false otherwise.
+     * @return true if this offset is zero, false otherwise.
+     */
+    boolean isZero();
+
+    static Offset<Duration> forDuration(Duration offset) {
+        return offset.isZero() ? DurationOffset.ZERO : new DurationOffset(offset);
+    }
+
+    static Offset<Instant> forInstant(Duration offset) {
+        return offset.isZero() ? InstantOffset.ZERO : new InstantOffset(offset);
+    }
+
+    class DefaultOffset<O, V> implements Offset<V>, Supplier<O> {
+
+        private final O value;
+        private final Predicate<O> isZero;
+        private final BiFunction<V, O, V> applicator;
+
+        DefaultOffset(O value, Predicate<O> isZero, BiFunction<V, O, V> applicator) {
+            this.value = value;
+            this.isZero = isZero;
+            this.applicator = applicator;
+        }
+
+        @Override
+        public V apply(V value) {
+            return this.isZero() ? value : this.applicator.apply(value, this.value);
+        }
+
+        @Override
+        public boolean isZero() {
+            return this.isZero.test(this.value);
+        }
+
+        @Override
+        public O get() {
+            return this.value;
+        }
+
+        @Override
+        public int hashCode() {
+            return this.value.hashCode();
+        }
+
+        @Override
+        public boolean equals(Object object) {
+            if (!(object instanceof Offset)) return false;
+            @SuppressWarnings("unchecked")
+            DefaultOffset<O, V> offset = (DefaultOffset<O, V>) object;
+            return this.value.equals(offset.value);
+        }
+
+        @Override
+        public String toString() {
+            return this.value.toString();
+        }
+    }
+
+    class TemporalOffset<V> extends DefaultOffset<Duration, V> {
+        private static final Predicate<Duration> IS_ZERO = Duration::isZero;
+
+        TemporalOffset(Duration offset, BiFunction<V, Duration, V> applicator) {
+            super(offset, IS_ZERO, applicator);
+        }
+    }
+
+    class DurationOffset extends TemporalOffset<Duration> {
+        static final Offset<Duration> ZERO = new DurationOffset(Duration.ZERO);
+        private static final BiFunction<Duration, Duration, Duration> DURATION_APPLICATOR = Duration::plus;
+
+        DurationOffset(Duration value) {
+            super(value, DURATION_APPLICATOR);
+        }
+    }
+
+    class InstantOffset extends TemporalOffset<Instant> {
+        static final Offset<Instant> ZERO = new InstantOffset(Duration.ZERO);
+        private static final BiFunction<Instant, Duration, Instant> INSTANT_APPLICATOR = Instant::plus;
+
+        InstantOffset(Duration value) {
+            super(value, INSTANT_APPLICATOR);
+        }
+    }
+}

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/offset/OffsetSerializationContextInitializer.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.clustering.ee.cache.offset;
+
+import java.time.Duration;
+
+import org.infinispan.protostream.SerializationContext;
+import org.infinispan.protostream.SerializationContextInitializer;
+import org.kohsuke.MetaInfServices;
+import org.wildfly.clustering.marshalling.protostream.AbstractSerializationContextInitializer;
+import org.wildfly.clustering.marshalling.protostream.FunctionalMarshaller;
+
+/**
+ * @author Paul Ferraro
+ */
+@MetaInfServices(SerializationContextInitializer.class)
+public class OffsetSerializationContextInitializer extends AbstractSerializationContextInitializer {
+
+    @Override
+    public void registerMarshallers(SerializationContext context) {
+        context.registerMarshaller(new FunctionalMarshaller<>(Offset.DurationOffset.class, Duration.class, Offset.DurationOffset::get, Offset.DurationOffset::new));
+        context.registerMarshaller(new FunctionalMarshaller<>(Offset.InstantOffset.class, Duration.class, Offset.InstantOffset::get, Offset.InstantOffset::new));
+    }
+}

clustering/ee/cache/src/main/java/org/wildfly/clustering/ee/cache/offset/OffsetValue.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.clustering.ee.cache.offset;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Objects;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.wildfly.common.function.Functions;
+
+/**
+ * Encapsulates a value that is offset from some basis, and updated via {@link OffsetValue#set(Object)}.
+ * @author Paul Ferraro
+ * @param <V> the value type
+ */
+public interface OffsetValue<V> extends Value<V> {
+
+    /**
+     * Returns the basis from which the associated offset will be applied.
+     * @return the basis from which the associated offset will be applied.
+     */
+    V getBasis();
+
+    /**
+     * Returns the current value, computed by applying the current offset to the basis.
+     * @return the computed value
+     */
+    @Override
+    default V get() {
+        return this.getOffset().apply(this.getBasis());
+    }
+
+    /**
+     * The current offset.
+     * @return an offset
+     */
+    Offset<V> getOffset();
+
+    /**
+     * Sets the current offset.
+     * @param offset an offset
+     */
+    default void setOffset(Offset<V> offset) {
+        this.set(offset.apply(this.getBasis()));
+    }
+
+    /**
+     * Returns a new offset value based on the current value.
+     * @return a new offset value
+     */
+    OffsetValue<V> rebase();
+
+    static OffsetValue<Duration> from(Duration duration) {
+        return new DurationOffsetValue(Functions.constantSupplier(duration));
+    }
+
+    static OffsetValue<Instant> from(Instant instant) {
+        return new InstantOffsetValue(Functions.constantSupplier(instant));
+    }
+
+    class DefaultOffsetValue<O, V> extends AbstractValue<V> implements OffsetValue<V> {
+        private final BiFunction<V, V, O> factory;
+        private final Function<O, Offset<V>> offsetFactory;
+        private final Function<Supplier<V>, OffsetValue<V>> offsetValueFactory;
+        private final Supplier<V> basis;
+        private final O zero;
+
+        private volatile Offset<V> offset;
+
+        DefaultOffsetValue(Supplier<V> basis, O zero, BiFunction<V, V, O> factory, Function<O, Offset<V>> offsetFactory, Function<Supplier<V>, OffsetValue<V>> offsetValueFactory) {
+            this.factory = factory;
+            this.offsetFactory = offsetFactory;
+            this.offsetValueFactory = offsetValueFactory;
+            this.zero = zero;
+            this.basis = basis;
+            this.offset = this.offsetFactory.apply(zero);
+        }
+
+        @Override
+        public V getBasis() {
+            return this.basis.get();
+        }
+
+        @Override
+        public void set(V value) {
+            V basis = this.getBasis();
+            this.offset = this.offsetFactory.apply(Objects.equals(basis, value) ? this.zero : this.factory.apply(basis, value));
+        }
+
+        @Override
+        public void setOffset(Offset<V> offset) {
+            this.offset = offset;
+        }
+
+        @Override
+        public Offset<V> getOffset() {
+            return this.offset;
+        }
+
+        @Override
+        public OffsetValue<V> rebase() {
+            return this.offsetValueFactory.apply(this);
+        }
+    }
+
+    static class TemporalOffsetValue<V> extends DefaultOffsetValue<Duration, V> {
+
+        TemporalOffsetValue(Supplier<V> basis, BiFunction<V, V, Duration> factory, Function<Duration, Offset<V>> offsetFactory, Function<Supplier<V>, OffsetValue<V>> offsetValueFactory) {
+            super(basis, Duration.ZERO, factory, offsetFactory, offsetValueFactory);
+        }
+    }
+
+    static class DurationOffsetValue extends TemporalOffsetValue<Duration> {
+        private static final BiFunction<Duration, Duration, Duration> MINUS = Duration::minus;
+        private static final BiFunction<Duration, Duration, Duration> FACTORY = MINUS.andThen(Duration::negated);
+
+        DurationOffsetValue(Supplier<Duration> basis) {
+            super(basis, FACTORY, Offset::forDuration, DurationOffsetValue::new);
+        }
+    }
+
+    static class InstantOffsetValue extends TemporalOffsetValue<Instant> {
+        private static final BiFunction<Instant, Instant, Duration> FACTORY = Duration::between;
+
+        InstantOffsetValue(Supplier<Instant> basis) {
+            super(basis, FACTORY, Offset::forInstant, InstantOffsetValue::new);
+        }
+    }
+}