[release-2.2] Fix the problem that publishing post always fails

src/main/java/run/halo/app/core/extension/Plugin.java

@@ -10,8 +10,6 @@
 import java.util.Map;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.Setter;
 import lombok.ToString;
 import org.pf4j.PluginState;
 import org.springframework.lang.NonNull;
@@ -93,8 +91,7 @@ public static class PluginSpec {
         private String configMapName;
     }
 
-    @Getter
-    @Setter
+    @Data
     public static class License {
         private String name;
         private String url;

src/main/java/run/halo/app/core/extension/Role.java

@@ -53,6 +53,7 @@ public class Role extends AbstractExtension {
      * @since 2.0.0
      */
     @Getter
+    @EqualsAndHashCode
     public static class PolicyRule implements Comparable<PolicyRule> {
         /**
          * APIGroups is the name of the APIGroup that contains the resources.

src/main/java/run/halo/app/extension/Metadata.java

@@ -12,7 +12,7 @@
  * @author johnniang
  */
 @Data
-@EqualsAndHashCode
+@EqualsAndHashCode(exclude = "version")
 public class Metadata implements MetadataOperator {
 
     /**

src/main/java/run/halo/app/extension/ReactiveExtensionClientImpl.java

@@ -7,6 +7,7 @@
 import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Objects;
 import java.util.function.Predicate;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.util.Predicates;
@@ -136,20 +137,25 @@ public <E extends Extension> Mono<E> update(E extension) {
             mono = get(extension.getClass(), extension.getMetadata().getName());
         }
         return mono
-            .map(old -> {
+            .flatMap(old -> {
                 // reset some fields
                 var oldMetadata = old.getMetadata();
                 var newMetadata = extension.getMetadata();
                 newMetadata.setCreationTimestamp(oldMetadata.getCreationTimestamp());
                 newMetadata.setDeletionTimestamp(oldMetadata.getDeletionTimestamp());
                 extension.setMetadata(newMetadata);
-                return converter.convertTo(extension);
+                if (Objects.equals(old, extension)) {
+                    return Mono.empty();
+                }
+                return Mono.just(extension);
             })
+            .map(converter::convertTo)
             .flatMap(extensionStore -> client.update(extensionStore.getName(),
                 extensionStore.getVersion(),
                 extensionStore.getData()))
             .map(updated -> converter.convertFrom((Class<E>) extension.getClass(), updated))
-            .doOnNext(updated -> watchers.onUpdate(extension, updated));
+            .doOnNext(updated -> watchers.onUpdate(extension, updated))
+            .switchIfEmpty(Mono.defer(() -> Mono.just(extension)));
     }
 
     @Override

src/main/java/run/halo/app/extension/Unstructured.java

@@ -22,6 +22,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import lombok.EqualsAndHashCode;
 
 /**
  * Unstructured is a generic Extension, which wraps ObjectNode to maintain the Extension data, like
@@ -65,6 +66,7 @@ public MetadataOperator getMetadata() {
         return new UnstructuredMetadata();
     }
 
+    @EqualsAndHashCode(exclude = "version")
     class UnstructuredMetadata implements MetadataOperator {
 
         @Override

src/main/java/run/halo/app/extension/controller/ControllerBuilder.java

@@ -106,7 +106,7 @@ public Controller build() {
         Assert.notNull(extension, "Extension must not be null");
         Assert.notNull(reconciler, "Reconciler must not be null");
 
-        var queue = new DefaultDelayQueue<Request>(nowSupplier, minDelay);
+        var queue = new DefaultQueue<Request>(nowSupplier, minDelay);
         var predicates = new WatcherPredicates.Builder()
             .withGroupVersionKind(extension.groupVersionKind())
             .onAddPredicate(onAddPredicate)

src/main/java/run/halo/app/extension/controller/DefaultQueue.java

@@ -0,0 +1,141 @@
+package run.halo.app.extension.controller;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class DefaultQueue<R> implements RequestQueue<R> {
+
+    private final Lock lock;
+
+    private final DelayQueue<DelayedEntry<R>> queue;
+
+    private final Supplier<Instant> nowSupplier;
+
+    private volatile boolean disposed = false;
+
+    private final Duration minDelay;
+
+    private final Set<R> processing;
+
+    private final Set<R> dirty;
+
+    public DefaultQueue(Supplier<Instant> nowSupplier) {
+        this(nowSupplier, Duration.ZERO);
+    }
+
+    public DefaultQueue(Supplier<Instant> nowSupplier, Duration minDelay) {
+        this.lock = new ReentrantLock();
+        this.nowSupplier = nowSupplier;
+        this.minDelay = minDelay;
+        this.processing = new HashSet<>();
+        this.dirty = new HashSet<>();
+        this.queue = new DelayQueue<>();
+    }
+
+    @Override
+    public boolean addImmediately(R request) {
+        log.debug("Adding request {} immediately", request);
+        var delayedEntry = new DelayedEntry<>(request, minDelay, nowSupplier);
+        return add(delayedEntry);
+    }
+
+    @Override
+    public boolean add(DelayedEntry<R> entry) {
+        lock.lock();
+        try {
+            if (isDisposed()) {
+                return false;
+            }
+            log.debug("Adding request {} after {}", entry.getEntry(), entry.getRetryAfter());
+            if (entry.getRetryAfter().compareTo(minDelay) < 0) {
+                log.warn("Request {} will be retried after {} ms, but minimum delay is {} ms",
+                    entry.getEntry(), entry.getRetryAfter().toMillis(), minDelay.toMillis());
+                entry = new DelayedEntry<>(entry.getEntry(), minDelay, nowSupplier);
+            }
+            if (dirty.contains(entry.getEntry())) {
+                return false;
+            }
+            dirty.add(entry.getEntry());
+            if (processing.contains(entry.getEntry())) {
+                return false;
+            }
+
+            boolean added = queue.add(entry);
+            log.debug("Added request {} after {}", entry.getEntry(), entry.getRetryAfter());
+            return added;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public DelayedEntry<R> take() throws InterruptedException {
+        var entry = queue.take();
+        log.debug("Take request {} at {}", entry.getEntry(), Instant.now());
+        lock.lockInterruptibly();
+        try {
+            if (isDisposed()) {
+                throw new InterruptedException(
+                    "Queue has been disposed. Cannot take any elements now");
+            }
+            processing.add(entry.getEntry());
+            dirty.remove(entry.getEntry());
+            return entry;
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public void done(R request) {
+        lock.lock();
+        try {
+            if (isDisposed()) {
+                return;
+            }
+            processing.remove(request);
+            if (dirty.contains(request)) {
+                queue.add(new DelayedEntry<>(request, minDelay, nowSupplier));
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public long size() {
+        return queue.size();
+    }
+
+    @Override
+    public DelayedEntry<R> peek() {
+        return queue.peek();
+    }
+
+    @Override
+    public void dispose() {
+        lock.lock();
+        try {
+            disposed = true;
+            queue.clear();
+            processing.clear();
+            dirty.clear();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public boolean isDisposed() {
+        return this.disposed;
+    }
+
+}

src/main/java/run/halo/app/extension/controller/RequestQueue.java

@@ -18,6 +18,10 @@ public interface RequestQueue<E> extends Disposable {
 
     void done(E request);
 
+    long size();
+
+    DelayedEntry<E> peek();
+
     class DelayedEntry<E> implements Delayed {
 
         private final E entry;

src/main/java/run/halo/app/extension/gc/GcReconciler.java

@@ -13,7 +13,7 @@
 import run.halo.app.extension.controller.Controller;
 import run.halo.app.extension.controller.ControllerBuilder;
 import run.halo.app.extension.controller.DefaultController;
-import run.halo.app.extension.controller.DefaultDelayQueue;
+import run.halo.app.extension.controller.DefaultQueue;
 import run.halo.app.extension.controller.Reconciler;
 import run.halo.app.extension.store.ExtensionStoreClient;
 
@@ -55,7 +55,7 @@ public Result reconcile(GcRequest request) {
 
     @Override
     public Controller setupWith(ControllerBuilder builder) {
-        var queue = new DefaultDelayQueue<GcRequest>(Instant::now, Duration.ofMillis(500));
+        var queue = new DefaultQueue<GcRequest>(Instant::now, Duration.ofMillis(500));
         var synchronizer = new GcSynchronizer(client, queue, schemeManager);
         return new DefaultController<>(
             "garbage-collector-controller",

src/main/java/run/halo/app/infra/ConditionList.java

@@ -1,9 +1,9 @@
 package run.halo.app.infra;
 
 import java.util.AbstractCollection;
-import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.Objects;
 import java.util.function.Consumer;
 import org.springframework.lang.NonNull;
@@ -20,7 +20,7 @@
  */
 public class ConditionList extends AbstractCollection<Condition> {
     private static final int EVICT_THRESHOLD = 20;
-    private final Deque<Condition> conditions = new ArrayDeque<>();
+    private final Deque<Condition> conditions = new LinkedList<>();
 
     @Override
     public boolean add(@NonNull Condition condition) {
@@ -113,4 +113,21 @@ public Iterator<Condition> iterator() {
     public void forEach(Consumer<? super Condition> action) {
         conditions.forEach(action);
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ConditionList that = (ConditionList) o;
+        return Objects.equals(conditions, that.conditions);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(conditions);
+    }
 }