비동기 테스트 추가
This commit is contained in:
현성필
@@ -0,0 +1,73 @@
|
||||
package com.eactive.testmaster.queue.service;
|
||||
|
||||
import com.eactive.testmaster.config.HazelcastJSBridge;
|
||||
import com.eactive.testmaster.queue.entity.QueueHandler;
|
||||
import com.hazelcast.core.HazelcastInstance;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@Service
|
||||
public class DynamicQueueManager {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(DynamicQueueManager.class);
|
||||
|
||||
private final HazelcastInstance hazelcastInstance;
|
||||
private final Map<String, QueueListenerService> queueListeners;
|
||||
private final HazelcastJSBridge hazelcastJSBridge;
|
||||
|
||||
@Autowired
|
||||
public DynamicQueueManager(HazelcastInstance hazelcastInstance, HazelcastJSBridge hazelcastJSBridge) {
|
||||
this.hazelcastInstance = hazelcastInstance;
|
||||
this.hazelcastJSBridge = hazelcastJSBridge;
|
||||
this.queueListeners = new ConcurrentHashMap<>();
|
||||
}
|
||||
|
||||
/**
|
||||
* Start a listener for the specified queue
|
||||
*/
|
||||
// public void startQueueListener(QueueHandler queueHandler) {
|
||||
// if (!queueListeners.containsKey(queueHandler.getQueueName())) {
|
||||
// QueueListenerService listener = new QueueListenerService(hazelcastInstance, hazelcastJSBridge);
|
||||
// listener.setQueueHandler(queueHandler);
|
||||
// queueListeners.put(queueHandler.getQueueName(), listener);
|
||||
// listener.startListening();
|
||||
// logger.info("Started listener for queue: {}", queueHandler.getQueueName());
|
||||
// } else {
|
||||
// logger.warn("Listener already exists for queue: {}", queueHandler.getQueueName());
|
||||
// }
|
||||
// }
|
||||
public void startQueueListener(QueueHandler queueHandler) {
|
||||
QueueListenerService existingListener = queueListeners.get(queueHandler.getQueueName());
|
||||
if (existingListener != null) {
|
||||
// If exists but not running, remove it so we can start fresh
|
||||
if (!existingListener.isRunning()) {
|
||||
queueListeners.remove(queueHandler.getQueueName());
|
||||
} else {
|
||||
logger.warn("Listener already exists and running for queue: {}", queueHandler.getQueueName());
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
QueueListenerService listener = new QueueListenerService(hazelcastInstance, hazelcastJSBridge);
|
||||
listener.setQueueHandler(queueHandler);
|
||||
queueListeners.put(queueHandler.getQueueName(), listener);
|
||||
listener.startListening();
|
||||
logger.info("Started listener for queue: {}", queueHandler.getQueueName());
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop a queue listener
|
||||
*/
|
||||
public void stopQueueListener(String queueName) {
|
||||
QueueListenerService listener = queueListeners.remove(queueName);
|
||||
if (listener != null) {
|
||||
listener.stopListening();
|
||||
logger.info("Stopped listener for queue: {}", queueName);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.eactive.testmaster.queue.service;
|
||||
|
||||
import com.eactive.testmaster.queue.entity.QueueHandler;
|
||||
import com.eactive.testmaster.queue.repository.QueueHandlerRepository;
|
||||
import javax.transaction.Transactional;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.domain.Page;
|
||||
import org.springframework.data.domain.Pageable;
|
||||
import org.springframework.data.jpa.domain.Specification;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@Service
|
||||
@Transactional
|
||||
public class QueueHandlerService {
|
||||
|
||||
@Autowired
|
||||
private QueueHandlerRepository queueHandlerRepository;
|
||||
|
||||
@Autowired
|
||||
private DynamicQueueManager dynamicQueueManager;
|
||||
|
||||
public void initAllQueues() {
|
||||
queueHandlerRepository.findAll().forEach(queueHandler -> {
|
||||
dynamicQueueManager.startQueueListener(queueHandler);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
public Page<QueueHandler> findAll(Specification<QueueHandler> specification, Pageable pageable) {
|
||||
return queueHandlerRepository.findAll(specification, pageable);
|
||||
}
|
||||
|
||||
public QueueHandler findById(String id) {
|
||||
return queueHandlerRepository.findById(id)
|
||||
.orElseThrow(() -> new RuntimeException("Queue Handler not found with id: " + id));
|
||||
}
|
||||
|
||||
public QueueHandler create(QueueHandler queueHandler) {
|
||||
return queueHandlerRepository.save(queueHandler);
|
||||
}
|
||||
|
||||
public QueueHandler update(String queueName, QueueHandler queueHandler) {
|
||||
QueueHandler handler = queueHandlerRepository.findById(queueName).orElseThrow(() -> new IllegalArgumentException("Queue Handler not found with name: " + queueName));
|
||||
handler.setHandlerScript(queueHandler.getHandlerScript());
|
||||
return queueHandlerRepository.save(handler);
|
||||
}
|
||||
|
||||
public void deleteById(String id) {
|
||||
queueHandlerRepository.deleteById(id);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package com.eactive.testmaster.queue.service;
|
||||
|
||||
import com.eactive.testmaster.config.HazelcastJSBridge;
|
||||
import com.eactive.testmaster.loadtest.service.Console;
|
||||
import com.eactive.testmaster.queue.entity.QueueHandler;
|
||||
import com.hazelcast.collection.IQueue;
|
||||
import com.hazelcast.core.HazelcastInstance;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.PreDestroy;
|
||||
import javax.script.Bindings;
|
||||
import javax.script.ScriptEngine;
|
||||
import javax.script.ScriptEngineManager;
|
||||
import javax.script.ScriptException;
|
||||
import javax.script.SimpleBindings;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@Service
|
||||
public class QueueListenerService {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(QueueListenerService.class);
|
||||
|
||||
private IQueue<Object> taskQueue;
|
||||
private final ExecutorService executorService;
|
||||
private final AtomicBoolean running;
|
||||
private final HazelcastInstance hazelcastInstance;
|
||||
private final HazelcastJSBridge hazelcastJSBridge;
|
||||
private QueueHandler queueHandler;
|
||||
private final ScriptEngine scriptEngine; // Add ScriptEngine as a field
|
||||
private final Console console;
|
||||
|
||||
public void setQueueHandler(QueueHandler queueHandler) {
|
||||
this.queueHandler = queueHandler;
|
||||
}
|
||||
|
||||
@Autowired
|
||||
public QueueListenerService(HazelcastInstance hazelcastInstance, HazelcastJSBridge hazelcastJSBridge) {
|
||||
this.hazelcastInstance = hazelcastInstance;
|
||||
this.hazelcastJSBridge = hazelcastJSBridge;
|
||||
|
||||
ScriptEngineManager manager = new ScriptEngineManager();
|
||||
this.scriptEngine = manager.getEngineByName("nashorn");
|
||||
this.hazelcastJSBridge.enhanceEngine(this.scriptEngine);
|
||||
this.executorService = Executors.newSingleThreadExecutor();
|
||||
this.running = new AtomicBoolean(false);
|
||||
this.console = new Console();
|
||||
}
|
||||
|
||||
// @PostConstruct
|
||||
// public void startListening() {
|
||||
// if (queueHandler != null) {
|
||||
// taskQueue = hazelcastInstance.getQueue(queueHandler.getQueueName());
|
||||
// executorService.submit(this::processQueueItems);
|
||||
// logger.info("Queue listener started");
|
||||
// this.running.set(true);
|
||||
// }
|
||||
// }
|
||||
|
||||
// private void processQueueItems() {
|
||||
// while (running.get()) {
|
||||
// try {
|
||||
// Object item = taskQueue.poll();
|
||||
// if (item != null) {
|
||||
// processItem(item);
|
||||
// }
|
||||
// } catch (Exception e) {
|
||||
// logger.error("Error processing queue item", e);
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
@PostConstruct
|
||||
public void startListening() {
|
||||
if (queueHandler != null) {
|
||||
// Set running flag BEFORE starting the processing
|
||||
this.running.set(true);
|
||||
taskQueue = hazelcastInstance.getQueue(queueHandler.getQueueName());
|
||||
executorService.submit(this::processQueueItems);
|
||||
logger.info("Queue listener started for queue: {}", queueHandler.getQueueName());
|
||||
}
|
||||
}
|
||||
|
||||
private void processQueueItems() {
|
||||
while (running.get()) {
|
||||
try {
|
||||
// Use take() instead of poll() to block until an item is available
|
||||
Object item = taskQueue.take();
|
||||
processItem(item);
|
||||
} catch (InterruptedException e) {
|
||||
// Handle interruption (likely from shutdown)
|
||||
logger.info("Queue listener interrupted, ending processing");
|
||||
Thread.currentThread().interrupt();
|
||||
break;
|
||||
} catch (Exception e) {
|
||||
logger.error("Error processing queue item", e);
|
||||
}
|
||||
}
|
||||
logger.info("Queue listener stopped processing items");
|
||||
}
|
||||
|
||||
|
||||
private void processItem(Object message) {
|
||||
|
||||
String script = queueHandler.getHandlerScript();
|
||||
if (script != null && !script.isEmpty()) {
|
||||
Bindings scriptBindings = new SimpleBindings();
|
||||
scriptBindings.put("console", this.console);
|
||||
scriptBindings.put("message", message);
|
||||
|
||||
try {
|
||||
scriptEngine.eval(script, scriptBindings);
|
||||
} catch (ScriptException e) {
|
||||
logger.error("Error executing request script", e);
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isRunning() {
|
||||
return running.get();
|
||||
}
|
||||
|
||||
// @PreDestroy
|
||||
// public void stopListening() {
|
||||
// logger.info("Stopping queue listener...");
|
||||
// running.set(false);
|
||||
// executorService.shutdown();
|
||||
// try {
|
||||
// if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
// executorService.shutdownNow();
|
||||
// }
|
||||
// } catch (InterruptedException e) {
|
||||
// Thread.currentThread().interrupt();
|
||||
// executorService.shutdownNow();
|
||||
// }
|
||||
// logger.info("Queue listener stopped");
|
||||
//
|
||||
// }
|
||||
|
||||
@PreDestroy
|
||||
public void stopListening() {
|
||||
logger.info("Stopping queue listener...");
|
||||
running.set(false);
|
||||
// Interrupt the blocking take() call
|
||||
executorService.shutdownNow();
|
||||
try {
|
||||
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||
logger.warn("Queue listener did not terminate in time");
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
logger.info("Queue listener stopped");
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user