21.2. 扩展 KIE 服务器以使用自定义数据传输

流程

1. 创建一个空的 Maven 项目，并在项目的 pom.xml 文件中定义以下打包类型和依赖项：

   示例项目中的 pom.xml 文件示例

   <packaging>jar</packaging>
   
   <properties>
     <version.org.kie>7.67.0.Final-redhat-00024</version.org.kie>
   </properties>
   
   <dependencies>
     <dependency>
       <groupId>org.kie</groupId>
       <artifactId>kie-api</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.kie</groupId>
       <artifactId>kie-internal</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.kie.server</groupId>
       <artifactId>kie-server-api</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.kie.server</groupId>
       <artifactId>kie-server-services-common</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.kie.server</groupId>
       <artifactId>kie-server-services-drools</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.drools</groupId>
       <artifactId>drools-core</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.drools</groupId>
       <artifactId>drools-compiler</artifactId>
       <version>${version.org.kie}</version>
     </dependency>
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
       <version>1.7.25</version>
     </dependency>
     <dependency>
       <groupId>org.apache.mina</groupId>
       <artifactId>mina-core</artifactId>
       <version>2.1.3</version>
     </dependency>
   </dependencies>

   Copy to Clipboard Toggle word wrap

2. 在项目中的 Java 类中实施 org.kie.server.services.api.KieServerExtension 接口，如下例所示：

   KieServerExtension 接口的实现示例

   public class MinaDroolsKieServerExtension implements KieServerExtension {
   
       private static final Logger logger = LoggerFactory.getLogger(MinaDroolsKieServerExtension.class);
   
       public static final String EXTENSION_NAME = "Drools-Mina";
   
       private static final Boolean disabled = Boolean.parseBoolean(System.getProperty("org.kie.server.drools-mina.ext.disabled", "false"));
       private static final String MINA_HOST = System.getProperty("org.kie.server.drools-mina.ext.port", "localhost");
       private static final int MINA_PORT = Integer.parseInt(System.getProperty("org.kie.server.drools-mina.ext.port", "9123"));
   
       // Taken from dependency on the `Drools` extension:
       private KieContainerCommandService batchCommandService;
   
       // Specific to MINA:
       private IoAcceptor acceptor;
   
       public boolean isActive() {
           return disabled == false;
       }
   
       public void init(KieServerImpl kieServer, KieServerRegistry registry) {
   
           KieServerExtension droolsExtension = registry.getServerExtension("Drools");
           if (droolsExtension == null) {
               logger.warn("No Drools extension available, quitting...");
               return;
           }
   
           List<Object> droolsServices = droolsExtension.getServices();
           for( Object object : droolsServices ) {
               // If the given service is null (not configured), continue to the next service:
               if (object == null) {
                   continue;
               }
               if( KieContainerCommandService.class.isAssignableFrom(object.getClass()) ) {
                   batchCommandService = (KieContainerCommandService) object;
                   continue;
               }
           }
           if (batchCommandService != null) {
               acceptor = new NioSocketAcceptor();
               acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new TextLineCodecFactory( Charset.forName( "UTF-8" ))));
   
               acceptor.setHandler( new TextBasedIoHandlerAdapter(batchCommandService) );
               acceptor.getSessionConfig().setReadBufferSize( 2048 );
               acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10 );
               try {
                   acceptor.bind( new InetSocketAddress(MINA_HOST, MINA_PORT) );
   
                   logger.info("{} -- Mina server started at {} and port {}", toString(), MINA_HOST, MINA_PORT);
               } catch (IOException e) {
                   logger.error("Unable to start Mina acceptor due to {}", e.getMessage(), e);
               }
   
           }
       }
   
       public void destroy(KieServerImpl kieServer, KieServerRegistry registry) {
           if (acceptor != null) {
               acceptor.dispose();
               acceptor = null;
           }
           logger.info("{} -- Mina server stopped", toString());
       }
   
       public void createContainer(String id, KieContainerInstance kieContainerInstance, Map<String, Object> parameters) {
           // Empty, already handled by the `Drools` extension
   
       }
   
       public void disposeContainer(String id, KieContainerInstance kieContainerInstance, Map<String, Object> parameters) {
         // Empty, already handled by the `Drools` extension
   
       }
   
       public List<Object> getAppComponents(SupportedTransports type) {
           // Nothing for supported transports (REST or JMS)
           return Collections.emptyList();
       }
   
       public <T> T getAppComponents(Class<T> serviceType) {
   
           return null;
       }
   
       public String getImplementedCapability() {
           return "BRM-Mina";
       }
   
       public List<Object> getServices() {
           return Collections.emptyList();
       }
   
       public String getExtensionName() {
           return EXTENSION_NAME;
       }
   
       public Integer getStartOrder() {
           return 20;
       }
   
       @Override
       public String toString() {
           return EXTENSION_NAME + " KIE Server extension";
       }
   }

   Copy to Clipboard Toggle word wrap

   KieServerExtension 接口是 KIE 服务器可以用来为新的 MINA 传输提供额外功能的主扩展接口。接口由以下组件组成：

   KieServerExtension 接口概述

   public interface KieServerExtension {
   
       boolean isActive();
   
       void init(KieServerImpl kieServer, KieServerRegistry registry);
   
       void destroy(KieServerImpl kieServer, KieServerRegistry registry);
   
       void createContainer(String id, KieContainerInstance kieContainerInstance, Map<String, Object> parameters);
   
       void disposeContainer(String id, KieContainerInstance kieContainerInstance, Map<String, Object> parameters);
   
       List<Object> getAppComponents(SupportedTransports type);
   
       <T> T getAppComponents(Class<T> serviceType);
   
       String getImplementedCapability();  

   1

   
   
       List<Object> getServices();
   
       String getExtensionName();  

   2

   
   
       Integer getStartOrder();  

   3

   
   }

   Copy to Clipboard Toggle word wrap

   1

   指定此扩展涵盖的能力。能力在 KIE 服务器中必须是唯一的。

   2

   为扩展名定义人类可读的名称。

   3

   确定何时应启动指定的扩展。对于其他扩展具有依赖项的扩展，此设置不得与父设置冲突。例如，在这个示例中，这个自定义扩展依赖于 Drools 扩展，它的 StartOrder 设置为 0，因此这个自定义附加组件扩展必须大于 0（ 示例实现中的设置为 20 ）。

   在前面的 MinaDroolsKieServerExtension 示例实现中，init 方法是从 Drools 扩展收集服务的主要元素，用于引导 MINA 服务器。KieServerExtension 接口中的其他方法都可以与标准实施保持一致，以满足接口要求。

   TextBasedIoHandlerAdapter 类是响应传入请求的 MINA 服务器上的处理程序。

3. 为 MINA 服务器实施 text BasedIoHandlerAdapter 处理程序，如下例所示：

   TextBasedIoHandlerAdapter 处理程序的实现示例

   public class TextBasedIoHandlerAdapter extends IoHandlerAdapter {
   
       private static final Logger logger = LoggerFactory.getLogger(TextBasedIoHandlerAdapter.class);
   
       private KieContainerCommandService batchCommandService;
   
       public TextBasedIoHandlerAdapter(KieContainerCommandService batchCommandService) {
           this.batchCommandService = batchCommandService;
       }
   
       @Override
       public void messageReceived( IoSession session, Object message ) throws Exception {
           String completeMessage = message.toString();
           logger.debug("Received message '{}'", completeMessage);
           if( completeMessage.trim().equalsIgnoreCase("quit") || completeMessage.trim().equalsIgnoreCase("exit") ) {
               session.close(false);
               return;
           }
   
           String[] elements = completeMessage.split("\\|");
           logger.debug("Container id {}", elements[0]);
           try {
               ServiceResponse<String> result = batchCommandService.callContainer(elements[0], elements[1], MarshallingFormat.JSON, null);
   
               if (result.getType().equals(ServiceResponse.ResponseType.SUCCESS)) {
                   session.write(result.getResult());
                   logger.debug("Successful message written with content '{}'", result.getResult());
               } else {
                   session.write(result.getMsg());
                   logger.debug("Failure message written with content '{}'", result.getMsg());
               }
           } catch (Exception e) {
   
           }
       }
   }

   Copy to Clipboard Toggle word wrap

   在本例中，处理器类接收文本信息，并在 Drools 服务中执行它们。

   在使用 TextBasedIoHandlerAdapter 处理程序实现时，请考虑以下处理器要求和行为：

   • 您提交到处理程序的任何对象都必须是一行，因为每个传入的传输请求都是一行。
   • 您必须在此一行中传递 KIE 容器 ID，以便处理程序需要格式 containerID|payload。
   • 您可以使用 marshaller 生成的方式设置响应。响应可以是多行。
   • 该处理程序支持 流模式，允许您发送命令而不与 KIE 服务器会话断开连接。要在流模式下结束 KIE 服务器会话，请将 exit 或 exit 命令发送至服务器。
4. 要使新数据传输可被 KIE 服务器发现，请在 文件中创建一个 META-INF/services/org.kie.server.api. KieServerExtension 文件，并在该文件中添加完全限定的类名称。在本例中，该文件包含一行 org.kie.server.ext.mina.MinastandaloneKieServerExtension。
5. 构建您的项目，并将生成的 JAR 文件复制到项目的 mina-core-2.0.9.jar 文件（此示例中取决于该扩展）到项目的 ~/kie-server.war/WEB-INF/lib 目录中。例如，在红帽 JBoss EAP 上，此目录的路径是 EAP_HOME/standalone/deployments/kie-server.war/WEB-INF/lib。

6. 启动 KIE 服务器并将构建的项目部署到正在运行的 KIE 服务器。您可以使用 Business Central 接口或 KIE 服务器 REST API 部署项目（ PUT 请求到 http://SERVER:PORT/kie-server/services/rest/server/containers/{containerId}）。

   在正在运行的 KIE 服务器上部署项目后，您可以在 KIE 服务器日志中查看新数据传输的状态，然后开始使用您的新数据传输：

   服务器日志中的新数据传输

   Drools-Mina KIE Server extension -- Mina server started at localhost and port 9123
   Drools-Mina KIE Server extension has been successfully registered as server extension

   Copy to Clipboard Toggle word wrap

   在本例中，您可以使用 Telnet 与 KIE 服务器中的新的基于 MINA 的数据传输进行交互：

   在命令终端中启动 Telnet 并连接到端口 9123 上的 KIE 服务器

   telnet 127.0.0.1 9123

   Copy to Clipboard Toggle word wrap

   在命令终端中与 KIE 服务器交互示例

   Trying 127.0.0.1...
   Connected to localhost.
   Escape character is '^]'.
   
   # Request body:
   demo|{"lookup":"defaultKieSession","commands":[{"insert":{"object":{"org.jbpm.test.Person":{"name":"john","age":25}}}},{"fire-all-rules":""}]}
   
   # Server response:
   {
     "results" : [ {
       "key" : "",
       "value" : 1
     } ],
     "facts" : [ ]
   }
   
   demo|{"lookup":"defaultKieSession","commands":[{"insert":{"object":{"org.jbpm.test.Person":{"name":"mary","age":22}}}},{"fire-all-rules":""}]}
   {
     "results" : [ {
       "key" : "",
       "value" : 1
     } ],
     "facts" : [ ]
   }
   
   demo|{"lookup":"defaultKieSession","commands":[{"insert":{"object":{"org.jbpm.test.Person":{"name":"james","age":25}}}},{"fire-all-rules":""}]}
   {
     "results" : [ {
       "key" : "",
       "value" : 1
     } ],
     "facts" : [ ]
   }
   exit
   Connection closed by foreign host.

   Copy to Clipboard Toggle word wrap

   服务器日志输出示例

   16:33:40,206 INFO  [stdout] (NioProcessor-2) Hello john
   16:34:03,877 INFO  [stdout] (NioProcessor-2) Hello mary
   16:34:19,800 INFO  [stdout] (NioProcessor-2) Hello james

   Copy to Clipboard Toggle word wrap