Implemented the executor removed event over mqtt #60
|
@ -0,0 +1,41 @@
|
||||||
|
package ch.unisg.executorpool.adapter.out.messaging;
|
||||||
|
|
||||||
|
import ch.unisg.executorpool.adapter.common.clients.TapasMqttClient;
|
||||||
|
import ch.unisg.executorpool.adapter.common.formats.ExecutorJsonRepresentation;
|
||||||
|
import ch.unisg.executorpool.application.port.out.ExecutorRemovedEventPort;
|
||||||
|
import ch.unisg.executorpool.domain.ExecutorAddedEvent;
|
||||||
|
import ch.unisg.executorpool.domain.ExecutorRemovedEvent;
|
||||||
|
import org.apache.logging.log4j.LogManager;
|
||||||
|
import org.apache.logging.log4j.Logger;
|
||||||
|
import org.eclipse.paho.client.mqttv3.MqttException;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.annotation.Primary;
|
||||||
|
import org.springframework.core.env.Environment;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
@Component
|
||||||
|
@Primary
|
||||||
|
public class PublishExecutorRemovedEventAdapter implements ExecutorRemovedEventPort {
|
||||||
|
|
||||||
|
private static final Logger LOGGER = LogManager.getLogger(PublishExecutorAddedEventAdapter.class);
|
||||||
|
|
||||||
|
// TODO Can't autowire. Find fix
|
||||||
|
/*
|
||||||
|
@Autowired
|
||||||
|
private ConfigProperties config;
|
||||||
|
*/
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private Environment environment;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void publishExecutorRemovedEvent(ExecutorRemovedEvent event){
|
||||||
|
try{
|
||||||
|
var mqttClient = TapasMqttClient.getInstance(environment.getProperty("mqtt.broker.uri"));
|
||||||
|
mqttClient.publishMessage("ch/unisg/tapas/executors/removed", ExecutorJsonRepresentation.serialize(event.getExecutorClass()));
|
||||||
|
}
|
||||||
|
catch (MqttException e){
|
||||||
|
LOGGER.error(e.getMessage(), e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -0,0 +1,7 @@
|
||||||
|
package ch.unisg.executorpool.application.port.out;
|
||||||
|
|
||||||
|
import ch.unisg.executorpool.domain.ExecutorRemovedEvent;
|
||||||
|
|
||||||
|
public interface ExecutorRemovedEventPort {
|
||||||
|
void publishExecutorRemovedEvent(ExecutorRemovedEvent event);
|
||||||
|
}
|
|
@ -2,21 +2,36 @@ package ch.unisg.executorpool.application.service;
|
||||||
|
|
||||||
import ch.unisg.executorpool.application.port.in.RemoveExecutorFromExecutorPoolCommand;
|
import ch.unisg.executorpool.application.port.in.RemoveExecutorFromExecutorPoolCommand;
|
||||||
import ch.unisg.executorpool.application.port.in.RemoveExecutorFromExecutorPoolUseCase;
|
import ch.unisg.executorpool.application.port.in.RemoveExecutorFromExecutorPoolUseCase;
|
||||||
|
import ch.unisg.executorpool.application.port.out.ExecutorRemovedEventPort;
|
||||||
import ch.unisg.executorpool.domain.ExecutorClass;
|
import ch.unisg.executorpool.domain.ExecutorClass;
|
||||||
import ch.unisg.executorpool.domain.ExecutorPool;
|
import ch.unisg.executorpool.domain.ExecutorPool;
|
||||||
|
import ch.unisg.executorpool.domain.ExecutorRemovedEvent;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
import javax.transaction.Transactional;
|
import javax.transaction.Transactional;
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
|
||||||
@RequiredArgsConstructor
|
|
||||||
@Component
|
@Component
|
||||||
@Transactional
|
@Transactional
|
||||||
public class RemoveExecutorFromExecutorPoolService implements RemoveExecutorFromExecutorPoolUseCase {
|
public class RemoveExecutorFromExecutorPoolService implements RemoveExecutorFromExecutorPoolUseCase {
|
||||||
|
|
||||||
|
private final ExecutorRemovedEventPort executorRemovedEventPort;
|
||||||
|
|
||||||
|
public RemoveExecutorFromExecutorPoolService(ExecutorRemovedEventPort executorRemovedEventPort){
|
||||||
|
this.executorRemovedEventPort = executorRemovedEventPort;
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Optional<ExecutorClass> removeExecutorFromExecutorPool(RemoveExecutorFromExecutorPoolCommand command){
|
public Optional<ExecutorClass> removeExecutorFromExecutorPool(RemoveExecutorFromExecutorPoolCommand command){
|
||||||
ExecutorPool executorPool = ExecutorPool.getExecutorPool();
|
ExecutorPool executorPool = ExecutorPool.getExecutorPool();
|
||||||
return executorPool.removeExecutorByIpAndPort(command.getExecutorUri());
|
var removedExecutor = executorPool.removeExecutorByIpAndPort(command.getExecutorUri());
|
||||||
|
|
||||||
|
if(removedExecutor.isPresent()){
|
||||||
|
var executorRemovedEvent = new ExecutorRemovedEvent(removedExecutor.get());
|
||||||
|
executorRemovedEventPort.publishExecutorRemovedEvent(executorRemovedEvent);
|
||||||
|
}
|
||||||
|
|
||||||
|
return removedExecutor;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,11 @@
|
||||||
|
package ch.unisg.executorpool.domain;
|
||||||
|
|
||||||
|
import ch.unisg.executorpool.domain.ExecutorClass;
|
||||||
|
import lombok.Getter;
|
||||||
|
|
||||||
|
public class ExecutorRemovedEvent {
|
||||||
|
@Getter
|
||||||
|
private ExecutorClass executorClass;
|
||||||
|
|
||||||
|
public ExecutorRemovedEvent(ExecutorClass executorClass) { this.executorClass = executorClass; }
|
||||||
|
}
|
Loading…
Reference in New Issue
Block a user