Reloadable Tail SCSt Source App

TailSourceConfiguration.java

/*
 * Copyright 2015-2016 the original author or authors.
 *
 * Licensed 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.springframework.cloud.stream.app.tail.source;

import java.io.File;
import java.util.concurrent.atomic.AtomicBoolean;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.dsl.file.Files;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.filters.CompositeFileListFilter;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.file.tail.FileTailingMessageProducerSupport;


@EnableBinding(Source.class)
@EnableConfigurationProperties(TailSourceProperties.class)
public class TailSourceConfiguration {

	@Autowired
	private TailSourceProperties properties;

	@Bean
	public IntegrationFlow tailFilesFlow() throws Exception {
		return IntegrationFlows
				// Fake file for the adapter initialization.
				// The real one is populated by the fileReadingMessageSource
				.from(Files.tailAdapter(new File(System.getProperty("java.io.tmpdir")))
						.delay(500)
						.end(false)
						.id("tailer")
						.autoStartup(false))
				.channel(Source.OUTPUT)
				.get();
	}

	private MessageSource<File> fileReadingMessageSource() {
		CompositeFileListFilter<File> filters = new CompositeFileListFilter<>();
		filters.addFilter(new SimplePatternFileListFilter(this.properties.getFilePattern()));

		FileReadingMessageSource fileSource = new FileReadingMessageSource() {

			private final AtomicBoolean running = new AtomicBoolean();

			@Override
			public void start() {
				if (!this.running.getAndSet(true)) {
					super.start();
				}
			}

			@Override
			public void stop() {
				if (this.running.getAndSet(false)) {
					super.stop();
				}
			}

		};
		fileSource.setDirectory(this.properties.getSourceDirectory());
		fileSource.setFilter(filters);
		fileSource.setUseWatchService(true);
		fileSource.setWatchEvents(FileReadingMessageSource.WatchEventType.CREATE,
				FileReadingMessageSource.WatchEventType.MODIFY);
		return fileSource;
	}

	@Bean
	public IntegrationFlow readDirectoryFlow(FileTailingMessageProducerSupport tailer) {
		return IntegrationFlows
				.from(fileReadingMessageSource(),
						e -> e.poller(Pollers.cron("*/1 * * * * *")))
				.<File>handle((payload, h) -> {
					tailer.stop();
					tailer.setFile(payload);
					tailer.start();
					return null;
				})
				.get();
	}

}

TailSourceProperties.java

/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed 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.springframework.cloud.stream.app.tail.source;

import java.io.File;

import org.springframework.boot.context.properties.ConfigurationProperties;

@ConfigurationProperties("tail")
public class TailSourceProperties {

	private File sourceDirectory;

	private String filePattern;

	public void setSourceDirectory(File sourceDirectory) {
		this.sourceDirectory = sourceDirectory;
	}

	public File getSourceDirectory() {
		return sourceDirectory;
	}

	public String getFilePattern() {
		return filePattern;
	}

	public void setFilePattern(String filePattern) {
		this.filePattern = filePattern;
	}

}

TailSourceTests.java

/*
 * Copyright 2015-2016 the original author or authors.
 *
 * Licensed 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.springframework.cloud.stream.app.tail.source;

import java.io.File;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.IntegrationTest;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.Assert;
import org.springframework.util.FileCopyUtils;

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = TailSourceTests.TailSourceApplication.class)
@DirtiesContext
@IntegrationTest("tail.filePattern=QUOTE_APPROVAL_DETAILS_CAT*")
public class TailSourceTests {

	@ClassRule
	public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();

	@Autowired
	protected Source source;

	@Autowired
	protected MessageCollector messageCollector;

	@BeforeClass
	public static void setup() throws IOException {
		File remoteFolder = TEMPORARY_FOLDER.newFolder();
		File aFile = new File(remoteFolder, "QUOTE_APPROVAL_DETAILS_CAT1.test");
		FileCopyUtils.copy("Hello\n\r".getBytes(), aFile);

		System.setProperty("tail.sourceDirectory", TEMPORARY_FOLDER.getRoot().getAbsolutePath());
	}

	@Test
	public void testTailReload() throws Exception {
		Message<?> received = this.messageCollector.forChannel(this.source.output()).poll(10, TimeUnit.SECONDS);
		Assert.notNull(received);
		System.out.println(received.getPayload());

		File remoteFolder = TEMPORARY_FOLDER.newFolder();
		File aFile = new File(remoteFolder, "QUOTE_APPROVAL_DETAILS_CAT2.test");
		FileCopyUtils.copy("Hello again!\n\r".getBytes(), aFile);

		received = this.messageCollector.forChannel(this.source.output()).poll(60, TimeUnit.SECONDS);
		Assert.notNull(received);
		System.out.println(received.getPayload());

	}

	@SpringBootApplication
	public static class TailSourceApplication {

	}

}