Loading src/main/java/kafka/SonataApiKafkaReader.java +22 −18 Changes for src/main/java/kafka/SonataApiKafkaReader.java: 22 added lines, 18 removed lines. Original line number Diff line number Diff line package kafka; import static constants.SonataKafkaProcessesNamesConstants.END_USER_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.K8S_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS; import static org.etsi.osl.hypo.core.common.constants.SonataKafkaChannelsForApiInteraction.SONATA_CANCEL_PROCESS; import static org.etsi.osl.hypo.core.common.constants.SonataKafkaChannelsForApiInteraction.TMF_DELETE_PROCESS_INFO; import io.micrometer.common.util.StringUtils; import io.smallrye.common.annotation.Blocking; import jakarta.inject.Inject; import jakarta.enterprise.context.ApplicationScoped; import jakarta.ws.rs.WebApplicationException; import org.eclipse.microprofile.reactive.messaging.Channel; import org.eclipse.microprofile.reactive.messaging.Emitter; import org.eclipse.microprofile.reactive.messaging.Incoming; import org.eclipse.microprofile.rest.client.inject.RestClient; import org.etsi.osl.hypo.core.common.sonata.ProcessInfo; import org.jboss.logging.Logger; import rest.client.SonataApiRestClient; import org.kie.kogito.process.ProcessInstance; import org.kie.kogito.process.Processes; @ApplicationScoped public class SonataApiKafkaReader { private static final Logger LOGGER = Logger.getLogger(SonataApiKafkaReader.class); @Inject @RestClient SonataApiRestClient sonataApiRestClient; private final Processes processes; private final Emitter<ProcessInfo> processInfoEmitter; @Inject @Channel(TMF_DELETE_PROCESS_INFO) Emitter<ProcessInfo> processInfoEmitter; public SonataApiKafkaReader( Processes processes, @Channel(TMF_DELETE_PROCESS_INFO) Emitter<ProcessInfo> processInfoEmitter) { this.processes = processes; this.processInfoEmitter = processInfoEmitter; } @Incoming(SONATA_CANCEL_PROCESS) @Blocking Loading @@ -40,15 +42,10 @@ public class SonataApiKafkaReader { ProcessInfo processInfo) { final String processId = processInfo.getSonataProcessReferenceId().toString(); final String processType = processInfo.getProcessName(); if (!StringUtils.isBlank(processType)) { if (StringUtils.isNotBlank(processType) && StringUtils.isNotBlank(processId)) { try { if (processType.equalsIgnoreCase(SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS)) { sonataApiRestClient.deleteServiceOrderServiceActivationProcess(processId); } else if (processType.equalsIgnoreCase(END_USER_SERVICE_TERMINATION_SUBPROCESS)) { sonataApiRestClient.deleteEndUserServiceTerminationSubprocess(processId); } else if (processType.equalsIgnoreCase(K8S_SERVICE_TERMINATION_SUBPROCESS)) { sonataApiRestClient.deleteK8sServiceTerminationSubprocess(processId); } this.cancelAnyProcess(processType, processId); LOGGER.infof( "The process with process id %s was found and deleted successfully", processId); publishMessageDeleteProcessInfo(processInfo); Loading @@ -67,4 +64,11 @@ public class SonataApiKafkaReader { "A message has been published to delete process info with id %s in channel %s", processInfo.getSonataProcessReferenceId(), TMF_DELETE_PROCESS_INFO); } public void cancelAnyProcess(String processId, String instanceId) { org.kie.kogito.process.Process<?> process = processes.processById(processId); if (process != null) { process.instances().findById(instanceId).ifPresent(ProcessInstance::abort); } } } src/main/java/rest/client/SonataApiRestClient.javadeleted 100644 → 0 +0 −32 Changes for src/main/java/rest/client/SonataApiRestClient.java: 0 added lines, 32 removed lines. Original line number Diff line number Diff line package rest.client; import static constants.SonataKafkaProcessesNamesConstants.END_USER_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.K8S_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.DELETE; import jakarta.ws.rs.Path; import jakarta.ws.rs.PathParam; import jakarta.ws.rs.Produces; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import org.eclipse.microprofile.rest.client.inject.RegisterRestClient; @RegisterRestClient(configKey = "sonata-api") @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) public interface SonataApiRestClient { @Path(SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS + "/{id}") @DELETE Response deleteServiceOrderServiceActivationProcess(@PathParam("id") String id); @Path(END_USER_SERVICE_TERMINATION_SUBPROCESS + "/{id}") @DELETE Response deleteEndUserServiceTerminationSubprocess(@PathParam("id") String id); @Path(K8S_SERVICE_TERMINATION_SUBPROCESS + "/{id}") @DELETE Response deleteK8sServiceTerminationSubprocess(@PathParam("id") String id); } Loading
src/main/java/kafka/SonataApiKafkaReader.java +22 −18 Changes for src/main/java/kafka/SonataApiKafkaReader.java: 22 added lines, 18 removed lines. Original line number Diff line number Diff line package kafka; import static constants.SonataKafkaProcessesNamesConstants.END_USER_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.K8S_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS; import static org.etsi.osl.hypo.core.common.constants.SonataKafkaChannelsForApiInteraction.SONATA_CANCEL_PROCESS; import static org.etsi.osl.hypo.core.common.constants.SonataKafkaChannelsForApiInteraction.TMF_DELETE_PROCESS_INFO; import io.micrometer.common.util.StringUtils; import io.smallrye.common.annotation.Blocking; import jakarta.inject.Inject; import jakarta.enterprise.context.ApplicationScoped; import jakarta.ws.rs.WebApplicationException; import org.eclipse.microprofile.reactive.messaging.Channel; import org.eclipse.microprofile.reactive.messaging.Emitter; import org.eclipse.microprofile.reactive.messaging.Incoming; import org.eclipse.microprofile.rest.client.inject.RestClient; import org.etsi.osl.hypo.core.common.sonata.ProcessInfo; import org.jboss.logging.Logger; import rest.client.SonataApiRestClient; import org.kie.kogito.process.ProcessInstance; import org.kie.kogito.process.Processes; @ApplicationScoped public class SonataApiKafkaReader { private static final Logger LOGGER = Logger.getLogger(SonataApiKafkaReader.class); @Inject @RestClient SonataApiRestClient sonataApiRestClient; private final Processes processes; private final Emitter<ProcessInfo> processInfoEmitter; @Inject @Channel(TMF_DELETE_PROCESS_INFO) Emitter<ProcessInfo> processInfoEmitter; public SonataApiKafkaReader( Processes processes, @Channel(TMF_DELETE_PROCESS_INFO) Emitter<ProcessInfo> processInfoEmitter) { this.processes = processes; this.processInfoEmitter = processInfoEmitter; } @Incoming(SONATA_CANCEL_PROCESS) @Blocking Loading @@ -40,15 +42,10 @@ public class SonataApiKafkaReader { ProcessInfo processInfo) { final String processId = processInfo.getSonataProcessReferenceId().toString(); final String processType = processInfo.getProcessName(); if (!StringUtils.isBlank(processType)) { if (StringUtils.isNotBlank(processType) && StringUtils.isNotBlank(processId)) { try { if (processType.equalsIgnoreCase(SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS)) { sonataApiRestClient.deleteServiceOrderServiceActivationProcess(processId); } else if (processType.equalsIgnoreCase(END_USER_SERVICE_TERMINATION_SUBPROCESS)) { sonataApiRestClient.deleteEndUserServiceTerminationSubprocess(processId); } else if (processType.equalsIgnoreCase(K8S_SERVICE_TERMINATION_SUBPROCESS)) { sonataApiRestClient.deleteK8sServiceTerminationSubprocess(processId); } this.cancelAnyProcess(processType, processId); LOGGER.infof( "The process with process id %s was found and deleted successfully", processId); publishMessageDeleteProcessInfo(processInfo); Loading @@ -67,4 +64,11 @@ public class SonataApiKafkaReader { "A message has been published to delete process info with id %s in channel %s", processInfo.getSonataProcessReferenceId(), TMF_DELETE_PROCESS_INFO); } public void cancelAnyProcess(String processId, String instanceId) { org.kie.kogito.process.Process<?> process = processes.processById(processId); if (process != null) { process.instances().findById(instanceId).ifPresent(ProcessInstance::abort); } } }
src/main/java/rest/client/SonataApiRestClient.javadeleted 100644 → 0 +0 −32 Changes for src/main/java/rest/client/SonataApiRestClient.java: 0 added lines, 32 removed lines. Original line number Diff line number Diff line package rest.client; import static constants.SonataKafkaProcessesNamesConstants.END_USER_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.K8S_SERVICE_TERMINATION_SUBPROCESS; import static constants.SonataKafkaProcessesNamesConstants.SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.DELETE; import jakarta.ws.rs.Path; import jakarta.ws.rs.PathParam; import jakarta.ws.rs.Produces; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import org.eclipse.microprofile.rest.client.inject.RegisterRestClient; @RegisterRestClient(configKey = "sonata-api") @Produces(MediaType.APPLICATION_JSON) @Consumes(MediaType.APPLICATION_JSON) public interface SonataApiRestClient { @Path(SERVICE_ORDER_SERVICE_ACTIVATION_PROCESS + "/{id}") @DELETE Response deleteServiceOrderServiceActivationProcess(@PathParam("id") String id); @Path(END_USER_SERVICE_TERMINATION_SUBPROCESS + "/{id}") @DELETE Response deleteEndUserServiceTerminationSubprocess(@PathParam("id") String id); @Path(K8S_SERVICE_TERMINATION_SUBPROCESS + "/{id}") @DELETE Response deleteK8sServiceTerminationSubprocess(@PathParam("id") String id); }