Spring Cloud DataFlow - How to use a custom TCP encoder/decoder in TCP source

Question

I have already asked this question for Spring XD. I am now trying to migrate to Spring CDF.

I found this link and I tried to reuse the code there and change the encoding with mine.

I created the following POM:

<?xml version="1.0" encoding="UTF-8"?>
<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>

    <groupId>com.example</groupId>
    <artifactId>tcp-ber-source</artifactId>
    <version>1.0</version>
    <packaging>jar</packaging>

    <name>TCP Ber Source</name>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.2.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>

        <tcp-app-starters-common.version>1.1.0.RELEASE</tcp-app-starters-common.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud.stream.app</groupId>
            <artifactId>tcp-app-starters-common</artifactId>
            <version>${tcp-app-starters-common.version}</version>
        </dependency>

        <dependency>
            <groupId>com.example</groupId>
            <artifactId>ber-byte-array-serializers</artifactId>
            <version>1.0</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Camden.SR3</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

Configuration:

@EnableBinding(Source.class)
@EnableConfigurationProperties(TcpBerSourceProperties.class)
public class TcpBerSourceConfiguration {

    private final Source channels;
    private final TcpBerSourceProperties properties;

    @Autowired
    public TcpSourceConfiguration(final TcpBerSourceProperties properties, final Source channels) {
        this.properties = properties;
        this.channels = channels;
    }

    @Bean
    public TcpReceivingChannelAdapter adapter(@Qualifier("tcpBerSourceConnectionFactory") final AbstractConnectionFactory connectionFactory) {
        final TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
        adapter.setConnectionFactory(connectionFactory);
        adapter.setOutputChannel(this.channels.output());
        return adapter;
    }

    @Bean
    public TcpConnectionFactoryFactoryBean tcpBerSourceConnectionFactory(@Qualifier("tcpBerSourceDecoder") final AbstractByteArraySerializer decoder) throws Exception {
        final TcpConnectionFactoryFactoryBean factoryBean = new TcpConnectionFactoryFactoryBean();
        factoryBean.setType("server");
        factoryBean.setPort(this.properties.getPort());
        factoryBean.setUsingNio(this.properties.isNio());
        factoryBean.setUsingDirectBuffers(this.properties.isUseDirectBuffers());
        factoryBean.setLookupHost(this.properties.isReverseLookup());
        factoryBean.setDeserializer(decoder);
        factoryBean.setSoTimeout(this.properties.getSocketTimeout());
        return factoryBean;
    }

    @Bean
    public BerEncoderDecoderFactoryBean tcpBerSourceDecoder() {
        final BerEncoderDecoderFactoryBean factoryBean = new BerEncoderDecoderFactoryBean(this.properties.getDecoder());
        factoryBean.setMaxMessageSize(this.properties.getBufferSize());
        return factoryBean;
    }
}

And this FactoryBean:

public class BerEncoderDecoderFactoryBean extends AbstractFactoryBean<AbstractByteArraySerializer> implements ApplicationEventPublisherAware {

    private final BerEncoding encoding;

    private ApplicationEventPublisher applicationEventPublisher;
    private Integer maxMessageSize;

    public BerEncoderDecoderFactoryBean(final BerEncoding encoding) {
        Assert.notNull(encoding, "'encoding' cannot be null");
        this.encoding = encoding;
    }

    @Override
    public void setApplicationEventPublisher(final ApplicationEventPublisher applicationEventPublisher) {
        this.applicationEventPublisher = applicationEventPublisher;
    }

    /**
     * The maximum message size allowed when decoding.
     * @param maxMessageSize the maximum message size.
     */
    public void setMaxMessageSize(final int maxMessageSize) {
        this.maxMessageSize = maxMessageSize;
    }

    @Override
    protected AbstractByteArraySerializer createInstance() throws Exception {
        final AbstractByteArraySerializer codec;
        switch (this.encoding) {
            case SPLIT:
                codec = new ByteArrayBerSplitSerializer();
                break;
            case EXTRACT:
                codec = new ByteArrayBerExtractSerializer();
                break;
            default:
                throw new IllegalArgumentException("Invalid encoding: " + this.encoding);
        }
        codec.setApplicationEventPublisher(this.applicationEventPublisher);
        if (this.maxMessageSize != null) {
            codec.setMaxMessageSize(this.maxMessageSize);
        }
        return codec;
    }

    @Override
    public Class<?> getObjectType() {
        return AbstractByteArraySerializer.class;
    }
}

BerEncoding is a simple enum, and TcpBerSourceProperties are pretty straightforward.

Is this the right approach?

If it is, how can I run this? I can't see @SpringBootApplication anywhere on the tcp stream app starters I found on the mentioned link to run as Spring Boot standalone applications?

Answer 1

You have to create your own Spring Boot App class and import the configuration class; see the documentation about creating custom apps.

We generate a standard app (for rabbit and kafka binders) from the starters as explained here.