Issue2878

factcast-factus-redis/src/main/java/org/factcast/factus/redis/AbstractRedisManagedProjection.java

@@ -16,12 +16,11 @@
 package org.factcast.factus.redis;
 
 import lombok.NonNull;
-import org.factcast.factus.projection.ManagedProjection;
 import org.redisson.api.RedissonClient;
 
 public abstract class AbstractRedisManagedProjection extends AbstractRedisProjection
-    implements ManagedProjection {
-  public AbstractRedisManagedProjection(@NonNull RedissonClient redisson) {
+    implements RedisManagedProjection {
+  protected AbstractRedisManagedProjection(@NonNull RedissonClient redisson) {
     super(redisson);
   }
 }

factcast-factus-redis/src/main/java/org/factcast/factus/redis/AbstractRedisProjection.java

@@ -26,18 +26,17 @@
 import org.factcast.factus.projection.Named;
 import org.factcast.factus.projection.WriterToken;
 import org.factcast.factus.projection.WriterTokenAware;
-import org.factcast.factus.projection.tx.AbstractOpenTransactionAwareProjection;
-import org.factcast.factus.redis.tx.RedisTransactional;
 import org.redisson.api.*;
 
-abstract class AbstractRedisProjection extends AbstractOpenTransactionAwareProjection<RTransaction>
+@SuppressWarnings("java:S2142")
+public abstract class AbstractRedisProjection
     implements RedisProjection, FactStreamPositionAware, WriterTokenAware, Named {
   @Getter protected final RedissonClient redisson;
 
-  private final RLock lock;
-  private final String stateBucketName;
+  protected final RLock lock;
+  protected final String stateBucketName;
 
-  @Getter private final String redisKey;
+  @Getter protected final String redisKey;
 
   protected AbstractRedisProjection(@NonNull RedissonClient redisson) {
     this.redisson = redisson;
@@ -50,37 +49,23 @@ protected AbstractRedisProjection(@NonNull RedissonClient redisson) {
   }
 
   @VisibleForTesting
-  RBucket<FactStreamPosition> stateBucket(@NonNull RTransaction tx) {
-    return tx.getBucket(stateBucketName, FactStreamPositionCodec.INSTANCE);
-  }
-
-  @VisibleForTesting
-  RBucket<FactStreamPosition> stateBucket() {
+  protected RBucket<FactStreamPosition> stateBucket() {
     return redisson.getBucket(stateBucketName, FactStreamPositionCodec.INSTANCE);
   }
 
   @Override
   public FactStreamPosition factStreamPosition() {
-    if (inTransaction()) {
-      return stateBucket(runningTransaction()).get();
-    } else {
-      return stateBucket().get();
-    }
+    return stateBucket().get();
   }
 
   @SuppressWarnings("ConstantConditions")
   @Override
   public void factStreamPosition(@Nullable FactStreamPosition position) {
-    if (inTransaction()) {
-      stateBucket(runningTransaction()).set(position);
-    } else {
-      stateBucket().set(position);
-    }
+    stateBucket().set(position);
   }
 
   @Override
   public WriterToken acquireWriteToken(@NonNull Duration maxWait) {
-    assertNoRunningTransaction();
     try {
       if (lock.tryLock(maxWait.toMillis(), TimeUnit.MILLISECONDS))
         return new RedisWriterToken(redisson, lock);
@@ -89,33 +74,4 @@ public WriterToken acquireWriteToken(@NonNull Duration maxWait) {
     }
     return null;
   }
-
-  @Override
-  protected @NonNull RTransaction beginNewTransaction() {
-    return redisson().createTransaction(transactionOptions());
-  }
-
-  @Override
-  protected void commit(@NonNull RTransaction runningTransaction) {
-    runningTransaction.commit();
-  }
-
-  @Override
-  protected void rollback(@NonNull RTransaction runningTransaction) {
-    runningTransaction.rollback();
-  }
-
-  protected final @NonNull TransactionOptions transactionOptions() {
-    RedisTransactional tx = this.getClass().getAnnotation(RedisTransactional.class);
-    if (tx != null) return RedisTransactional.Defaults.with(tx);
-    else return TransactionOptions.defaults();
-  }
-
-  @Override
-  public final int maxBatchSizePerTransaction() {
-    RedisTransactional tx = this.getClass().getAnnotation(RedisTransactional.class);
-    if (tx != null) {
-      return tx.bulkSize();
-    } else return super.maxBatchSizePerTransaction();
-  }
 }

factcast-factus-redis/src/main/java/org/factcast/factus/redis/AbstractRedisSubscribedProjection.java

@@ -21,7 +21,7 @@
 
 public abstract class AbstractRedisSubscribedProjection extends AbstractRedisProjection
     implements SubscribedProjection {
-  public AbstractRedisSubscribedProjection(@NonNull RedissonClient redisson) {
+  protected AbstractRedisSubscribedProjection(@NonNull RedissonClient redisson) {
     super(redisson);
   }
 }

factcast-factus-redis/src/main/java/org/factcast/factus/redis/RedisManagedProjection.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright © 2017-2022 factcast.org
+ *
+ * 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.factcast.factus.redis;
+
+import org.factcast.factus.projection.ManagedProjection;
+
+public interface RedisManagedProjection extends ManagedProjection, RedisProjection {}

factcast-factus-redis/src/main/java/org/factcast/factus/redis/tx/AbstractRedisTxManagedProjection.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright © 2017-2022 factcast.org
+ *
+ * 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.factcast.factus.redis.tx;
+
+import lombok.NonNull;
+import org.factcast.factus.projection.ManagedProjection;
+import org.redisson.api.RedissonClient;
+
+public abstract class AbstractRedisTxManagedProjection extends AbstractRedisTxProjection
+    implements ManagedProjection {
+  protected AbstractRedisTxManagedProjection(@NonNull RedissonClient redisson) {
+    super(redisson);
+  }
+}

factcast-factus-redis/src/main/java/org/factcast/factus/redis/tx/AbstractRedisTxProjection.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright © 2017-2022 factcast.org
+ *
+ * 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.factcast.factus.redis.tx;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.time.Duration;
+import javax.annotation.Nullable;
+import lombok.NonNull;
+import lombok.experimental.Delegate;
+import org.factcast.core.FactStreamPosition;
+import org.factcast.factus.projection.WriterToken;
+import org.factcast.factus.projection.tx.OpenTransactionAware;
+import org.factcast.factus.projection.tx.TransactionBehavior;
+import org.factcast.factus.redis.AbstractRedisProjection;
+import org.factcast.factus.redis.FactStreamPositionCodec;
+import org.redisson.api.*;
+
+abstract class AbstractRedisTxProjection extends AbstractRedisProjection
+    implements OpenTransactionAware<RTransaction> {
+
+  @Delegate private final TransactionBehavior<RTransaction> tx;
+
+  protected AbstractRedisTxProjection(@NonNull RedissonClient redisson) {
+    super(redisson);
+    this.tx =
+        new TransactionBehavior<>(
+            new RedisTxAdapter(redisson, getClass().getAnnotation(RedisTransactional.class)));
+  }
+
+  @VisibleForTesting
+  protected RBucket<FactStreamPosition> stateBucket(@NonNull RTransaction tx) {
+    return tx.getBucket(stateBucketName, FactStreamPositionCodec.INSTANCE);
+  }
+
+  @Override
+  public FactStreamPosition factStreamPosition() {
+    if (inTransaction()) {
+      return stateBucket(runningTransaction()).get();
+    } else {
+      return super.factStreamPosition();
+    }
+  }
+
+  @SuppressWarnings("ConstantConditions")
+  @Override
+  public void factStreamPosition(@Nullable FactStreamPosition position) {
+    if (inTransaction()) {
+      stateBucket(runningTransaction()).set(position);
+    } else {
+      super.factStreamPosition(position);
+    }
+  }
+
+  @Override
+  public WriterToken acquireWriteToken(@NonNull Duration maxWait) {
+    assertNoRunningTransaction();
+    return super.acquireWriteToken(maxWait);
+  }
+
+  @Override
+  public AutoCloseable acquireWriteToken() {
+    assertNoRunningTransaction();
+    return super.acquireWriteToken();
+  }
+}

factcast-factus-redis/src/main/java/org/factcast/factus/redis/tx/AbstractRedisTxSubscribedProjection.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright © 2017-2022 factcast.org
+ *
+ * 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.factcast.factus.redis.tx;
+
+import lombok.NonNull;
+import org.factcast.factus.projection.SubscribedProjection;
+import org.redisson.api.RedissonClient;
+
+public abstract class AbstractRedisTxSubscribedProjection extends AbstractRedisTxProjection
+    implements SubscribedProjection {
+  protected AbstractRedisTxSubscribedProjection(@NonNull RedissonClient redisson) {
+    super(redisson);
+  }
+}

factcast-factus-redis/src/main/java/org/factcast/factus/redis/tx/RedisTransactional.java

@@ -18,13 +18,17 @@
 import java.lang.annotation.*;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nullable;
+import lombok.experimental.UtilityClass;
 import org.redisson.api.TransactionOptions;
 
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
 @Inherited
 public @interface RedisTransactional {
-  int bulkSize() default 1000;
+
+  int DEFAULT_BULK_SIZE = 1000;
+
+  int bulkSize() default DEFAULT_BULK_SIZE;
 
   long timeout() default Defaults.timeout;
 
@@ -34,6 +38,7 @@
 
   long retryInterval() default Defaults.retryInterval;
 
+  @UtilityClass
   class Defaults {
     static final long timeout = 30000;
     static final long responseTimeout = 5001;

factcast-factus-redis/src/main/java/org/factcast/factus/redis/tx/RedisTxAdapter.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright © 2017-2024 factcast.org
+ *
+ * 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.factcast.factus.redis.tx;
+
+import com.google.common.annotations.VisibleForTesting;
+import javax.annotation.Nullable;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import org.factcast.factus.projection.tx.TransactionAdapter;
+import org.redisson.api.RTransaction;
+import org.redisson.api.RedissonClient;
+import org.redisson.api.TransactionOptions;
+
+@RequiredArgsConstructor
+public class RedisTxAdapter implements TransactionAdapter<RTransaction> {
+  @NonNull private final RedissonClient client;
+  @Nullable private final RedisTransactional annotation;
+
+  @Override
+  @NonNull
+  public RTransaction beginNewTransaction() {
+    return client.createTransaction(transactionOptions());
+  }
+
+  @VisibleForTesting
+  @NonNull
+  public TransactionOptions transactionOptions() {
+    if (annotation != null) return RedisTransactional.Defaults.with(annotation);
+    else return TransactionOptions.defaults();
+  }
+
+  @Override
+  public void commit(@NonNull RTransaction runningTransaction) {
+    runningTransaction.commit();
+  }
+
+  @Override
+  public void rollback(@NonNull RTransaction runningTransaction) {
+    runningTransaction.rollback();
+  }
+
+  @Override
+  public int maxBatchSizePerTransaction() {
+    if (annotation != null) {
+      return annotation.bulkSize();
+    } else return RedisTransactional.DEFAULT_BULK_SIZE;
+  }
+}

factcast-factus-redis/src/test/java/org/factcast/factus/redis/ARedisTransactionalManagedProjection.java → factcast-factus-redis/src/test/java/org/factcast/factus/redis/tx/ARedisTxManagedProjection.java

@@ -13,15 +13,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.factcast.factus.redis;
+package org.factcast.factus.redis.tx;
 
 import lombok.NonNull;
 import org.factcast.factus.serializer.ProjectionMetaData;
 import org.redisson.api.RedissonClient;
 
 @ProjectionMetaData(revision = 1)
-public class ARedisTransactionalManagedProjection extends AbstractRedisManagedProjection {
-  public ARedisTransactionalManagedProjection(@NonNull RedissonClient redisson) {
+public class ARedisTxManagedProjection extends AbstractRedisTxManagedProjection {
+  public ARedisTxManagedProjection(@NonNull RedissonClient redisson) {
     super(redisson);
   }
 }