mirror of https://github.com/halo-dev/halo
Merge remote-tracking branch 'upstream/main' into refactor/console-as-subproject
commit
29ec459eac
|
@ -26,7 +26,8 @@ import org.springframework.data.domain.Pageable;
|
|||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.data.util.Predicates;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.transaction.ReactiveTransactionManager;
|
||||
import org.springframework.transaction.reactive.TransactionalOperator;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.retry.Retry;
|
||||
|
@ -40,7 +41,6 @@ import run.halo.app.extension.store.ReactiveExtensionStoreClient;
|
|||
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
||||
|
||||
private final ReactiveExtensionStoreClient client;
|
||||
|
@ -60,6 +60,28 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
private final ConcurrentMap<GroupKind, AtomicBoolean> indexBuildingState =
|
||||
new ConcurrentHashMap<>();
|
||||
|
||||
private TransactionalOperator transactionalOperator;
|
||||
|
||||
public ReactiveExtensionClientImpl(ReactiveExtensionStoreClient client,
|
||||
ExtensionConverter converter, SchemeManager schemeManager, ObjectMapper objectMapper,
|
||||
IndexerFactory indexerFactory, IndexedQueryEngine indexedQueryEngine,
|
||||
ReactiveTransactionManager reactiveTransactionManager) {
|
||||
this.client = client;
|
||||
this.converter = converter;
|
||||
this.schemeManager = schemeManager;
|
||||
this.objectMapper = objectMapper;
|
||||
this.indexerFactory = indexerFactory;
|
||||
this.indexedQueryEngine = indexedQueryEngine;
|
||||
this.transactionalOperator = TransactionalOperator.create(reactiveTransactionManager);
|
||||
}
|
||||
|
||||
/**
|
||||
* Only for test.
|
||||
*/
|
||||
void setTransactionalOperator(TransactionalOperator transactionalOperator) {
|
||||
this.transactionalOperator = transactionalOperator;
|
||||
}
|
||||
|
||||
@Override
|
||||
public <E extends Extension> Flux<E> list(Class<E> type, Predicate<E> predicate,
|
||||
Comparator<E> comparator) {
|
||||
|
@ -151,7 +173,6 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public <E extends Extension> Mono<E> create(E extension) {
|
||||
checkClientWritable(extension);
|
||||
return Mono.just(extension)
|
||||
|
@ -185,7 +206,6 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public <E extends Extension> Mono<E> update(E extension) {
|
||||
checkClientWritable(extension);
|
||||
// Refactor the atomic reference if we have a better solution.
|
||||
|
@ -223,7 +243,6 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public <E extends Extension> Mono<E> delete(E extension) {
|
||||
checkClientWritable(extension);
|
||||
// set deletionTimestamp
|
||||
|
@ -247,7 +266,8 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
var indexer = indexerFactory.getIndexer(gvk);
|
||||
return client.create(name, data)
|
||||
.map(created -> converter.convertFrom(type, created))
|
||||
.doOnNext(indexer::indexRecord);
|
||||
.doOnNext(indexer::indexRecord)
|
||||
.as(transactionalOperator::transactional);
|
||||
});
|
||||
}
|
||||
|
||||
|
@ -258,7 +278,8 @@ public class ReactiveExtensionClientImpl implements ReactiveExtensionClient {
|
|||
var indexer = indexerFactory.getIndexer(oldExtension.groupVersionKind());
|
||||
return client.update(name, version, data)
|
||||
.map(updated -> converter.convertFrom(type, updated))
|
||||
.doOnNext(indexer::updateRecord);
|
||||
.doOnNext(indexer::updateRecord)
|
||||
.as(transactionalOperator::transactional);
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
@ -1,9 +1,6 @@
|
|||
package run.halo.app.extension.gc;
|
||||
|
||||
import static run.halo.app.extension.Comparators.compareCreationTimestamp;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.function.Predicate;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import run.halo.app.extension.Extension;
|
||||
import run.halo.app.extension.ExtensionClient;
|
||||
|
@ -64,8 +61,6 @@ class GcSynchronizer implements Synchronizer<GcRequest> {
|
|||
if (event instanceof SchemeRegistered registeredEvent) {
|
||||
var newScheme = registeredEvent.getNewScheme();
|
||||
listDeleted(newScheme.type()).forEach(watcher::onDelete);
|
||||
client.list(newScheme.type(), deleted(), compareCreationTimestamp(true))
|
||||
.forEach(watcher::onDelete);
|
||||
}
|
||||
});
|
||||
client.watch(watcher);
|
||||
|
@ -77,16 +72,8 @@ class GcSynchronizer implements Synchronizer<GcRequest> {
|
|||
<E extends Extension> List<E> listDeleted(Class<E> type) {
|
||||
var options = new ListOptions()
|
||||
.setFieldSelector(
|
||||
FieldSelector.of(QueryFactory.all("metadata.deletionTimestamp"))
|
||||
FieldSelector.of(QueryFactory.isNotNull("metadata.deletionTimestamp"))
|
||||
);
|
||||
return client.listAll(type, options, Sort.by("metadata.creationTimestamp"))
|
||||
.stream()
|
||||
.sorted(compareCreationTimestamp(true))
|
||||
.toList();
|
||||
return client.listAll(type, options, Sort.by(Sort.Order.asc("metadata.creationTimestamp")));
|
||||
}
|
||||
|
||||
private <E extends Extension> Predicate<E> deleted() {
|
||||
return extension -> extension.getMetadata().getDeletionTimestamp() != null;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -20,7 +20,6 @@ import run.halo.app.extension.GroupVersionKind;
|
|||
import run.halo.app.extension.ListOptions;
|
||||
import run.halo.app.extension.ListResult;
|
||||
import run.halo.app.extension.PageRequest;
|
||||
import run.halo.app.extension.index.query.All;
|
||||
import run.halo.app.extension.index.query.QueryIndexViewImpl;
|
||||
import run.halo.app.extension.router.selector.FieldSelector;
|
||||
import run.halo.app.extension.router.selector.LabelSelector;
|
||||
|
@ -155,12 +154,13 @@ public class IndexedQueryEngineImpl implements IndexedQueryEngine {
|
|||
stopWatch.stop();
|
||||
|
||||
stopWatch.start("retrieve matched metadata names");
|
||||
var hasLabelSelector = hasLabelSelector(options.getLabelSelector());
|
||||
final List<String> matchedByLabels = hasLabelSelector
|
||||
? retrieveForLabelMatchers(options.getLabelSelector().getMatchers(), fieldPathEntryMap,
|
||||
allMetadataNames)
|
||||
: allMetadataNames;
|
||||
if (hasLabelSelector(options.getLabelSelector())) {
|
||||
var matchedByLabels = retrieveForLabelMatchers(options.getLabelSelector().getMatchers(),
|
||||
fieldPathEntryMap, allMetadataNames);
|
||||
if (allMetadataNames.size() != matchedByLabels.size()) {
|
||||
indexView.removeByIdNotIn(new TreeSet<>(matchedByLabels));
|
||||
}
|
||||
}
|
||||
stopWatch.stop();
|
||||
|
||||
stopWatch.start("retrieve matched metadata names by fields");
|
||||
|
@ -188,8 +188,6 @@ public class IndexedQueryEngineImpl implements IndexedQueryEngine {
|
|||
}
|
||||
|
||||
boolean hasFieldSelector(FieldSelector fieldSelector) {
|
||||
return fieldSelector != null
|
||||
&& fieldSelector.query() != null
|
||||
&& !(fieldSelector.query() instanceof All);
|
||||
return fieldSelector != null;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -5,11 +5,11 @@ import lombok.extern.slf4j.Slf4j;
|
|||
import org.springframework.boot.context.event.ApplicationStartedEvent;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.StopWatch;
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
// @Component
|
||||
// TODO Remove this class on next version
|
||||
public class IndicesInitializer {
|
||||
|
||||
private final IndicesService indicesService;
|
||||
|
|
|
@ -36,6 +36,8 @@ import org.mockito.Mock;
|
|||
import org.mockito.Spy;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.springframework.dao.DataIntegrityViolationException;
|
||||
import org.springframework.transaction.ReactiveTransactionManager;
|
||||
import org.springframework.transaction.reactive.TransactionalOperator;
|
||||
import reactor.core.Exceptions;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
@ -63,6 +65,9 @@ class ReactiveExtensionClientTest {
|
|||
@Mock
|
||||
IndexerFactory indexerFactory;
|
||||
|
||||
@Mock
|
||||
ReactiveTransactionManager reactiveTransactionManager;
|
||||
|
||||
@Spy
|
||||
ObjectMapper objectMapper = JsonMapper.builder()
|
||||
.addModule(new JavaTimeModule())
|
||||
|
@ -76,6 +81,10 @@ class ReactiveExtensionClientTest {
|
|||
lenient().when(schemeManager.get(eq(FakeExtension.class)))
|
||||
.thenReturn(fakeScheme);
|
||||
lenient().when(schemeManager.get(eq(fakeScheme.groupVersionKind()))).thenReturn(fakeScheme);
|
||||
var transactionalOperator = mock(TransactionalOperator.class);
|
||||
client.setTransactionalOperator(transactionalOperator);
|
||||
lenient().when(transactionalOperator.transactional(any(Mono.class)))
|
||||
.thenAnswer(invocation -> invocation.getArgument(0));
|
||||
}
|
||||
|
||||
FakeExtension createFakeExtension(String name, Long version) {
|
||||
|
|
Loading…
Reference in New Issue