Spring Cloud Data Flow Starter Apps自定义代码添加可行性咨询

Absolutely! Spring Cloud Data Flow (SCDF) gives you several flexible ways to inject custom code via hooks and callbacks—even when using out-of-the-box starter apps like the file source. For your specific use case of logging processed file details to a database, here are three practical, battle-tested approaches:

1. Extend the Starter App with Event Listeners #

Starter apps like file are built on Spring Cloud Stream, which exposes internal events you can hook into. For the file source, you can listen for FileReadingMessageSource.FileReadEvent—this event triggers every time a file is successfully read and sent to the stream.

How to implement: #

1. Create a custom Spring Boot app that reuses the file source logic (leveraging Spring Cloud Stream's FileReadingMessageSource), or extend the core logic of the official file starter app.
2. Add a custom event listener component to log file details to your database:

@Component
public class ProcessedFileLogger {

    private final JdbcTemplate jdbcTemplate;

    // Inject your database access bean (JdbcTemplate, Spring Data Repository, etc.)
    public ProcessedFileLogger(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @EventListener
    public void onFileRead(FileReadingMessageSource.FileReadEvent event) {
        File processedFile = event.getFile();
        
        // Insert file details into your database
        String insertSql = """
            INSERT INTO processed_files (filename, file_path, processed_timestamp, file_size)
            VALUES (?, ?, ?, ?)
        """;
        
        jdbcTemplate.update(insertSql,
            processedFile.getName(),
            processedFile.getAbsolutePath(),
            LocalDateTime.now(),
            processedFile.length());
    }
}

3. Package your modified app as a JAR and register it with SCDF (replace the default file source with your custom version).

This approach keeps logging logic tightly coupled with file processing, ensuring every read file gets logged immediately.

2. Use a Custom Stream Processor as a Middleman #

If you don’t want to modify the original starter app, insert a lightweight custom processor into your stream pipeline. This processor receives the file message, logs its details to the database, then passes the message along to the next component.

Example processor code: #

@Component
public class FileLoggingProcessor {

    private final ProcessedFileRepository fileRepository;

    public FileLoggingProcessor(ProcessedFileRepository fileRepository) {
        this.fileRepository = fileRepository;
    }

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public Message<?> logAndForward(Message<File> fileMessage) {
        File file = fileMessage.getPayload();
        
        // Save to database using Spring Data JPA (or your preferred ORM)
        ProcessedFile record = new ProcessedFile();
        record.setFilename(file.getName());
        record.setFilePath(file.getAbsolutePath());
        record.setProcessedAt(LocalDateTime.now());
        record.setFileSize(file.length());
        fileRepository.save(record);
        
        // Forward the original message to keep the stream pipeline intact
        return fileMessage;
    }
}

Pipeline setup in SCDF: #

Register your custom processor with SCDF, then create a stream like:

file --directory=/path/to/files | file-logging-processor | your-target-sink

This is a non-intrusive way to add custom logic without altering starter apps.

3. Listen to SCDF Global Events #

SCDF emits a wide range of system events (like MessageSentEvent) that you can subscribe to from an external application. This method lets you log file details without touching the stream pipeline itself.

How to implement: #

1. Create a standalone Spring Boot app that connects to SCDF’s event bus.
2. Add an event listener to capture messages sent by the file source:

@Component
public class ScdfEventLogger {

    private final JdbcTemplate jdbcTemplate;

    public ScdfEventLogger(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }

    @EventListener
    public void onMessageSent(MessageSentEvent event) {
        // Extract the file payload from the event
        if (event.getMessage().getPayload() instanceof File file) {
            // Log to database (same logic as previous examples)
            String insertSql = """
                INSERT INTO processed_files (filename, file_path, processed_timestamp)
                VALUES (?, ?, ?)
            """;
            jdbcTemplate.update(insertSql,
                file.getName(),
                file.getAbsolutePath(),
                LocalDateTime.now());
        }
    }
}

3. Configure the app to listen to SCDF’s event stream (using Spring Cloud Bus or SCDF’s built-in event subscription mechanisms).

This is ideal if you want to centralize logging across multiple streams or avoid modifying pipeline components.

Which approach should you choose? #

• Extend the starter app: Best for tight coupling between file reading and logging.
• Custom processor: Most flexible and non-intrusive, great for adding reusable logic to multiple pipelines.
• Global event listener: Perfect for centralized monitoring across your entire SCDF environment.

内容的提问来源于stack exchange，提问作者Sreenivasulu Guduru