KieServerDataSetManager { public void registerInKieServer(@Observes final ServerInstanceRegistered serverInstanceRegistered) { final ServerInstance serverInstance = serverInstanceRegistered.getServerInstance(); final String serverInstanceId = serverInstance.getServerInstanceId(); final String serverTemplateId = serverInstance.getServerTemplateId(); LOGGER.info("Server instance '{}' connected, registering data sets", serverInstanceId); final ServerTemplate serverTemplate = specManagementService.getServerTemplate(serverTemplateId); if(serverTemplate.getCapabilities().contains(Capability.PROCESS.name()) == false){ LOGGER.info("Missing BPM capability for server instance '{}', not possible to register data set queries", serverInstanceId); event.fire(new KieServerDataSetRegistered(serverInstanceId, serverTemplateId)); return; } final List<DataSetDef> dataSetDefs = dataSetDefRegistry.getDataSetDefs(false); LOGGER.debug("Found {} data sets to register", dataSetDefs.size()); if (dataSetDefs.isEmpty()) { return; } this.executorService.execute(() -> { try { LOGGER.debug("Registering data set definitions on connected server instance '{}'", serverInstanceId); final Set<QueryDefinition> queryDefinitions = dataSetDefs.stream() .filter(dataSetDef -> dataSetDef.getProvider().getName().equals("REMOTE")) .map( dataSetDef -> QueryDefinition.builder() .name(dataSetDef.getUUID()) .expression(((SQLDataSetDef) dataSetDef).getDbSQL()) .source(((SQLDataSetDef) dataSetDef).getDataSource()) .target(extractQueryTarget(dataSetDef)) .build() ).collect(Collectors.toSet()); registerQueriesWithRetry(serverTemplateId, serverInstanceId, queryDefinitions); } catch (Exception e) { LOGGER.warn("Unable to register query definition on '{}' due to {}", serverInstanceId, e.getMessage(), e); } }); } @Inject  KieServerDataSetManager(DataSetDefRegistry dataSetDefRegistry,
                                   KieServerIntegration kieServerIntegration,
                                   Event<KieServerDataSetRegistered> event,
                                   @Managed ExecutorService executorService,
                                   SpecManagementService specManagementService); void registerInKieServer(@Observes final ServerInstanceRegistered serverInstanceRegistered);  }
@Test public void testRegisterKieServerWithoutProcessCapability() { final ServerInstance serverInstance = new ServerInstance(); serverInstance.setServerInstanceId("instanceId"); serverInstance.setServerTemplateId("templateId"); when(specManagementService.getServerTemplate(serverInstance.getServerTemplateId())).thenReturn(new ServerTemplate()); kieServerDataSetManager.registerInKieServer(new ServerInstanceRegistered(serverInstance)); assertEquals(1, receivedEvents.size()); verify(dataSetDefRegistry, never()).getDataSetDefs(false); }
@Test public void testRegisterKieServerWithProcessCapability() { final ServerInstance serverInstance = new ServerInstance(); serverInstance.setServerInstanceId("instanceId"); serverInstance.setServerTemplateId("templateId"); when(specManagementService.getServerTemplate(serverInstance.getServerTemplateId())).thenReturn(new ServerTemplate(serverInstance.getServerTemplateId(), null, singletonList(Capability.PROCESS.name()), emptyMap(), emptyList())); kieServerDataSetManager.registerInKieServer(new ServerInstanceRegistered(serverInstance)); assertEquals(0, receivedEvents.size()); verify(dataSetDefRegistry).getDataSetDefs(false); }