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 BATCH
  execution.parallelism = 1
  job.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;

    @Override
    public SeaTunnelRow deserialize(byte[] message) {
        return new SeaTunnelRow(new Object[] {new String(message)});
    }

    @Override
    public 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;
    @Override
    public String getPluginName() {
        return "Webservice";
    }

    @Override
    public 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 format
            String 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 SPI
                    throw 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);
        }
    }

    @Override
    public void setJobContext(JobContext jobContext) {
        this.jobContext = jobContext;
    }

    @Override
    public Boundedness getBoundedness() {
        return Boundedness.BOUNDED;
    }

    @Override
    public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
        return this.rowType;
    }

    @Override
    public 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);
    }
    @Override
    public void open() throws Exception {
        log.info("WebserviceSourceReader open");
    }

    @Override
    public void close() throws IOException {
        log.info("WebserviceSourceReader close");
    }

    @Override
    public 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 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.

-->
<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>