SeaTunnel扩展Source插件，自定义connector-webservice

代码结构

在seatunnel-connectors-v2中新建connector-webservice模块，可以直接赋值connector-http-base模块，webservice和http的方式比较类似，有些类直接复制了http中的代码。

核心类有WebserviceConfig，WebserviceParameter，WebserviceSource，WebserviceSourceReader

配置文件

env {# You can set engine configuration here STREAMING BATCHexecution.parallelism = 1job.mode = "BATCH"#execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
}source {# This is a example source plugin **only for test and demonstrate the feature source plugin**Webservice {url = "http://www.xxx.com.cn/xxx/WeatherWebService.asmx?wsdl"method = "getSupportCity"namespaceUri = "http://xxx.com.cn/"params = {"byProvinceName"="xxx"}result_table_name="table_3"}
}
transform {Sql {source_table_name = "table_3"result_table_name = "table_4"query = "select content  as fname from table_3"}
}
sink {Jdbc {removeDatabase="true"code="target"_compsName="sss"description=""mapType="1"source_table_name="table_4"writeMode="0"type="5"database="xxx"password="xxx"driver="com.mysql.cj.jdbc.Driver"url="jdbc:mysql://192.168.xxx:3306/xxx_test"pluginName="Jdbc"datasource="197"emptyType="2"user="xxx"table="xxx"generate_sink_sql="true"}
}

代码说明

WebserviceConfig

package org.apache.seatunnel.connectors.seatunnel.webservice.config;import lombok.Data;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;import java.io.Serializable;
import java.util.Map;@Data
public class WebserviceConfig  implements Serializable {public static final boolean DEFAULT_ENABLE_MULTI_LINES = false;public static final Option<String> FORMAT =Options.key("format").stringType().defaultValue("JSON").withDescription("Http response format");public static final Option<String> URL =Options.key("url").stringType().noDefaultValue().withDescription("Webservice request url");public static final Option<String> METHOD =Options.key("method").stringType().noDefaultValue().withDescription("Webservice request method");public static final Option<String> NAMESPACE_URI =Options.key("namespaceUri").stringType().noDefaultValue().withDescription("Webservice request namespaceUri");public static final Option<Map<String, String>> PARAMS =Options.key("params").mapType().noDefaultValue().withDescription("Webservice request params");}

WebserviceParameter

package org.apache.seatunnel.connectors.seatunnel.webservice.config;import lombok.Data;import java.io.Serializable;
import java.util.Map;
import java.util.stream.Collectors;import org.apache.seatunnel.shade.com.typesafe.config.Config;
@Data
public class WebserviceParameter implements Serializable {protected String url;protected String method;protected String namespaceUri;protected Map<String, String> params;protected String body;public void buildWithConfig(Config pluginConfig) {this.setUrl(pluginConfig.getString(WebserviceConfig.URL.key()));this.setMethod(pluginConfig.getString(WebserviceConfig.METHOD.key()));this.setNamespaceUri(pluginConfig.getString(WebserviceConfig.NAMESPACE_URI.key()));if (pluginConfig.hasPath(WebserviceConfig.PARAMS.key())) {this.setParams(pluginConfig.getConfig(WebserviceConfig.PARAMS.key()).entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey,entry -> String.valueOf(entry.getValue().unwrapped()),(v1, v2) -> v2)));}}}

DeserializationCollector

/** Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements.  See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License.  You may obtain a copy of the License at**    http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.apache.seatunnel.connectors.seatunnel.webservice.source;import lombok.AllArgsConstructor;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.format.json.JsonDeserializationSchema;import java.io.IOException;@AllArgsConstructor
public class DeserializationCollector {private DeserializationSchema<SeaTunnelRow> deserializationSchema;public void collect(byte[] message, Collector<SeaTunnelRow> out) throws IOException {if (deserializationSchema instanceof JsonDeserializationSchema) {((JsonDeserializationSchema) deserializationSchema).collect(message, out);} else {SeaTunnelRow deserialize = deserializationSchema.deserialize(message);out.collect(deserialize);}}
}

SimpleTextDeserializationSchema

/** Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements.  See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License.  You may obtain a copy of the License at**    http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/package org.apache.seatunnel.connectors.seatunnel.webservice.source;import lombok.AllArgsConstructor;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;@AllArgsConstructor
public class SimpleTextDeserializationSchema implements DeserializationSchema<SeaTunnelRow> {private SeaTunnelRowType rowType;@Overridepublic SeaTunnelRow deserialize(byte[] message) {return new SeaTunnelRow(new Object[] {new String(message)});}@Overridepublic SeaTunnelDataType<SeaTunnelRow> getProducedType() {return rowType;}
}

WebserviceSource

package org.apache.seatunnel.connectors.seatunnel.webservice.source;import com.google.auto.service.AutoService;
import org.apache.seatunnel.api.common.JobContext;
import org.apache.seatunnel.api.common.PrepareFailException;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.source.SeaTunnelSource;
import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.config.CheckConfigUtil;
import org.apache.seatunnel.common.config.CheckResult;
import org.apache.seatunnel.common.constants.PluginType;
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource;
import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
import org.apache.seatunnel.connectors.seatunnel.webservice.config.WebserviceConfig;
import org.apache.seatunnel.connectors.seatunnel.webservice.config.WebserviceParameter;
import org.apache.seatunnel.format.json.JsonDeserializationSchema;
import org.apache.seatunnel.shade.com.typesafe.config.Config;@AutoService(SeaTunnelSource.class)
public class WebserviceSource extends AbstractSingleSplitSource<SeaTunnelRow> {protected final WebserviceParameter webserviceParameter = new WebserviceParameter();protected SeaTunnelRowType rowType;protected JobContext jobContext;protected String contentField;protected DeserializationSchema<SeaTunnelRow> deserializationSchema;@Overridepublic String getPluginName() {return "Webservice";}@Overridepublic void prepare(Config pluginConfig) throws PrepareFailException {CheckResult result = CheckConfigUtil.checkAllExists(pluginConfig, WebserviceConfig.URL.key());if (!result.isSuccess()) {throw new RuntimeException(String.format("PluginName: %s, PluginType: %s, Message: %s",getPluginName(), PluginType.SOURCE, result.getMsg()));}this.webserviceParameter.buildWithConfig(pluginConfig);buildSchemaWithConfig(pluginConfig);}protected void buildSchemaWithConfig(Config pluginConfig) {if (pluginConfig.hasPath(CatalogTableUtil.SCHEMA.key())) {this.rowType = CatalogTableUtil.buildWithConfig(pluginConfig).getSeaTunnelRowType();// default use json formatString format = WebserviceConfig.FORMAT.defaultValue();if (pluginConfig.hasPath(WebserviceConfig.FORMAT.key())) {format = pluginConfig.getString(WebserviceConfig.FORMAT.key());}switch (format.toLowerCase()) {case "json":this.deserializationSchema =new JsonDeserializationSchema(false, false, rowType);break;default:// TODO: use format SPIthrow new RuntimeException(String.format("Unsupported data format [%s], http connector only support json format now",format));}} else {this.rowType = CatalogTableUtil.buildSimpleTextSchema();this.deserializationSchema = new SimpleTextDeserializationSchema(this.rowType);}}@Overridepublic void setJobContext(JobContext jobContext) {this.jobContext = jobContext;}@Overridepublic Boundedness getBoundedness() {return Boundedness.BOUNDED;}@Overridepublic SeaTunnelDataType<SeaTunnelRow> getProducedType() {return this.rowType;}@Overridepublic AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext readerContext) throws Exception {return new WebserviceSourceReader(webserviceParameter, readerContext,deserializationSchema,contentField);}
}

WebserviceSourceReader

package org.apache.seatunnel.connectors.seatunnel.webservice.source;import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.template.Template;
import cn.hutool.http.webservice.SoapClient;
import cn.hutool.http.webservice.SoapProtocol;
import lombok.extern.slf4j.Slf4j;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.source.Boundedness;
import org.apache.seatunnel.api.source.Collector;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.common.utils.JsonUtils;
import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader;
import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
import com.jayway.jsonpath.Configuration;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.Option;
import com.jayway.jsonpath.ReadContext;
import org.apache.seatunnel.connectors.seatunnel.webservice.config.WebserviceParameter;import java.io.IOException;
import java.util.HashMap;@Slf4j
public class WebserviceSourceReader extends AbstractSingleSplitReader<SeaTunnelRow> {protected final SingleSplitReaderContext context;private static final Option[] DEFAULT_OPTIONS = {Option.SUPPRESS_EXCEPTIONS, Option.ALWAYS_RETURN_LIST, Option.DEFAULT_PATH_LEAF_TO_NULL};private final String contentJson;private final Configuration jsonConfiguration =Configuration.defaultConfiguration().addOptions(DEFAULT_OPTIONS);protected final WebserviceParameter webserviceParameter;private final DeserializationCollector deserializationCollector;public WebserviceSourceReader(WebserviceParameter webserviceParameter,SingleSplitReaderContext context,DeserializationSchema<SeaTunnelRow> deserializationSchema,String contentJson) {this.webserviceParameter = webserviceParameter;this.context = context;this.contentJson = contentJson;this.deserializationCollector = new DeserializationCollector(deserializationSchema);}@Overridepublic void open() throws Exception {log.info("WebserviceSourceReader open");}@Overridepublic void close() throws IOException {log.info("WebserviceSourceReader close");}@Overridepublic void pollNext(Collector<SeaTunnelRow> output) throws Exception {try {SoapClient client = SoapClient.create(webserviceParameter.getUrl()).setMethod(webserviceParameter.getMethod(), webserviceParameter.getNamespaceUri());for (String key : webserviceParameter.getParams().keySet()) {String param = webserviceParameter.getParams().get(key);client = client.setParam(key, param);}String result = client.send(false);
//        deserializationCollector.collect(result.getBytes(), output);SeaTunnelRow seaTunnelRow = new SeaTunnelRow(new Object[]{getSoapBody(result)});output.collect(seaTunnelRow);log.info("WebserviceSourceReader pollNext");} catch (Exception e) {e.printStackTrace();} finally {if (Boundedness.BOUNDED.equals(context.getBoundedness())) {// signal to the source that we have reached the end of the data.log.info("Closed the bounded http source");context.signalNoMoreElement();}}}public String getSoapBody(String xml) {if (xml.indexOf("<soap:Body>") != -1) {return StrUtil.subBetween(xml, "<soap:Body>", "</soap:Body>");} else {return StrUtil.subBetween(xml, "<soap12:Body>", "</soap12:Body>");}}
}

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<!--Licensed to the Apache Software Foundation (ASF) under one or morecontributor license agreements.  See the NOTICE file distributed withthis work for additional information regarding copyright ownership.The ASF licenses this file to You under the Apache License, Version 2.0(the "License"); you may not use this file except in compliance withthe License.  You may obtain a copy of the License athttp://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an "AS IS" BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.apache.seatunnel</groupId><artifactId>seatunnel-connectors-v2</artifactId><version>2.3.3-SNAPSHOT</version></parent><artifactId>connector-webservice</artifactId><name>SeaTunnel : Connectors V2 : Webservice</name><properties><rabbitmq.version>5.9.0</rabbitmq.version><json-path.version>2.7.0</json-path.version></properties><dependencies><dependency><groupId>org.apache.seatunnel</groupId><artifactId>connector-common</artifactId><version>${project.version}</version></dependency><dependency><groupId>org.apache.seatunnel</groupId><artifactId>seatunnel-format-json</artifactId><version>${project.version}</version></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.22</version></dependency><dependency><groupId>com.jayway.jsonpath</groupId><artifactId>json-path</artifactId><version>${json-path.version}</version></dependency></dependencies><scm><tag>2.3.2</tag></scm>
</project>