Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Consolidate message container hierarchy

Browse Source 
Initial refactoring for extracting a common abstract class for various containers.
This commit is contained in:
Soby Chacko
2023-02-22 12:36:49 -05:00
parent abe3786615
commit fa7a2fa818
3 changed files with 109 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
spring-pulsar/src/main/java/org/springframework/pulsar/core/AbstractPulsarMessageContainer.java Normal file
View File

@@ -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();
}

77
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AbstractPulsarMessageListenerContainer.java
View File

@@ -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<T> implements PulsarMessageListenerContainer,
BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware {
protected final LogAccessor logger = new LogAccessor(this.getClass());
public non-sealed abstract class AbstractPulsarMessageListenerContainer<T> extends AbstractPulsarMessageContainer
implements PulsarMessageListenerContainer {
private final PulsarConsumerFactory<T> pulsarConsumerFactory;
@@ -50,20 +42,8 @@ public non-sealed abstract class AbstractPulsarMessageListenerContainer<T> 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<T> 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<T> 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<T> 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) {

74
spring-pulsar/src/main/java/org/springframework/pulsar/reader/AbstractPulsarMessageReaderContainer.java
View File

@@ -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 <T> reader data type.
* @author Soby Chacko
*/
public non-sealed abstract class AbstractPulsarMessageReaderContainer<T> implements PulsarMessageReaderContainer,
BeanNameAware, ApplicationEventPublisherAware, ApplicationContextAware {
protected final LogAccessor logger = new LogAccessor(this.getClass());
public non-sealed abstract class AbstractPulsarMessageReaderContainer<T> extends AbstractPulsarMessageContainer
implements PulsarMessageReaderContainer {
private final PulsarReaderFactory<T> 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<? super T> pulsarReaderFactory,
PulsarReaderContainerProperties pulsarReaderContainerProperties) {
@@ -73,11 +52,6 @@ public non-sealed abstract class AbstractPulsarMessageReaderContainer<T> 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<T> 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<T> 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) {