spring xd 传递文件名和记录数

Question

我有一个 spring xd 源组件，它从 s3 中提取文件并逐行溢出，因为每一行都是一条记录，接收器是日志。现在我想获取文件名和记录数或文件中的行（文件中的记录数）并将其作为 header 我如何在 xd 中实现此目的？

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-aws="http://www.springframework.org/schema/integration/aws"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
        http://www.springframework.org/schema/integration/aws http://www.springframework.org/schema/integration/aws/spring-integration-aws-1.0.xsd">

    <int:poller fixed-delay="${fixed-delay}" default="true"/>

    <bean id="credentials" class="org.springframework.integration.aws.core.BasicAWSCredentials">
        <property name="accessKey" value="${accessKey}"/>
        <property name="secretKey" value="${secretKey}"/>
    </bean>

    <bean
            class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="location">
            <value>dms-aws-s3-nonprod.properties</value>
        </property>
    </bean>

    <bean id="clientConfiguration" class="com.amazonaws.ClientConfiguration">
        <property name="proxyHost" value="${proxyHost}"/>
        <property name="proxyPort" value="${proxyPort}"/>
        <property name="preemptiveBasicProxyAuth" value="false"/>
    </bean>
<bean id="s3Operations" class="org.springframework.integration.aws.s3.core.CustomC1AmazonS3Operations">
    <constructor-arg index="0" ref="credentials"/>
    <constructor-arg index="1" ref="clientConfiguration"/>
    <property name="awsEndpoint" value="s3.amazonaws.com"/>
    <property name="temporaryDirectory" value="${temporaryDirectory}"/>
    <property name="awsSecurityKey"  value="${awsSecurityKey}"/>
</bean>



<!-- aws-endpoint="https://s3.amazonaws.com"  -->
<int-aws:s3-inbound-channel-adapter aws-endpoint="s3.amazonaws.com"
                                    bucket="${bucket}"
                                    s3-operations="s3Operations"
                                    credentials-ref="credentials"
                                    file-name-wildcard="${file-name-wildcard}"
                                    remote-directory="${remote-directory}"
                                    channel="splitChannel"
                                    local-directory="${local-directory}"
                                    accept-sub-folders="false"
                                    delete-source-files="true"
                                    archive-bucket="${archive-bucket}"
                                    archive-directory="${archive-directory}">
</int-aws:s3-inbound-channel-adapter>

<int:splitter input-channel="splitChannel" output-channel="output"
              expression="T(org.apache.commons.io.FileUtils).lineIterator(payload)"/>

<int:channel id="output"/>

和我的直播

xd-shell>stream create feedTest16 --definition "aws-s3-source | log" --deploy

Answer 1

考虑在此事上使用 FileSplitter。它是专门为这样的用例设计的：

<int-file:splitter input-channel="splitChannel" output-channel="output" markers="true"/>

FileSplitter.FileMarker END 消息将包含所需的 lineCount。

spring xd 传递文件名和记录数

spring xd passing file name and number of records

spring-integration

spring-xd