nheron/kie-platform
nheron/kie-platform
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Use Kafka

Browse source
This commit is contained in:
Nicolas Héron 2020-06-26 08:54:45 +02:00
commit 033628c336
14 changed files with 340 additions and 149 deletions
Download patch file Download diff file Expand all files Collapse all files

48
drools-framework-admin-console-parent/drools-framework-admin-console/src/main/java/org/chtijbug/drools/console/DroolsSpringBootConsoleApplication.java
View file

@ -3,8 +3,12 @@ package org.chtijbug.drools.console;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.chtijbug.drools.KieContainerResponse;
import org.chtijbug.drools.KieContainerUpdate;
import org.chtijbug.drools.ReverseProxyUpdate;
import org.chtijbug.drools.common.KafkaTopicConstants;
import org.chtijbug.drools.console.middle.DababaseContentInit;
@ -23,10 +27,9 @@ import org.springframework.context.annotation.PropertySource;
import org.springframework.context.event.EventListener;
import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import java.util.HashMap;
@ -92,12 +95,49 @@ public class DroolsSpringBootConsoleApplication extends SpringBootServletInitial
JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public ProducerFactory<String, KieContainerUpdate> producerKieContainerUpdateFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
configProps.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaTemplate<String, ReverseProxyUpdate> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public KafkaTemplate<String, KieContainerUpdate> kafkaKieContainerUpdateTemplate() {
return new KafkaTemplate<>(producerKieContainerUpdateFactory());
}
public ConsumerFactory<String, KieContainerResponse> greetingConsumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG,"Console");
return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(KieContainerResponse.class));
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, KieContainerResponse>
ruleKafkaListenerKieContainerUpdateFactory() {
ConcurrentKafkaListenerContainerFactory<String, KieContainerResponse> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(greetingConsumerFactory());
return factory;
}
public static void main(String[] args) {
SpringApplication.run(DroolsSpringBootConsoleApplication.class, args);
}

19
drools-framework-admin-console-parent/drools-framework-admin-console/src/main/java/org/chtijbug/drools/console/service/ProjectPersistService.java
View file

@ -2,6 +2,7 @@ package org.chtijbug.drools.console.service;
import com.vaadin.flow.component.UI;
import com.vaadin.flow.server.VaadinSession;
import org.chtijbug.drools.KieContainerUpdate;
import org.chtijbug.drools.ReverseProxyUpdate;
import org.chtijbug.drools.common.KafkaTopicConstants;
import org.chtijbug.drools.console.AddLog;
@ -62,6 +63,9 @@ public class ProjectPersistService {
@Autowired
private KafkaTemplate<String, ReverseProxyUpdate> kafkaTemplateProxyUpdate;
@Autowired
private KafkaTemplate<String, KieContainerUpdate> kafkaKieContainerUpdateTemplate;
public ProjectPersistService() {
this.config = AppContext.getApplicationContext().getBean(KieConfigurationData.class);
@ -169,6 +173,8 @@ public class ProjectPersistService {
return projectPersist;
}
public void waitForJobToBeEnded(String url, String username, String password, ProjectPersist projectPersist, AddLog workOnGoingView, UI ui) {
UserConnected userConnected = userConnectedService.getUserConnected();
@ -187,8 +193,11 @@ public class ProjectPersistService {
executeWrite(url, username, password, workOnGoingView, result.getJobId(), ui);
for (String serverName : projectPersist.getServerNames()) {
List<ContainerRuntimePojoPersist> existingContainers = containerRuntimeRepository.findByServerNameAndContainerId(serverName, projectPersist.getContainerID());
if (!existingContainers.isEmpty()) {
for (ContainerRuntimePojoPersist containerRuntimePojoPersist : existingContainers) {
@ -208,7 +217,15 @@ public class ProjectPersistService {
}
KieContainerUpdate kieContainerUpdate = new KieContainerUpdate();
kieContainerUpdate.setMainClass(projectPersist.getMainClass());
kieContainerUpdate.setArtifactID(projectPersist.getArtifactID());
kieContainerUpdate.setGroupID(projectPersist.getGroupID());
kieContainerUpdate.setProjectVersion(projectPersist.getProjectVersion());
kieContainerUpdate.setContainerID(projectPersist.getContainerID());
kieContainerUpdate.setAction(KieContainerUpdate.STATUS.TODEPLOY);
kafkaKieContainerUpdateTemplate.send(serverName,kieContainerUpdate);
workOnGoingView.addRow("Deploy Request="+kieContainerUpdate,ui);
}
}
};

15
drools-framework-admin-console-parent/drools-framework-admin-console/src/main/java/org/chtijbug/drools/console/vaadinComponent/leftMenu/Action/DeploymentAction.java
View file

@ -3,6 +3,8 @@ package org.chtijbug.drools.console.vaadinComponent.leftMenu.Action;
import com.vaadin.flow.component.button.Button;
import com.vaadin.flow.component.icon.VaadinIcon;
import com.vaadin.flow.component.orderedlayout.VerticalLayout;
import org.chtijbug.drools.KieContainerResponse;
import org.chtijbug.drools.common.KafkaTopicConstants;
import org.chtijbug.drools.console.service.ProjectPersistService;
import org.chtijbug.drools.console.service.util.AppContext;
import org.chtijbug.drools.console.vaadinComponent.ComponentPerso.DialogPerso;
@ -11,6 +13,7 @@ import org.chtijbug.drools.console.vaadinComponent.componentView.AssociateProjec
import org.chtijbug.drools.console.vaadinComponent.componentView.DefineProject;
import org.chtijbug.drools.console.view.DeploymentView;
import org.chtijbug.drools.proxy.persistence.model.ProjectPersist;
import org.springframework.kafka.annotation.KafkaListener;
public class DeploymentAction extends VerticalLayout {
@ -22,11 +25,11 @@ public class DeploymentAction extends VerticalLayout {
private Button deployer;
private ProjectPersistService projectPersistService;
private DeploymentView deploymentView;
public DeploymentAction(SqueletteComposant squeletteComposant,DeploymentView deploymentView){
setClassName("leftMenu-global-action");
this.deploymentView=deploymentView;
projectPersistService= AppContext.getApplicationContext().getBean(ProjectPersistService.class);
definirProject =new Button("Define your project", VaadinIcon.CODE.create());
@ -76,6 +79,14 @@ public class DeploymentAction extends VerticalLayout {
});
}
@KafkaListener(
topics = KafkaTopicConstants.RESPONSE_DEPLOY_TOPIC,groupId = "Console",
containerFactory = "ruleKafkaListenerKieContainerUpdateFactory")
public void updateKieServerLogInfo(KieContainerResponse kieContainerResponse){
deploymentView.addRow("Deploy Response="+kieContainerResponse.toString(),getUI().get());
}
private boolean isActive(Button button){
return button.getClassNames().contains("active");
}

1
drools-framework-base-tools-parent/drools-framework-common/src/main/java/org/chtijbug/drools/common/KafkaTopicConstants.java
View file

@ -3,5 +3,6 @@ package org.chtijbug.drools.common;
public class KafkaTopicConstants {
public final static String LOGING_TOPIC ="logging";
public final static String RESPONSE_TOPIC ="Response";
public final static String RESPONSE_DEPLOY_TOPIC ="ResponseDeploy";
public final static String REVERSE_PROXY="proxy";
}

10
drools-framework-base-tools-parent/drools-framework-kie-server-parent/drools-framework-kie-server-services-drools/src/main/java/org/chtijbug/kieserver/services/drools/DroolsChtijbugKieServerExtension.java
View file

@ -19,7 +19,6 @@ import org.chtijbug.drools.kieserver.extension.KieServerAddOnElement;
import org.chtijbug.drools.kieserver.extension.KieServerGlobalVariableDefinition;
import org.chtijbug.drools.kieserver.extension.KieServerListenerDefinition;
import org.chtijbug.drools.kieserver.extension.KieServerLoggingDefinition;
import org.chtijbug.kieserver.services.drools.sftp.SftpServerService;
import org.drools.compiler.kie.builder.impl.InternalKieModule;
import org.drools.compiler.kie.builder.impl.KieRepositoryImpl;
import org.kie.api.builder.KieRepository;
@ -34,7 +33,6 @@ import org.slf4j.LoggerFactory;
import javax.xml.bind.annotation.XmlRootElement;
import javax.xml.bind.annotation.XmlType;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.util.*;
@ -54,7 +52,7 @@ public class DroolsChtijbugKieServerExtension implements KieServerExtension {
private boolean initialized = false;
private SftpServerService sftpServerService;
@Override
public boolean isInitialized() {
@ -73,12 +71,6 @@ public class DroolsChtijbugKieServerExtension implements KieServerExtension {
this.registry = registry;
services.add(rulesExecutionService);
initialized = true;
sftpServerService = new SftpServerService();
try {
sftpServerService.initServer();
} catch (IOException e) {
logger.error("Impossible to create sftp server", e);
}
}
private void initExtensionsList() {

83
drools-framework-base-tools-parent/drools-framework-kie-server-parent/drools-framework-kie-server-services-drools/src/main/java/org/chtijbug/kieserver/services/drools/sftp/SftpServerService.java
View file

@ -1,83 +0,0 @@
package org.chtijbug.kieserver.services.drools.sftp;
import org.apache.sshd.common.NamedFactory;
import org.apache.sshd.common.PropertyResolverUtils;
import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
import org.apache.sshd.server.Command;
import org.apache.sshd.server.ServerFactoryManager;
import org.apache.sshd.server.SshServer;
import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
import org.apache.sshd.server.scp.ScpCommandFactory;
import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.PostConstruct;
import javax.inject.Singleton;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.Arrays;
@Singleton
public class SftpServerService {
private final String BANNER =
"\n\nWelcome to Pymma Server kie server ssh Server!\n\n";
private Logger logger = LoggerFactory.getLogger(this.getClass());
private SshServer sshd;
private String server = "0.0.0.0";
private Integer port = -1;
private String login = "kieserver";
private String password = "kieserver1";
private String sftpDir = System.getProperty("org.chtijbug.server.tracedir");
public SftpServerService() {
}
@PostConstruct
public void initServer() throws IOException {
if (System.getProperty("org.chtijbug.server.sftpPort") != null) {
try {
port = Integer.valueOf(System.getProperty("org.chtijbug.server.sftpPort"));
} catch (NumberFormatException e) {
port = 9080;
}
}
if (System.getProperty("org.chtijbug.server.tracedir") != null) {
sftpDir = System.getProperty("org.chtijbug.server.tracedir");
}
sshd = SshServer.setUpDefaultServer();
PropertyResolverUtils.updateProperty(
sshd,
ServerFactoryManager.WELCOME_BANNER,
BANNER);
sshd.setHost(server);
sshd.setPort(port);
sshd.setPasswordAuthenticator(new MyPasswordAuthenticator(login, password));
sshd.setPublickeyAuthenticator(new MyPublickeyAuthenticator());
sshd.setKeyPairProvider(new SimpleGeneratorHostKeyProvider());
final String filePath = sftpDir;
File f = new File(filePath);
if (f.exists() == false) {
f.mkdir();
}
sshd.setFileSystemFactory(new VirtualFileSystemFactory(Paths.get(filePath)));
SftpSubsystemFactory factory = new SftpSubsystemFactory.Builder().build();
sshd.setSubsystemFactories(Arrays.<NamedFactory<Command>>asList(factory));
//sshd.setShellFactory(new ProcessShellFactory(new String[] { "/bin/sh", "-i", "-l" }));
sshd.setCommandFactory(new ScpCommandFactory());
// sshd.setShellFactory( new SshSessionFactory() );
sshd.start();
logger.info("Serveur SSH demarre sur host " + this.server + " port " + this.port);
}
public void setServer(String server) {
}
}

62
drools-framework-base-tools-parent/drools-framework-runtime-entity/src/main/java/org/chtijbug/drools/KieContainerResponse.java Normal file
View file

@ -0,0 +1,62 @@
package org.chtijbug.drools;
import java.util.ArrayList;
import java.util.List;
public class KieContainerResponse {
public enum STATUS {
ERROR,
SUCCESS
}
private KieContainerUpdate kieContainerUpdate;
private String messageError;
private STATUS status;
private List<String> errorMessages= new ArrayList<>();
public KieContainerUpdate getKieContainerUpdate() {
return kieContainerUpdate;
}
public void setKieContainerUpdate(KieContainerUpdate kieContainerUpdate) {
this.kieContainerUpdate = kieContainerUpdate;
}
public String getMessageError() {
return messageError;
}
public void setMessageError(String messageError) {
this.messageError = messageError;
}
public List<String> getErrorMessages() {
return errorMessages;
}
public void setErrorMessages(List<String> errorMessages) {
this.errorMessages = errorMessages;
}
public STATUS getStatus() {
return status;
}
public void setStatus(STATUS status) {
this.status = status;
}
@Override
public String toString() {
final StringBuffer sb = new StringBuffer("KieContainerResponse{");
sb.append("kieContainerUpdate=").append(kieContainerUpdate);
sb.append(", messageError='").append(messageError).append('\'');
sb.append(", status=").append(status);
sb.append('}');
return sb.toString();
}
}

94
drools-framework-base-tools-parent/drools-framework-runtime-entity/src/main/java/org/chtijbug/drools/KieContainerUpdate.java Normal file
View file

@ -0,0 +1,94 @@
package org.chtijbug.drools;
public class KieContainerUpdate {
public enum STATUS {
TODEPLOY,
TODELETE
}
private String mainClass;
private String groupID;
private String artifactID;
private String processID;
private String projectVersion;
private String containerID;
private STATUS action;
public String getMainClass() {
return mainClass;
}
public void setMainClass(String mainClass) {
this.mainClass = mainClass;
}
public String getGroupID() {
return groupID;
}
public void setGroupID(String groupID) {
this.groupID = groupID;
}
public String getArtifactID() {
return artifactID;
}
public void setArtifactID(String artifactID) {
this.artifactID = artifactID;
}
public String getProcessID() {
return processID;
}
public void setProcessID(String processID) {
this.processID = processID;
}
public String getProjectVersion() {
return projectVersion;
}
public void setProjectVersion(String projectVersion) {
this.projectVersion = projectVersion;
}
public String getContainerID() {
return containerID;
}
public void setContainerID(String containerID) {
this.containerID = containerID;
}
public STATUS getAction() {
return action;
}
public void setAction(STATUS action) {
this.action = action;
}
@Override
public String toString() {
final StringBuffer sb = new StringBuffer("KieContainerUpdate{");
sb.append("mainClass='").append(mainClass).append('\'');
sb.append(", groupID='").append(groupID).append('\'');
sb.append(", artifactID='").append(artifactID).append('\'');
sb.append(", processID='").append(processID).append('\'');
sb.append(", projectVersion='").append(projectVersion).append('\'');
sb.append(", containerID='").append(containerID).append('\'');
sb.append(", action=").append(action);
sb.append('}');
return sb.toString();
}
}

4
drools-framework-business-proxy-parent/drools-framework-business-indexer-app/src/main/java/org/chtijbug/drools/indexer/DroolsBusinessIndexerServer.java
View file

@ -45,11 +45,13 @@ public class DroolsBusinessIndexerServer {
private String bootstrapAddress;
@Value(value = "${kafka.index.groupid})")
private String groupID;
public ConsumerFactory<String, ChtijbugObjectRequest> greetingConsumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "greeting");
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupID);
return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(ChtijbugObjectRequest.class));
}
@Bean

3
drools-framework-business-proxy-parent/drools-framework-business-indexer-app/src/main/java/org/chtijbug/drools/indexer/service/StoreLoggingService.java
View file

@ -24,8 +24,9 @@ public class StoreLoggingService {
@Autowired
private BusinessTransactionActionRepository actionRepository;
@KafkaListener(
topics = KafkaTopicConstants.LOGING_TOPIC,
topics = KafkaTopicConstants.LOGING_TOPIC,groupId = "${kafka.index.groupid}",
containerFactory = "ruleKafkaListenerContainerFactory")
public void store(ChtijbugObjectRequest result) {
if (result != null) {

2
drools-framework-business-proxy-parent/drools-framework-business-indexer-app/src/main/resources/application.properties
View file

@ -2,7 +2,7 @@ kafka.bootstrapAddress=localhost:9092,localhost:9093,localhost:9094
kieserver.login=kieserver
kieserver.password=kieserver1
kafka.index.groupid=index1
spring.data.mongodb.database=businessProxyDB
spring.data.mongodb.host=localhost:28017
server.port=5547

56
drools-framework-business-proxy-parent/drools-framework-business-proxy-app/src/main/java/org/chtijbug/drools/proxy/DroolsBusinessProxyServer.java
View file

@ -18,9 +18,13 @@ package org.chtijbug.drools.proxy;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.chtijbug.drools.ChtijbugObjectRequest;
import org.chtijbug.drools.KieContainerResponse;
import org.chtijbug.drools.KieContainerUpdate;
import org.chtijbug.drools.common.KafkaTopicConstants;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
@ -28,10 +32,9 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import java.net.InetAddress;
@ -45,8 +48,8 @@ import java.util.Map;
@EnableKafka
public class DroolsBusinessProxyServer {
@Value(value = "${org.kie.server.id}")
private String groupID;
@Value(value = "${kafka.bootstrapAddress}")
private String bootstrapAddress;
@ -62,10 +65,12 @@ public class DroolsBusinessProxyServer {
public NewTopic loggingTopic() {
return new NewTopic(KafkaTopicConstants.LOGING_TOPIC, 1, (short) 1);
}
@Bean
public NewTopic actionResponseTopic() {
return new NewTopic(KafkaTopicConstants.RESPONSE_TOPIC, 1, (short) 1);
}
@Bean
public ProducerFactory<String, ChtijbugObjectRequest> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
@ -80,6 +85,45 @@ public class DroolsBusinessProxyServer {
JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public ProducerFactory<String, KieContainerResponse> producerKieContainerResponseactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapAddress);
configProps.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
JsonSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaTemplate<String, KieContainerResponse> kafkaKieContainerUpdateResponsableTemplate() {
return new KafkaTemplate<>(producerKieContainerResponseactory());
}
@Bean
public NewTopic actionDeployResponseTopic() {
return new NewTopic(KafkaTopicConstants.RESPONSE_DEPLOY_TOPIC, 1, (short) 1);
}
public ConsumerFactory<String, KieContainerUpdate> greetingConsumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.GROUP_ID_CONFIG, groupID);
return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(KieContainerUpdate.class));
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, KieContainerUpdate>
ruleKafkaListenerKieContainerUpdateFactory() {
ConcurrentKafkaListenerContainerFactory<String, KieContainerUpdate> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(greetingConsumerFactory());
return factory;
}
@Bean
public KafkaTemplate<String, ChtijbugObjectRequest> kafkaTemplate() {

12
drools-framework-business-proxy-parent/drools-framework-business-proxy-app/src/main/java/org/chtijbug/drools/proxy/camel/AutodeployRouter.java
View file

@ -1,12 +0,0 @@
package org.chtijbug.drools.proxy.camel;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;
@Component
public class AutodeployRouter extends RouteBuilder {
@Override
public void configure() throws Exception {
from("quartz2://myGroup/myTimerName?cron=0/5+*+*+?+*+*").to("bean:kieService?method=updateConfig()");
}
}

72
drools-framework-business-proxy-parent/drools-framework-business-proxy-app/src/main/java/org/chtijbug/drools/proxy/service/KieServiceCommon.java
View file

@ -18,6 +18,9 @@ package org.chtijbug.drools.proxy.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.camel.CamelContext;
import org.apache.camel.Route;
import org.chtijbug.drools.KieContainerResponse;
import org.chtijbug.drools.KieContainerUpdate;
import org.chtijbug.drools.common.KafkaTopicConstants;
import org.chtijbug.drools.proxy.PlatfomKieServerStateRepository;
import org.chtijbug.drools.proxy.camel.DroolsRouter;
import org.chtijbug.drools.proxy.persistence.model.ContainerPojoPersist;
@ -39,6 +42,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
@ -75,6 +80,9 @@ public class KieServiceCommon {
private String hostName = "localhost";
private Map<String, DroolsRouter> routes = new HashMap<>();
@Autowired
KafkaTemplate<String, KieContainerResponse> kafkaKieContainerUpdateResponseTemplate;
public KieServiceCommon() {
// for now, if no server impl is passed as parameter, create one
// this.server = KieServerLocator.getInstance();
@ -210,7 +218,7 @@ public class KieServiceCommon {
Set<Class<?>> classes = kieContainerInstance.getExtraClasses();
String className = container.getClassName();
Class foundClass = this.getClassFromName(classes, className);
if (foundClass!=null) {
if (foundClass != null) {
ClassLoader classLoader = foundClass.getClassLoader();
Class<?> theClass = classLoader.loadClass(className);
camelContext.setApplicationContextClassLoader(classLoader);
@ -264,41 +272,55 @@ public class KieServiceCommon {
return result.getResult();
}
public void updateConfig() throws Exception {
@KafkaListener(
topics = "${org.kie.server.id}", groupId = "${org.kie.server.id}",
containerFactory = "ruleKafkaListenerKieContainerUpdateFactory")
public void updateConfig(KieContainerUpdate update) {
try {
String serverName = KieServiceCommon.getKieServerID();
String isSwarm = System.getProperty("org.kie.server.swarm");
List<ContainerRuntimePojoPersist> containers = null;
containers = containerRuntimeRepository.findByServerNameAndStatusAndHostname(serverName, ContainerRuntimePojoPersist.STATUS.TODEPLOY.toString(), hostName);
for (ContainerRuntimePojoPersist element : containers) {
ContainerPojoPersist containerIds = containerRepository.findByServerNameAndContainerId(serverName, element.getContainerId());
if (containerIds != null) {
this.disposeContainer(element.getContainerId());
if (update.getAction().equals(KieContainerUpdate.STATUS.TODEPLOY)) {
this.disposeContainer(update.getContainerID());
KieContainerResource newContainer = new KieContainerResource();
newContainer.setContainerId(element.getContainerId());
newContainer.setContainerId(update.getContainerID());
newContainer.setReleaseId(new ReleaseId());
newContainer.getReleaseId().setArtifactId(containerIds.getArtifactId());
newContainer.getReleaseId().setGroupId(containerIds.getGroupId());
newContainer.getReleaseId().setVersion(containerIds.getVersion());
this.createContainer(element.getContainerId(), newContainer);
newContainer.getReleaseId().setArtifactId(update.getArtifactID());
newContainer.getReleaseId().setGroupId(update.getGroupID());
newContainer.getReleaseId().setVersion(update.getProjectVersion());
this.createContainer(update.getContainerID(), newContainer);
ContainerPojoPersist containerIds = containerRepository.findByServerNameAndContainerId(serverName, update.getContainerID());
this.initCamelBusinessRoute(containerIds);
List<ContainerRuntimePojoPersist> containers = containerRuntimeRepository.findByServerNameAndContainerId(serverName, update.getContainerID());
for (ContainerRuntimePojoPersist element : containers) {
element.setStatus(ContainerRuntimePojoPersist.STATUS.UP.toString());
containerRuntimeRepository.save(element);
}
}
containers = containerRuntimeRepository.findByServerNameAndStatusAndHostname(serverName, ContainerRuntimePojoPersist.STATUS.TODELETE.toString(), hostName);
} else {
this.disposeContainer(update.getContainerID());
this.deleteCamelBusinessRoute(update.getContainerID());
ContainerPojoPersist containerIds = containerRepository.findByServerNameAndContainerId(serverName, update.getContainerID());
List<ContainerRuntimePojoPersist> containers = containerRuntimeRepository.findByServerNameAndContainerId(serverName, update.getContainerID());
for (ContainerRuntimePojoPersist element : containers) {
ContainerPojoPersist containerIds = containerRepository.findByServerNameAndContainerId(serverName, element.getContainerId());
if (containerIds != null) {
this.disposeContainer(element.getContainerId());
this.deleteCamelBusinessRoute(element.getContainerId());
element.setStatus(ContainerRuntimePojoPersist.STATUS.UP.toString());
containerRuntimeRepository.save(element);
}
containerRepository.delete(containerIds);
}
}
KieContainerResponse kieContainerResponse = new KieContainerResponse();
kieContainerResponse.setStatus(KieContainerResponse.STATUS.SUCCESS);
kafkaKieContainerUpdateResponseTemplate.send(KafkaTopicConstants.RESPONSE_DEPLOY_TOPIC,kieContainerResponse);
}catch (Exception e){
KieContainerResponse kieContainerResponse = new KieContainerResponse();
kieContainerResponse.setStatus(KieContainerResponse.STATUS.ERROR);
kieContainerResponse.setKieContainerUpdate(update);
kieContainerResponse.setMessageError(e.getMessage());
for (StackTraceElement stackTraceElement : e.getStackTrace()){
kieContainerResponse.getErrorMessages().add(stackTraceElement.toString());
}
kafkaKieContainerUpdateResponseTemplate.send(KafkaTopicConstants.RESPONSE_DEPLOY_TOPIC,kieContainerResponse);
}
}