Initial commit for concurrent message listeners

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

@@ -27,6 +27,7 @@ import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.util.Assert;
 
 /**
  * Base implementation for the {@link PulsarMessageListenerContainer}.
@@ -47,12 +48,16 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 
 	private final PulsarContainerProperties pulsarContainerProperties;
 
-	private final PulsarConsumerFactory<T> pulsarConsumerFactory;
+	protected final PulsarConsumerFactory<T> pulsarConsumerFactory;
 
 	private boolean autoStartup = true;
 
 	private int phase;
 
+	protected final Object lifecycleMonitor = new Object();
+
+	private volatile boolean running = false;
+
 	@SuppressWarnings("unchecked")
 	protected AbstractPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
 			PulsarContainerProperties pulsarContainerProperties) {
@@ -66,6 +71,15 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	protected void setRunning(boolean running) {
+		this.running = running;
+	}
+
 	/**
 	 * Get the event publisher.
 	 * @return the publisher
@@ -135,4 +149,28 @@ public abstract class AbstractPulsarMessageListenerContainer<T> implements Pulsa
 		return this.pulsarContainerProperties;
 	}
 
+	protected abstract void doStart();
+
+	protected abstract void doStop();
+
+	@Override
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!isRunning()) {
+				Assert.state(this.pulsarContainerProperties.getMessageListener() instanceof PulsarRecordMessageListener,
+						() -> "A " + PulsarRecordMessageListener.class.getName() + " implementation must be provided");
+				doStart();
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (!isRunning()) {
+				doStop();
+			}
+		}
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainer.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2022 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.listener;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.pulsar.core.PulsarConsumerFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Creates a concurrent execution context of {@link DefaultPulsarMessageListenerContainer}
+ * instances based on the {@link #setConcurrency(int) concurrency}. Concurrency > 1 is not
+ * allowed for exclusive subscriptions.
+ *
+ * @param <T> the payload type.
+ * @author Soby Chacko
+ */
+public class ConcurrentPulsarMessageListenerContainer<T> extends AbstractPulsarMessageListenerContainer<T> {
+
+	private final List<DefaultPulsarMessageListenerContainer<T>> containers = new ArrayList<>();
+
+	private int concurrency = 1;
+
+	private final List<AsyncTaskExecutor> executors = new ArrayList<>();
+
+	protected ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory<? super T> pulsarConsumerFactory,
+			PulsarContainerProperties pulsarContainerProperties) {
+		super(pulsarConsumerFactory, pulsarContainerProperties);
+	}
+
+	public int getConcurrency() {
+		return this.concurrency;
+	}
+
+	/**
+	 * The maximum number of concurrent {@link DefaultPulsarMessageListenerContainer}s
+	 * running. Messages from within the same partition will be processed sequentially.
+	 * Concurrency > 1 is not allowed for exclusive subscriptions.
+	 * @param concurrency the concurrency.
+	 */
+	public void setConcurrency(int concurrency) {
+		Assert.isTrue(concurrency > 0, "concurrency must be greater than 0");
+		this.concurrency = concurrency;
+	}
+
+	@Override
+	public void doStart() {
+		if (!isRunning()) {
+			PulsarContainerProperties containerProperties = getContainerProperties();
+
+			setRunning(true);
+
+			for (int i = 0; i < this.concurrency; i++) {
+				DefaultPulsarMessageListenerContainer<T> container = constructContainer(containerProperties);
+				configureChildContainer(i, container);
+
+				container.start();
+				this.containers.add(container);
+			}
+		}
+	}
+
+	private DefaultPulsarMessageListenerContainer<T> constructContainer(PulsarContainerProperties containerProperties) {
+		return new DefaultPulsarMessageListenerContainer<>(this.pulsarConsumerFactory, containerProperties);
+	}
+
+	private void configureChildContainer(int index, DefaultPulsarMessageListenerContainer<T> container) {
+		String beanName = getBeanName();
+		beanName = (beanName == null ? "consumer" : beanName) + "-" + index;
+		container.setBeanName(beanName);
+		ApplicationContext applicationContext = getApplicationContext();
+		if (applicationContext != null) {
+			container.setApplicationContext(applicationContext);
+		}
+		ApplicationEventPublisher publisher = getApplicationEventPublisher();
+		if (publisher != null) {
+			container.setApplicationEventPublisher(publisher);
+		}
+
+		AsyncTaskExecutor exec = container.getContainerProperties().getConsumerTaskExecutor();
+		if (exec == null) {
+			if ((this.executors.size() > index)) {
+				exec = this.executors.get(index);
+			}
+			else {
+				exec = new SimpleAsyncTaskExecutor(beanName + "-C-");
+				this.executors.add(exec);
+			}
+			container.getContainerProperties().setConsumerTaskExecutor(exec);
+		}
+	}
+
+	@Override
+	public void doStop() {
+		if (isRunning()) {
+			setRunning(false);
+			for (DefaultPulsarMessageListenerContainer<T> pulsarMessageListenerContainer : this.containers) {
+				pulsarMessageListenerContainer.stop();
+			}
+			this.containers.clear();
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return false;
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java

@@ -59,8 +59,6 @@ import org.springframework.util.StringUtils;
  */
 public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMessageListenerContainer<T> {
 
-	private volatile boolean running = false;
-
 	private volatile CompletableFuture<?> listenerConsumerFuture;
 
 	private volatile Listener listenerConsumer;
@@ -76,11 +74,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 	}
 
 	@Override
-	public void start() {
-		doStart();
-	}
-
-	private void doStart() {
+	protected void doStart() {
 
 		PulsarContainerProperties containerProperties = getPulsarContainerProperties();
 
@@ -114,7 +108,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 	}
 
 	@Override
-	public void stop() {
+	public void doStop() {
 		setRunning(false);
 		this.logger.info("Pausing this consumer.");
 		this.listenerConsumer.consumer.pause();
@@ -127,15 +121,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
 		}
 	}
 
-	@Override
-	public boolean isRunning() {
-		return this.running;
-	}
-
-	protected void setRunning(boolean running) {
-		this.running = running;
-	}
-
 	@Override
 	public void destroy() {