From fa7a2fa81854e0fe3ead635764171c8d07672615 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 22 Feb 2023 12:36:49 -0500 Subject: [PATCH] Consolidate message container hierarchy Initial refactoring for extracting a common abstract class for various containers. --- .../core/AbstractPulsarMessageContainer.java | 103 ++++++++++++++++++ ...bstractPulsarMessageListenerContainer.java | 77 +------------ .../AbstractPulsarMessageReaderContainer.java | 74 +------------ 3 files changed, 109 insertions(+), 145 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/AbstractPulsarMessageContainer.java diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/AbstractPulsarMessageContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/AbstractPulsarMessageContainer.java new file mode 100644 index 00000000..4708230c --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/AbstractPulsarMessageContainer.java @@ -0,0 +1,103 @@ +/* + * 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 org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.context.SmartLifecycle; +import org.springframework.core.log.LogAccessor; +import org.springframework.lang.Nullable; + +/** + * Base class for the various container implementations. + * + * @author Soby Chacko + */ +public abstract class AbstractPulsarMessageContainer implements ApplicationEventPublisherAware, BeanNameAware, + ApplicationContextAware, SmartLifecycle, DisposableBean { + + protected final LogAccessor logger = new LogAccessor(this.getClass()); + + private ApplicationEventPublisher applicationEventPublisher; + + private String beanName; + + private ApplicationContext applicationContext; + + private int phase; + + protected boolean autoStartup = true; + + protected volatile boolean running = false; + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + /** + * Get the event publisher. + * @return the publisher + */ + @Nullable + public ApplicationEventPublisher getApplicationEventPublisher() { + return this.applicationEventPublisher; + } + + @Override + public void setBeanName(String name) { + this.beanName = name; + } + + /** + * Return the bean name. + * @return the bean name. + */ + @Nullable + public String getBeanName() { + return this.beanName; // the container factory sets this to the listener id + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Nullable + protected ApplicationContext getApplicationContext() { + return this.applicationContext; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public int getPhase() { + return this.phase; + } + + protected abstract void doStart(); + + protected abstract void doStop(); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java index 77bffce7..57bf5db4 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java @@ -19,14 +19,8 @@ package org.springframework.pulsar.listener; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.RedeliveryBackoff; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ApplicationEventPublisher; -import org.springframework.context.ApplicationEventPublisherAware; -import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; +import org.springframework.pulsar.core.AbstractPulsarMessageContainer; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.util.Assert; @@ -39,10 +33,8 @@ import io.micrometer.observation.ObservationRegistry; * @author Soby Chacko * @author Alexander Preuß */ -public non-sealed abstract class AbstractPulsarMessageListenerContainer implements PulsarMessageListenerContainer, - BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware { - - protected final LogAccessor logger = new LogAccessor(this.getClass()); +public non-sealed abstract class AbstractPulsarMessageListenerContainer extends AbstractPulsarMessageContainer + implements PulsarMessageListenerContainer { private final PulsarConsumerFactory pulsarConsumerFactory; @@ -50,20 +42,8 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer imple private final ObservationRegistry observationRegistry; - private ApplicationEventPublisher applicationEventPublisher; - - private String beanName; - - private ApplicationContext applicationContext; - - private boolean autoStartup = true; - - private int phase; - protected final Object lifecycleMonitor = new Object(); - private volatile boolean running = false; - private volatile boolean paused; protected RedeliveryBackoff negativeAckRedeliveryBackoff; @@ -94,11 +74,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer imple return this.observationRegistry; } - @Override - public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { - this.applicationEventPublisher = applicationEventPublisher; - } - @Override public boolean isRunning() { return this.running; @@ -108,39 +83,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer imple this.running = running; } - /** - * Get the event publisher. - * @return the publisher - */ - @Nullable - public ApplicationEventPublisher getApplicationEventPublisher() { - return this.applicationEventPublisher; - } - - @Override - public void setBeanName(String name) { - this.beanName = name; - } - - /** - * Return the bean name. - * @return the bean name. - */ - @Nullable - public String getBeanName() { - return this.beanName; // the container factory sets this to the listener id - } - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = applicationContext; - } - - @Nullable - protected ApplicationContext getApplicationContext() { - return this.applicationContext; - } - @Override public void setupMessageListener(Object messageListener) { this.pulsarContainerProperties.setMessageListener(messageListener); @@ -156,19 +98,6 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer imple this.autoStartup = autoStartup; } - public void setPhase(int phase) { - this.phase = phase; - } - - @Override - public int getPhase() { - return this.phase; - } - - protected abstract void doStart(); - - protected abstract void doStop(); - @Override public final void start() { synchronized (this.lifecycleMonitor) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/AbstractPulsarMessageReaderContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/AbstractPulsarMessageReaderContainer.java index 1765916a..6b8bb295 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/AbstractPulsarMessageReaderContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/AbstractPulsarMessageReaderContainer.java @@ -18,14 +18,7 @@ package org.springframework.pulsar.reader; import org.apache.pulsar.client.api.ReaderListener; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ApplicationEventPublisher; -import org.springframework.context.ApplicationEventPublisherAware; -import org.springframework.core.log.LogAccessor; -import org.springframework.lang.Nullable; +import org.springframework.pulsar.core.AbstractPulsarMessageContainer; import org.springframework.pulsar.core.PulsarReaderFactory; import org.springframework.util.Assert; @@ -35,29 +28,15 @@ import org.springframework.util.Assert; * @param reader data type. * @author Soby Chacko */ -public non-sealed abstract class AbstractPulsarMessageReaderContainer implements PulsarMessageReaderContainer, - BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware { - - protected final LogAccessor logger = new LogAccessor(this.getClass()); +public non-sealed abstract class AbstractPulsarMessageReaderContainer extends AbstractPulsarMessageContainer + implements PulsarMessageReaderContainer { private final PulsarReaderFactory pulsarReaderFactory; private final PulsarReaderContainerProperties pulsarReaderContainerProperties; - private ApplicationEventPublisher applicationEventPublisher; - - private String beanName; - - private ApplicationContext applicationContext; - - private boolean autoStartup = true; - - private int phase; - protected final Object lifecycleMonitor = new Object(); - private volatile boolean running = false; - @SuppressWarnings("unchecked") protected AbstractPulsarMessageReaderContainer(PulsarReaderFactory pulsarReaderFactory, PulsarReaderContainerProperties pulsarReaderContainerProperties) { @@ -73,11 +52,6 @@ public non-sealed abstract class AbstractPulsarMessageReaderContainer impleme return this.pulsarReaderContainerProperties; } - @Override - public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { - this.applicationEventPublisher = applicationEventPublisher; - } - @Override public boolean isRunning() { return this.running; @@ -87,35 +61,6 @@ public non-sealed abstract class AbstractPulsarMessageReaderContainer impleme this.running = running; } - /** - * Get the event publisher. - * @return the publisher - */ - @Nullable - public ApplicationEventPublisher getApplicationEventPublisher() { - return this.applicationEventPublisher; - } - - @Override - public void setBeanName(String name) { - this.beanName = name; - } - - @Nullable - public String getBeanName() { - return this.beanName; // the container factory sets this to the listener id - } - - @Override - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - this.applicationContext = applicationContext; - } - - @Nullable - protected ApplicationContext getApplicationContext() { - return this.applicationContext; - } - @Override public void setupReaderListener(Object messageListener) { this.pulsarReaderContainerProperties.setReaderListener(messageListener); @@ -131,19 +76,6 @@ public non-sealed abstract class AbstractPulsarMessageReaderContainer impleme this.autoStartup = autoStartup; } - public void setPhase(int phase) { - this.phase = phase; - } - - @Override - public int getPhase() { - return this.phase; - } - - protected abstract void doStart(); - - protected abstract void doStop(); - @Override public final void start() { synchronized (this.lifecycleMonitor) {