From 981eb053a3871434630ba2b5b431a9fa42bdb8f6 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 22 Sep 2023 21:23:33 -0500 Subject: [PATCH] Add graceful restart to client (#452) This commit introduces a restartable client that participates in SmartLifecycle and handles stop and start gracefully. Resolves #422 --- build.gradle | 1 + .../DefaultReactivePulsarSenderFactory.java | 47 +++- .../core/RestartableComponentSupport.java | 138 ++++++++++++ ...faultReactivePulsarSenderFactoryTests.java | 25 +++ .../core/CachingPulsarProducerFactory.java | 49 ++++- .../core/DefaultPulsarClientFactory.java | 18 +- .../pulsar/core/ProducerUtils.java | 17 +- .../pulsar/core/PulsarClientProxy.java | 203 +++++++++++++++++ .../pulsar/core/RestartableComponentBase.java | 49 +++++ .../core/RestartableComponentSupport.java | 143 ++++++++++++ .../core/RestartableSingletonFactory.java | 147 +++++++++++++ .../CachingPulsarProducerFactoryTests.java | 25 ++- .../core/DefaultPulsarClientFactoryTests.java | 25 +++ .../pulsar/core/PulsarClientProxyTests.java | 204 ++++++++++++++++++ .../RestartableSingletonFactoryTests.java | 180 ++++++++++++++++ 15 files changed, 1258 insertions(+), 13 deletions(-) create mode 100644 spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/RestartableComponentSupport.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProxy.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentBase.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentSupport.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableSingletonFactory.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarClientProxyTests.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/RestartableSingletonFactoryTests.java diff --git a/build.gradle b/build.gradle index 636024f3..f73f565a 100644 --- a/build.gradle +++ b/build.gradle @@ -34,6 +34,7 @@ nohttp { source.exclude "**/build/**" source.exclude "**/out/**" source.exclude "**/target/**" + source.exclude "**/*.dylib" } check { diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java index 000c022f..23209ef5 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java @@ -19,6 +19,7 @@ package org.springframework.pulsar.reactive.core; import java.util.Collections; import java.util.List; import java.util.Objects; +import java.util.concurrent.atomic.AtomicReference; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; @@ -31,6 +32,7 @@ import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import org.springframework.pulsar.core.DefaultTopicResolver; +import org.springframework.pulsar.core.PulsarClientProxy; import org.springframework.pulsar.core.TopicResolver; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -42,10 +44,15 @@ import org.springframework.util.CollectionUtils; * @author Christophe Bornet * @author Chris Bono */ -public final class DefaultReactivePulsarSenderFactory implements ReactivePulsarSenderFactory { +public final class DefaultReactivePulsarSenderFactory + implements ReactivePulsarSenderFactory, RestartableComponentSupport { + + private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 100; private final LogAccessor logger = new LogAccessor(this.getClass()); + private final AtomicReference currentState = RestartableComponentSupport.initialState(); + private final ReactivePulsarClient reactivePulsarClient; private final TopicResolver topicResolver; @@ -110,6 +117,9 @@ public final class DefaultReactivePulsarSenderFactory implements ReactivePuls private ReactiveMessageSender doCreateReactiveMessageSender(Schema schema, @Nullable String topic, @Nullable List> customizers) { Objects.requireNonNull(schema, "Schema must be specified"); + + this.logger.warn(() -> "**** Du CreateMessageSender for topic=" + topic); + String resolvedTopic = this.topicResolver.resolveTopic(topic, () -> getDefaultTopic()).orElseThrow(); this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic)); @@ -139,6 +149,41 @@ public final class DefaultReactivePulsarSenderFactory implements ReactivePuls return this.defaultTopic; } + /** + * Return the phase that this lifecycle object is supposed to run in. + *

+ * This component has a phase that comes after the {@link PulsarClientProxy + * restartable client} but before other lifecycle and smart lifecycle components whose + * phase values are "0" and "max", respectively. + * @return a phase that is after the restartable client and before other default + * components. + * @see PulsarClientProxy#getPhase() + */ + @Override + public int getPhase() { + return LIFECYCLE_PHASE; + } + + @Override + public AtomicReference currentState() { + return this.currentState; + } + + @Override + public LogAccessor logger() { + return this.logger; + } + + @Override + public void doStop() { + try { + this.reactiveMessageSenderCache.close(); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + /** * Builder for {@link DefaultReactivePulsarSenderFactory}. * diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/RestartableComponentSupport.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/RestartableComponentSupport.java new file mode 100644 index 00000000..e150b533 --- /dev/null +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/RestartableComponentSupport.java @@ -0,0 +1,138 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.reactive.core; + +import java.util.concurrent.atomic.AtomicReference; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; + +/** + * Provides a simple base implementation for a component that can be restarted (stopped + * then started) and still be in a usable state. + *

+ * This is an interface that provides default methods that rely on the current component + * state which must be maintained by the implementing component. + *

+ * This can serve as a base implementation for coordinated checkpoint and restore by + * simply implementing the {@link #doStart() start} and/or {@link #doStop() stop} callback + * to re-acquire and release resources, respectively. + *

+ * Implementors are required to provide the component state and a logger. + * + * @author Chris Bono + */ +interface RestartableComponentSupport extends SmartLifecycle, DisposableBean { + + /** + * Gets the initial state for the implementing component. + * @return the initial component state + */ + static AtomicReference initialState() { + return new AtomicReference<>(State.CREATED); + } + + /** + * Callback to get the current state from the component. + * @return the current state of the component + */ + AtomicReference currentState(); + + /** + * Callback to get the component specific logger. + * @return the component specific logger + */ + LogAccessor logger(); + + /** + * Lifecycle state of this factory. + */ + enum State { + + /** Component initially created. */ + CREATED, + /** Component in the process of being started. */ + STARTING, + /** Component has been started. */ + STARTED, + /** Component in the process of being stopped. */ + STOPPING, + /** Component has been stopped. */ + STOPPED, + /** Component has been destroyed. */ + DESTROYED; + + } + + @Override + default boolean isRunning() { + return State.STARTED.equals(currentState().get()); + } + + @Override + default void start() { + State current = currentState().getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state); + if (isCreatedOrStopped(current)) { + logger().debug(() -> "Starting..."); + doStart(); + currentState().set(State.STARTED); + logger().debug(() -> "Started"); + } + } + + private static boolean isCreatedOrStopped(@Nullable State state) { + return State.CREATED.equals(state) || State.STOPPED.equals(state); + } + + /** + * Callback invoked during startup - default implementation does nothing. + */ + default void doStart() { + } + + @Override + default void stop() { + State current = currentState().getAndUpdate(state -> isCreatedOrStarted(state) ? State.STOPPING : state); + if (isCreatedOrStarted(current)) { + logger().debug(() -> "Stopping..."); + doStop(); + currentState().set(State.STOPPED); + logger().debug(() -> "Stopped"); + } + } + + private static boolean isCreatedOrStarted(@Nullable State state) { + return State.CREATED.equals(state) || State.STARTED.equals(state); + } + + /** + * Callback invoked during stop - default implementation does nothing. + */ + default void doStop() { + } + + @Override + default void destroy() { + logger().debug(() -> "Destroying..."); + stop(); + currentState().set(State.DESTROYED); + logger().debug(() -> "Destroyed"); + } + +} diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java index e4f3c258..31d573c8 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java @@ -20,8 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatNullPointerException; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.clearInvocations; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import java.util.Arrays; import java.util.Collections; @@ -210,4 +214,25 @@ class DefaultReactivePulsarSenderFactoryTests { } + @Nested + class RestartFactoryTests { + + @Test + void restartLifecycle() throws Exception { + var cache = spy(AdaptedReactivePulsarClientFactory.createCache()); + var senderFactory = (DefaultReactivePulsarSenderFactory) newSenderFactoryWithCache(cache); + senderFactory.start(); + senderFactory.createSender(schema, "topic1"); + senderFactory.stop(); + senderFactory.stop(); + verify(cache, times(1)).close(); + clearInvocations(cache); + senderFactory.start(); + senderFactory.createSender(schema, "topic2"); + senderFactory.stop(); + verify(cache, times(1)).close(); + } + + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java index 769e8bc1..bfd85deb 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.pulsar.client.api.MessageId; @@ -58,12 +59,17 @@ import org.springframework.util.Assert; * @author Alexander Preuß * @author Christophe Bornet */ -public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactory implements DisposableBean { +public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactory + implements RestartableComponentSupport { + + private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 100; private final LogAccessor logger = new LogAccessor(this.getClass()); private final CacheProvider, Producer> producerCache; + private final AtomicReference currentState = RestartableComponentSupport.initialState(); + /** * Construct a caching producer factory with the specified values for the cache * configuration. @@ -85,7 +91,7 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor (key, producer, cause) -> { this.logger.debug(() -> "Producer %s evicted from cache due to %s" .formatted(ProducerUtils.formatProducer(producer), cause)); - closeProducer(producer); + closeProducer(producer, true); }); } @@ -113,22 +119,51 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor } } + /** + * Return the phase that this lifecycle object is supposed to run in. + *

+ * Because this object depends on the restartable client, it uses a phase slightly + * larger than the one used by the restartable client. This ensures that it starts + * after and stops before the restartable client. + * @return the phase to execute in (just after the restartable client) + * @see PulsarClientProxy#getPhase() + */ @Override - public void destroy() { - this.producerCache.invalidateAll((key, producer) -> closeProducer(producer)); + public int getPhase() { + return LIFECYCLE_PHASE; } - private void closeProducer(Producer producer) { + @Override + public AtomicReference currentState() { + return this.currentState; + } + + @Override + public LogAccessor logger() { + return this.logger; + } + + @Override + public void doStop() { + this.producerCache.invalidateAll((key, producer) -> closeProducer(producer, false)); + } + + private void closeProducer(Producer producer, boolean async) { Producer actualProducer = null; if (producer instanceof ProducerWithCloseCallback wrappedProducer) { actualProducer = wrappedProducer.getActualProducer(); } if (actualProducer == null) { - this.logger.warn(() -> "Unable to get actual producer for %s - will skip closing it" + this.logger.trace(() -> "Unable to get actual producer for %s - will skip closing it" .formatted(ProducerUtils.formatProducer(producer))); return; } - ProducerUtils.closeProducerAsync(actualProducer, this.logger); + if (async) { + ProducerUtils.closeProducerAsync(actualProducer, this.logger); + } + else { + ProducerUtils.closeProducer(actualProducer, this.logger, Duration.ofSeconds(15L)); + } } /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java index 77921700..6057f435 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java @@ -19,6 +19,9 @@ package org.springframework.pulsar.core; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; +import org.springframework.context.EnvironmentAware; +import org.springframework.core.env.Environment; +import org.springframework.core.log.LogAccessor; import org.springframework.util.Assert; /** @@ -27,10 +30,14 @@ import org.springframework.util.Assert; * @author Soby Chacko * @author Chris Bono */ -public class DefaultPulsarClientFactory implements PulsarClientFactory { +public class DefaultPulsarClientFactory implements PulsarClientFactory, EnvironmentAware { + + private final LogAccessor logger = new LogAccessor(this.getClass()); private final PulsarClientBuilderCustomizer customizer; + private boolean useRestartableClient; + /** * Construct a factory that creates clients using a default Pulsar client builder with * no modifications other than the specified service url. @@ -51,9 +58,18 @@ public class DefaultPulsarClientFactory implements PulsarClientFactory { @Override public PulsarClient createClient() throws PulsarClientException { + if (this.useRestartableClient) { + this.logger.info(() -> "Using restartable client"); + return new PulsarClientProxy(this.customizer); + } var clientBuilder = PulsarClient.builder(); this.customizer.customize(clientBuilder); return clientBuilder.build(); } + @Override + public void setEnvironment(Environment environment) { + this.useRestartableClient = environment.getProperty("spring.pulsar.client.restartable", Boolean.class, true); + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java index 8f9f4dc3..66ad5b0f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java @@ -16,6 +16,10 @@ package org.springframework.pulsar.core; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + import org.apache.pulsar.client.api.Producer; import org.springframework.core.log.LogAccessor; @@ -34,11 +38,20 @@ final class ProducerUtils { return "(%s:%s)".formatted(producer.getProducerName(), producer.getTopic()); } - static void closeProducerAsync(Producer producer, LogAccessor logger) { - producer.closeAsync().exceptionally(e -> { + static CompletableFuture closeProducerAsync(Producer producer, LogAccessor logger) { + return producer.closeAsync().exceptionally(e -> { logger.warn(e, () -> "Failed to close producer %s".formatted(ProducerUtils.formatProducer(producer))); return null; }); } + static void closeProducer(Producer producer, LogAccessor logger, Duration maxWaitTime) { + try { + producer.closeAsync().get(maxWaitTime.toMillis(), TimeUnit.MILLISECONDS); + } + catch (Exception e) { + logger.warn(e, () -> "Failed to close producer %s".formatted(ProducerUtils.formatProducer(producer))); + } + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProxy.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProxy.java new file mode 100644 index 00000000..ad087810 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProxy.java @@ -0,0 +1,203 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import org.apache.pulsar.client.api.ConsumerBuilder; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.ReaderBuilder; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TableViewBuilder; +import org.apache.pulsar.client.api.transaction.TransactionBuilder; + +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.util.Assert; + +/** + * A {@link PulsarClient} implementation that delegates to another actual Pulsar client. + * The proxy client can be stopped and then started and still be in a usable state without + * any knowledge of the restart or changes required to the users of the client. + *

+ * The proxy client participates in the Spring {@link SmartLifecycle Lifecycle} and closes + * the underlying client when {@link SmartLifecycle#stop() stopped} and creates another + * delegate client when subsequently {@link SmartLifecycle#start() started}. + * + * @author Chris Bono + */ +public final class PulsarClientProxy extends RestartableSingletonFactory implements PulsarClient { + + private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 200; + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final PulsarClientBuilderCustomizer customizer; + + /** + * Construct a factory that creates clients using a customized Pulsar client builder. + * @param customizer the customizer to apply to the builder + */ + PulsarClientProxy(PulsarClientBuilderCustomizer customizer) { + Assert.notNull(customizer, "customizer must not be null"); + this.customizer = customizer; + } + + /** + * Return the phase that this lifecycle object is supposed to run in. + *

+ * Lifecycle objects are started in ascending phase order (those w/ smaller phases are + * started before those with larger phases). + *

+ * Lifecycle objects are stopped in descending phase order (those w/ larger phases are + * stopped before those with smaller phases). + *

+ * The phases range from {@link Integer#MIN_VALUE} to {@link Integer#MAX_VALUE}. + *

+ * The restartable client has a phase value that is roughly at the half-way marker on + * the left hand side of the phase continuum (in the middle of "min" and + * "0"). If another component depends on the restartable client it should + * use a phase that is greater than this value. + *

+ * Because {@link Lifecycle regular} lifecycle objects have a default phase of + * "0" and {@link SmartLifecycle smart} lifecycle objects have a default + * phase of "max", the restartable client will be started before (and + * stopped after) the majority of all other default configured lifecycle objects. + * @return the phase to execute in ({@link #LIFECYCLE_PHASE}) + */ + @Override + public int getPhase() { + return LIFECYCLE_PHASE; + } + + @Override + protected PulsarClient createInstance() { + this.logger.debug(() -> "Creating client"); + var clientBuilder = PulsarClient.builder(); + this.customizer.customize(clientBuilder); + try { + return clientBuilder.build(); + } + catch (PulsarClientException e) { + throw new RuntimeException(e); + } + } + + @Override + protected void stopInstance(PulsarClient pulsarClient) { + this.logger.debug(() -> "Closing client"); + try { + pulsarClient.close(); + } + catch (PulsarClientException e) { + throw new RuntimeException(e); + } + } + + @Override + protected boolean discardInstanceAfterStop() { + return false; + } + + // --- PulsarClient implementation below here + + @Override + public ProducerBuilder newProducer() { + return this.getInstance().newProducer(); + } + + @Override + public ProducerBuilder newProducer(Schema schema) { + return this.getInstance().newProducer(schema); + } + + @Override + public ConsumerBuilder newConsumer() { + return this.getInstance().newConsumer(); + } + + @Override + public ConsumerBuilder newConsumer(Schema schema) { + return this.getInstance().newConsumer(schema); + } + + @Override + public ReaderBuilder newReader() { + return this.getInstance().newReader(); + } + + @Override + public ReaderBuilder newReader(Schema schema) { + return this.getInstance().newReader(schema); + } + + @SuppressWarnings("deprecation") + @Override + public TableViewBuilder newTableViewBuilder(Schema schema) { + return this.getInstance().newTableViewBuilder(schema); + } + + @Override + public TableViewBuilder newTableView() { + return this.getInstance().newTableView(); + } + + @Override + public TableViewBuilder newTableView(Schema schema) { + return this.getInstance().newTableView(schema); + } + + @Override + public void updateServiceUrl(String serviceUrl) throws PulsarClientException { + this.getInstance().updateServiceUrl(serviceUrl); + } + + @Override + public CompletableFuture> getPartitionsForTopic(String topic) { + return this.getInstance().getPartitionsForTopic(topic); + } + + @Override + public void close() throws PulsarClientException { + this.getInstance().close(); + } + + @Override + public CompletableFuture closeAsync() { + return this.getInstance().closeAsync(); + } + + @Override + public void shutdown() throws PulsarClientException { + this.getInstance().shutdown(); + } + + @Override + public boolean isClosed() { + return this.getInstance().isClosed(); + } + + @Override + public TransactionBuilder newTransaction() { + return this.getInstance().newTransaction(); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentBase.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentBase.java new file mode 100644 index 00000000..935c9b42 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentBase.java @@ -0,0 +1,49 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import java.util.concurrent.atomic.AtomicReference; + +import org.springframework.core.log.LogAccessor; + +/** + * Provides a simple base implementation for a component that can be restarted (stopped + * then started) and still be in a usable state. + *

+ * Subclasses can use this as a base implementation for coordinated checkpoint and restore + * by simply implementing the {@link #doStart() start} and/or {@link #doStop() stop} + * callback to re-acquire and release resources, respectively. + * + * @author Chris Bono + */ +abstract class RestartableComponentBase implements RestartableComponentSupport { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final AtomicReference state = RestartableComponentSupport.initialState(); + + @Override + public AtomicReference currentState() { + return this.state; + } + + @Override + public LogAccessor logger() { + return this.logger; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentSupport.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentSupport.java new file mode 100644 index 00000000..a10203e1 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentSupport.java @@ -0,0 +1,143 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import java.util.concurrent.atomic.AtomicReference; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; + +/** + * Provides a simple base implementation for a component that can be restarted (stopped + * then started) and still be in a usable state. + *

+ * This is an interface that provides default methods that rely on the current component + * state which must be maintained by the implementing component. + *

+ * This can serve as a base implementation for coordinated checkpoint and restore by + * simply implementing the {@link #doStart() start} and/or {@link #doStop() stop} callback + * to re-acquire and release resources, respectively. + *

+ * Implementors are required to provide the component state and an optional logger - see + * {@link RestartableComponentBase} for an example. + * + * @author Chris Bono + * @see RestartableComponentBase + */ +interface RestartableComponentSupport extends SmartLifecycle, DisposableBean { + + /** Trace logger used for method invocations. */ + LogAccessor logger = new LogAccessor(RestartableComponentSupport.class); + + /** + * Gets the initial state for the implementing component. + * @return the initial component state + */ + static AtomicReference initialState() { + return new AtomicReference<>(State.CREATED); + } + + /** + * Callback to get the current state from the component. + * @return the current state of the component + */ + AtomicReference currentState(); + + /** + * Callback to get the component specific logger. + * @return the component specific logger + */ + LogAccessor logger(); + + /** + * Lifecycle state of this factory. + */ + enum State { + + /** Component initially created. */ + CREATED, + /** Component in the process of being started. */ + STARTING, + /** Component has been started. */ + STARTED, + /** Component in the process of being stopped. */ + STOPPING, + /** Component has been stopped. */ + STOPPED, + /** Component has been destroyed. */ + DESTROYED; + + } + + @Override + default boolean isRunning() { + return State.STARTED.equals(currentState().get()); + } + + @Override + default void start() { + State current = currentState().getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state); + if (isCreatedOrStopped(current)) { + logger().debug(() -> "Starting..."); + doStart(); + currentState().set(State.STARTED); + logger().debug(() -> "Started"); + } + } + + private static boolean isCreatedOrStopped(@Nullable State state) { + return State.CREATED.equals(state) || State.STOPPED.equals(state); + } + + /** + * Callback invoked during startup - default implementation does nothing. + */ + default void doStart() { + } + + @Override + default void stop() { + State current = currentState().getAndUpdate(state -> isCreatedOrStarted(state) ? State.STOPPING : state); + if (isCreatedOrStarted(current)) { + logger().debug(() -> "Stopping..."); + doStop(); + currentState().set(State.STOPPED); + logger().debug(() -> "Stopped"); + } + } + + private static boolean isCreatedOrStarted(@Nullable State state) { + return State.CREATED.equals(state) || State.STARTED.equals(state); + } + + /** + * Callback invoked during stop - default implementation does nothing. + */ + default void doStop() { + } + + @Override + default void destroy() { + logger().debug(() -> "Destroying..."); + stop(); + currentState().set(State.DESTROYED); + logger().debug(() -> "Destroyed"); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableSingletonFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableSingletonFactory.java new file mode 100644 index 00000000..4e87f0b1 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableSingletonFactory.java @@ -0,0 +1,147 @@ +/* + * Copyright 2022-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import java.util.concurrent.atomic.AtomicBoolean; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +/** + * Provides a simple base implementation for a restartable singleton factory. + *

+ * It is restartable in the sense that it can be stopped and then started and still be in + * a usable state. + *

+ * Because it releases its resources when {@link SmartLifecycle#stop() stopped} and + * re-acquires them when subsequently {@link SmartLifecycle#start() started}, it can also + * be used as a base implementation for coordinated checkpoint and restore. + * + * @param the bean type + * @author Chris Bono + */ +abstract class RestartableSingletonFactory extends RestartableComponentBase implements InitializingBean { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final AtomicBoolean initialized = new AtomicBoolean(false); + + private T instance; + + protected RestartableSingletonFactory() { + super(); + } + + protected RestartableSingletonFactory(T instance) { + super(); + Assert.notNull(instance, () -> "instance must not be null"); + this.instance = instance; + this.initialized.set(true); + } + + @Override + public void afterPropertiesSet() throws Exception { + ensureInstanceCreated(); + } + + @Override + public void doStart() { + ensureInstanceCreated(); + } + + private void ensureInstanceCreated() { + if (this.initialized.compareAndSet(false, true)) { + this.logger.debug(() -> "Creating instance"); + this.instance = createInstance(); + } + } + + @Override + public void doStop() { + if (this.instance != null) { + this.logger.debug(() -> "Stopping instance"); + stopInstance(this.instance); + if (this.discardInstanceAfterStop()) { + this.logger.debug(() -> "Discarding instance"); + this.instance = null; + } + } + this.initialized.set(false); + } + + @Override + public void destroy() { + super.destroy(); + this.instance = null; + } + + /** + * Gets the singleton instance. + * @return the singleton instance + */ + public final T getInstance() { + return this.instance; + } + + /** + * Template method that subclasses must override to construct the backing singleton + * instance returned by this factory. + *

+ * Implementations should throw a {@link RuntimeException} if an error occurs during + * creation. + *

+ * Invoked on {@link #afterPropertiesSet() initialization} of this bean if the + * instance has not already been set via the constructor OR during {@link #start()} if + * the instance is null. + * @return the single object managed by the factory + */ + protected abstract T createInstance(); + + /** + * Callback to allow the singleton instance to be "stopped" (ie. allow it to + * release any resources). + *

+ * Implementations should throw a {@link RuntimeException} if an error occurs during + * destruction. + *

+ * The default implementation is empty. + * @param instance the singleton instance, as returned by {@link #createInstance()} + */ + protected void stopInstance(@Nullable T instance) { + } + + /** + * Whether to discard the singleton instance (set reference to it null) when stopped + * (default is true). + * @return whether to discard the singleton when stopped + */ + protected boolean discardInstanceAfterStop() { + return true; + } + + /** + * Whether the singleton instance has been initialized. + * @return whether the singleton instance has been initialized + */ + protected boolean initialized() { + return this.initialized.get(); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java index 05be92bc..8d0f5e21 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java @@ -230,8 +230,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Override protected CachingPulsarProducerFactory producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, @Nullable List> defaultConfigCustomizers) { - var producerFactory = new CachingPulsarProducerFactory(pulsarClient, defaultTopic, - defaultConfigCustomizers, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2); + var producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers, + new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2); producerFactories.add(producerFactory); return producerFactory; } @@ -311,4 +311,25 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { } + @Nested + class RestartFactoryTests { + + @Test + void restartLifecycle() throws PulsarClientException { + var producerFactory = (CachingPulsarProducerFactory) producerFactory(pulsarClient, null, null); + producerFactory.start(); + var producer1 = producerFactory.createProducer(schema, "topic1"); + var producer2 = producerFactory.createProducer(schema, "topic2"); + assertThat(actualProducer(producer1).isConnected()).isTrue(); + assertThat(actualProducer(producer2).isConnected()).isTrue(); + producerFactory.stop(); + assertThat(actualProducer(producer1).isConnected()).isFalse(); + assertThat(actualProducer(producer2).isConnected()).isFalse(); + producerFactory.start(); + var producer3 = producerFactory.createProducer(schema, "topic3"); + assertThat(actualProducer(producer3).isConnected()).isTrue(); + } + + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java index fc4e224d..1a94bdac 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java @@ -23,6 +23,8 @@ import static org.assertj.core.api.Assertions.assertThatRuntimeException; import org.apache.pulsar.client.api.PulsarClientException; import org.junit.jupiter.api.Test; +import org.springframework.mock.env.MockEnvironment; + /** * Tests for {@link DefaultPulsarClientFactory}. * @@ -60,4 +62,27 @@ class DefaultPulsarClientFactoryTests { assertThatRuntimeException().isThrownBy(clientFactory::createClient).withMessage("Who turned out the lights?"); } + @Test + void createsRestartableClientByDefault() throws PulsarClientException { + var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150"); + clientFactory.setEnvironment(new MockEnvironment()); + assertThat(clientFactory.createClient()).isInstanceOf(PulsarClientProxy.class); + } + + @Test + void createsRestartableClientWhenPropertySetTrue() throws PulsarClientException { + var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150"); + var env = new MockEnvironment().withProperty("spring.pulsar.client.restartable", "true"); + clientFactory.setEnvironment(env); + assertThat(clientFactory.createClient()).isInstanceOf(PulsarClientProxy.class); + } + + @Test + void createsDefaultClientWhenPropertySetFalse() throws PulsarClientException { + var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150"); + var env = new MockEnvironment().withProperty("spring.pulsar.client.restartable", "false"); + clientFactory.setEnvironment(env); + assertThat(clientFactory.createClient()).isNotInstanceOf(PulsarClientProxy.class); + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarClientProxyTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarClientProxyTests.java new file mode 100644 index 00000000..64bb6848 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarClientProxyTests.java @@ -0,0 +1,204 @@ +/* + * Copyright 2023-2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.test.util.ReflectionTestUtils; + +/** + * Tests for {@link PulsarClientProxy}. + * + * @author Chris Bono + */ +class PulsarClientProxyTests implements PulsarTestContainerSupport { + + @Test + void constructWithCustomizer() throws Exception { + var restartableClient = new PulsarClientProxy( + (clientBuilder) -> clientBuilder.serviceUrl("pulsar://localhost:5150")); + restartableClient.afterPropertiesSet(); + assertThat(restartableClient.getInstance()).hasFieldOrPropertyWithValue("conf.serviceUrl", + "pulsar://localhost:5150"); + } + + @Test + void constructWithNullCustomizer() { + assertThatIllegalArgumentException().isThrownBy(() -> new PulsarClientProxy(null)) + .withMessage("customizer must not be null"); + } + + @Test + void restartLifecycle() throws Exception { + var serviceUrl = PulsarTestContainerSupport.getPulsarBrokerUrl(); + var restartableClient = new PulsarClientProxy((builder) -> builder.serviceUrl(serviceUrl)); + restartableClient.afterPropertiesSet(); + var delegateClient = restartableClient.getInstance(); + assertThat(delegateClient).isNotNull(); + assertThat(delegateClient).hasFieldOrPropertyWithValue("conf.serviceUrl", serviceUrl); + assertThat(delegateClient.isClosed()).isFalse(); + assertThat(restartableClient.isClosed()).isFalse(); + + // Stop and verify the client is closed + restartableClient.stop(); + assertThat(restartableClient.isClosed()).isTrue(); + assertThat(delegateClient.isClosed()).isTrue(); + assertThat(restartableClient.getInstance()).isSameAs(delegateClient); + + // Restart and verify the client is created again during start + restartableClient.start(); + var newDelegateClient = restartableClient.getInstance(); + assertThat(newDelegateClient).isNotNull(); + assertThat(newDelegateClient).hasFieldOrPropertyWithValue("conf.serviceUrl", serviceUrl); + assertThat(newDelegateClient.isClosed()).isFalse(); + assertThat(restartableClient.isClosed()).isFalse(); + assertThat(newDelegateClient).isNotSameAs(delegateClient); + + // Destroy and verify the client is destroyed as well + restartableClient.destroy(); + assertThat(newDelegateClient.isClosed()).isTrue(); + assertThat(restartableClient.getInstance()).isNull(); + } + + @Nested + class DelegateClientTests { + + private PulsarClientProxy restartableClient; + + private PulsarClient delegateClient; + + @BeforeEach + void createClient() throws Exception { + var serviceUrl = PulsarTestContainerSupport.getPulsarBrokerUrl(); + restartableClient = new PulsarClientProxy((builder) -> builder.serviceUrl(serviceUrl)); + restartableClient.afterPropertiesSet(); + delegateClient = mock(PulsarClient.class); + ReflectionTestUtils.setField(restartableClient, "instance", delegateClient); + assertThat(restartableClient.getInstance()).isSameAs(delegateClient); + } + + @Test + void newProducer() { + this.restartableClient.newProducer(); + verify(this.delegateClient).newProducer(); + } + + @Test + void newProducerWithSchema() { + this.restartableClient.newProducer(Schema.STRING); + verify(this.delegateClient).newProducer(Schema.STRING); + } + + @Test + void newConsumer() { + this.restartableClient.newConsumer(); + verify(this.delegateClient).newConsumer(); + } + + @Test + void newConsumerWithSchema() { + this.restartableClient.newConsumer(Schema.STRING); + verify(this.delegateClient).newConsumer(Schema.STRING); + } + + @Test + void newReader() { + this.restartableClient.newReader(); + verify(this.delegateClient).newReader(); + } + + @Test + void newReaderWithSchema() { + this.restartableClient.newReader(Schema.STRING); + verify(this.delegateClient).newReader(Schema.STRING); + } + + @SuppressWarnings("deprecation") + @Test + void newTableViewBuilder() { + this.restartableClient.newTableViewBuilder(Schema.STRING); + verify(this.delegateClient).newTableViewBuilder(Schema.STRING); + } + + @Test + void newTableView() { + this.restartableClient.newTableView(); + verify(this.delegateClient).newTableView(); + } + + @Test + void newTableViewWithSchema() { + this.restartableClient.newTableView(Schema.STRING); + verify(this.delegateClient).newTableView(Schema.STRING); + } + + @Test + void updateServiceUrl() throws PulsarClientException { + this.restartableClient.updateServiceUrl("pulsar://foo:6150"); + verify(this.delegateClient).updateServiceUrl("pulsar://foo:6150"); + } + + @Test + void getPartitionsForTopic() { + this.restartableClient.getPartitionsForTopic("zTopic"); + verify(this.delegateClient).getPartitionsForTopic("zTopic"); + } + + @Test + void close() throws PulsarClientException { + this.restartableClient.close(); + verify(this.delegateClient).close(); + } + + @Test + void closeAsync() { + this.restartableClient.closeAsync(); + verify(this.delegateClient).closeAsync(); + } + + @Test + void shutdown() throws PulsarClientException { + this.restartableClient.shutdown(); + verify(this.delegateClient).shutdown(); + } + + @Test + void isClosed() { + this.restartableClient.isClosed(); + verify(this.delegateClient).isClosed(); + } + + @Test + void newTransaction() { + this.restartableClient.newTransaction(); + verify(this.delegateClient).newTransaction(); + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/RestartableSingletonFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/RestartableSingletonFactoryTests.java new file mode 100644 index 00000000..707ca1d5 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/RestartableSingletonFactoryTests.java @@ -0,0 +1,180 @@ +/* + * Copyright 2023 the original author or authors. + * + * 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 + * + * https://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.springframework.pulsar.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.EnumSource.Mode; +import org.mockito.InOrder; + +import org.springframework.pulsar.core.RestartableComponentSupport.State; +import org.springframework.test.util.ReflectionTestUtils; + +/** + * Tests for {@link RestartableSingletonFactory}. + */ +public class RestartableSingletonFactoryTests { + + @ParameterizedTest + @EnumSource(value = State.class, mode = Mode.INCLUDE, names = { "CREATED", "STOPPED" }) + void startCreatesInstanceAndSetsStateToStarted(State initialFactoryState) { + var fooFactory = spy(new FooFactory()); + setFactoryState(fooFactory, initialFactoryState); + assertThat(fooFactory.getInstance()).isNull(); + fooFactory.start(); + assertThat(fooFactory.getInstance()).isInstanceOf(Foo.class); + InOrder inOrder = inOrder(fooFactory); + inOrder.verify(fooFactory).doStart(); + inOrder.verify(fooFactory).createInstance(); + verifyFactoryState(fooFactory, State.STARTED); + } + + @ParameterizedTest + @EnumSource(value = State.class, mode = Mode.EXCLUDE, names = { "CREATED", "STOPPED" }) + void startDoesNothingWhenStateIsInvalid(State initialFactoryState) { + var fooFactory = spy(new FooFactory()); + setFactoryState(fooFactory, initialFactoryState); + fooFactory.start(); + verify(fooFactory, never()).createInstance(); + verify(fooFactory, never()).doStart(); + } + + @ParameterizedTest + @EnumSource(value = State.class, mode = Mode.INCLUDE, names = { "CREATED", "STARTED" }) + void stopStopsTheInstanceAndSetsStateToStopped(State initialFactoryState) { + var fooFactory = spy(new FooFactory()); + fooFactory.start(); // creates instance we use to verify destroyInstamce + var foo = fooFactory.getInstance(); + setFactoryState(fooFactory, initialFactoryState); + fooFactory.stop(); + assertThat(fooFactory.getInstance()).isNull(); + InOrder inOrder = inOrder(fooFactory); + inOrder.verify(fooFactory).doStop(); + inOrder.verify(fooFactory).stopInstance(eq(foo)); + verifyFactoryState(fooFactory, State.STOPPED); + } + + @ParameterizedTest + @EnumSource(value = State.class, mode = Mode.EXCLUDE, names = { "CREATED", "STARTED" }) + void stopDoesNothingWhenStateIsInvalid(State initialFactoryState) { + var fooFactory = spy(new FooFactory()); + setFactoryState(fooFactory, initialFactoryState); + fooFactory.stop(); + verify(fooFactory, never()).doStop(); + verify(fooFactory, never()).stopInstance(any(Foo.class)); + } + + @Test + void stopDoesNotDiscardInstanceWhenDiscardInstanceReturnsFalse() { + var fooFactory = new FooFactory() { + @Override + protected boolean discardInstanceAfterStop() { + return false; + } + }; + fooFactory.start(); + fooFactory.stop(); + assertThat(fooFactory.getInstance()).isNotNull(); + } + + @Test + void destroyCallsStopAndSetsStateToDestroyed() { + var fooFactory = spy(new FooFactory()); + fooFactory.start(); + fooFactory.destroy(); + verify(fooFactory).stop(); + verifyFactoryState(fooFactory, State.DESTROYED); + } + + @Test + void initializationCreatesInstance() throws Exception { + var fooFactory = spy(new FooFactory()); + fooFactory.afterPropertiesSet(); + verify(fooFactory).createInstance(); + } + + @Test + void createInstanceNotCalledWhenInstanceSetInConstructor() throws Exception { + var foo = new Foo("the-one"); + var fooFactory = spy(new FooFactory(foo)); + fooFactory.afterPropertiesSet(); + fooFactory.start(); + verify(fooFactory, never()).createInstance(); + assertThat(fooFactory.getInstance()).isEqualTo(foo); + } + + @Test + void createInstanceOnlyCalledOnceDuringStartup() throws Exception { + var fooFactory = spy(new FooFactory()); + fooFactory.afterPropertiesSet(); + fooFactory.start(); + verify(fooFactory).createInstance(); + } + + void setFactoryState(RestartableSingletonFactory factory, State factoryState) { + ReflectionTestUtils.setField(factory, "state", new AtomicReference<>(factoryState)); + verifyFactoryState(factory, factoryState); + } + + void verifyFactoryState(RestartableSingletonFactory factory, State expectedState) { + assertThat(factory).extracting("state").satisfies((state) -> { + assertThat(state.toString()).contains(expectedState.name()); + }); + } + + static class FooFactory extends RestartableSingletonFactory { + + FooFactory() { + super(); + } + + FooFactory(Foo instance) { + super(instance); + } + + @Override + protected Foo createInstance() { + return new Foo("restart:" + System.currentTimeMillis()); + } + + } + + static class FooKeepsInstanceFactory extends FooFactory { + + @Override + protected boolean discardInstanceAfterStop() { + return false; + } + + } + + record Foo(String name) { + + } + +}