package org.springframework.messaging;
public interface Message<T> {
    T getPayload();
    MessageHeaders getHeaders();
}

@FunctionalInterface
public interface MessageChannel {
	long INDEFINITE_TIMEOUT = -1;
	default boolean send(Message<?> message) {
		return send(message, INDEFINITE_TIMEOUT);
	}
	boolean send(Message<?> message, long timeout);
}

public interface SubscribableChannel extends MessageChannel {
	boolean subscribe(MessageHandler handler);
	boolean unsubscribe(MessageHandler handler);
}

@FunctionalInterface
public interface MessageHandler {
    void handleMessage(Message<?> message) throws MessagingException;
}

        <!--stream-->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>

spring:
    cloud:
        stream:
            # 定義消息中間件
            binders:
              MyRabbit:
                  type: rabbit
                  environment:
                    spring:
                        rabbitmq:
                            host: localhost
                            port: 5672
                            username: root
                            password: root
                            vhost: /
            bindings:
            # 生產(chǎn)者中定義，定義發(fā)布對(duì)象
              myInput:
                destination: myStreamExchange
                group: myStreamGroup
                binder: MyRabbit
            # 消費(fèi)者中定義，定義訂閱的對(duì)象
              myOutput-in-0:
                destination: myStreamExchange
                group: myStreamGroup
                binder: MyRabbit
        # 消費(fèi)者中定義，定義輸出的函數(shù)
        function:
            definition: myOutput

@Resource
	private StreamBridge streamBridge;
	public void sendNormal() {
		streamBridge.send("myInput", "hello world");
	}

@Bean("myOutput")
	public Consumer<Message<String>> myOutput() {
		return (message) -> {
			MessageHeaders headers = message.getHeaders();
			System.out.println("myOutput head is : " + headers);
			String payload = message.getPayload();
			System.out.println("myOutput payload is : " + payload);
		};
	}

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
    <version>${spring.cloud.stream.version}</version>
</dependency>

public class FileProvisioningProvider implements ProvisioningProvider<
	ExtendedConsumerProperties<FileConsumerProperties>, ExtendedProducerProperties<FileProducerProperties>> {
	public FileProvisioningProvider() {
		super();
	}
	@Override
	public ProducerDestination provisionProducerDestination(String name, ExtendedProducerProperties<FileProducerProperties> properties) throws ProvisioningException {
		return new FileMessageDestination(name);
	}
	@Override
	public ConsumerDestination provisionConsumerDestination(String name, String group, ExtendedConsumerProperties<FileConsumerProperties> properties) throws ProvisioningException {
		return new FileMessageDestination(name);
	}
	private static class FileMessageDestination implements ProducerDestination, ConsumerDestination {
		private final String destination;
		private FileMessageDestination(final String destination) {
			this.destination = destination;
		}
		@Override
		public String getName() {
			return destination.trim();
		}
		@Override
		public String getNameForPartition(int partition) {
			throw new UnsupportedOperationException("Partitioning is not implemented for file messaging.");
		}
	}
}

super.onInit();
		executorService = Executors.newScheduledThreadPool(1);
	}
	@Override
	public void doStart() {
		executorService.scheduleWithFixedDelay(() -> {
			String payload = getPayload();
			if (payload != null) {
				Message<String> receivedMessage = MessageBuilder.withPayload(payload).build();
				sendMessage(receivedMessage);
			}
		}, 0, 50, TimeUnit.MILLISECONDS);
	}
	@Override
	protected void doStop() {
		executorService.shutdownNow();
	}
	private String getPayload() {
		try {
			List<String> allLines = Files.readAllLines(Paths.get(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
			String currentPayload = allLines.get(allLines.size() - 1);
			if (!currentPayload.equals(previousPayload)) {
				previousPayload = currentPayload;
				return currentPayload;
			}
		} catch (IOException e) {
			FileUtil.touch(new File(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"));
		}
		return null;
	}
}

public class FileMessageHandler extends AbstractMessageHandler {
	FileExtendedBindingProperties fileExtendedBindingProperties;
	ProducerDestination destination;
	public FileMessageHandler(ProducerDestination destination, FileExtendedBindingProperties fileExtendedBindingProperties) {
		this.destination = destination;
		this.fileExtendedBindingProperties = fileExtendedBindingProperties;
	}
	@Override
	protected void handleMessageInternal(Message<?> message) {
		try {
			if (message.getPayload() instanceof byte[]) {
				Files.write(Paths.get(fileExtendedBindingProperties.getPath() + File.separator + destination.getName() + ".txt"), (byte[]) message.getPayload());
			} else {
				throw new RuntimeException("處理消息失敗");
			}
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}
}

public class FileMessageChannelBinder extends AbstractMessageChannelBinder
	<ExtendedConsumerProperties<FileConsumerProperties>, ExtendedProducerProperties<FileProducerProperties>, FileProvisioningProvider>
	implements ExtendedPropertiesBinder<MessageChannel, FileConsumerProperties, FileProducerProperties> {
	FileExtendedBindingProperties fileExtendedBindingProperties;
	public FileMessageChannelBinder(String[] headersToEmbed, FileProvisioningProvider provisioningProvider, FileExtendedBindingProperties fileExtendedBindingProperties) {
		super(headersToEmbed, provisioningProvider);
		this.fileExtendedBindingProperties = fileExtendedBindingProperties;
	}
	@Override
	protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties<FileProducerProperties> producerProperties, MessageChannel errorChannel) throws Exception {
		FileMessageHandler fileMessageHandler = new FileMessageHandler(destination, fileExtendedBindingProperties);
		return fileMessageHandler;
	}
	@Override
	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties<FileConsumerProperties> properties) throws Exception {
		FileMessageProducerAdapter fileMessageProducerAdapter = new FileMessageProducerAdapter(destination, fileExtendedBindingProperties);
		return fileMessageProducerAdapter;
	}
	@Override
	public FileConsumerProperties getExtendedConsumerProperties(String channelName) {
		return fileExtendedBindingProperties.getExtendedConsumerProperties(channelName);
	}
	@Override
	public FileProducerProperties getExtendedProducerProperties(String channelName) {
		return fileExtendedBindingProperties.getExtendedProducerProperties(channelName);
	}
	@Override
	public String getDefaultsPrefix() {
		return fileExtendedBindingProperties.getDefaultsPrefix();
	}
	@Override
	public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
		return fileExtendedBindingProperties.getExtendedPropertiesEntryClass();
	}
}

@EnableConfigurationProperties(FileExtendedBindingProperties.class)
@Configuration
public class FileMessageBinderConfiguration {
	@Bean
	@ConditionalOnMissingBean
	public FileProvisioningProvider fileMessageBinderProvisioner() {
		return new FileProvisioningProvider();
	}
	@Bean
	@ConditionalOnMissingBean
	public FileMessageChannelBinder fileMessageBinder(FileProvisioningProvider fileMessageBinderProvisioner, FileExtendedBindingProperties fileExtendedBindingProperties) {
		return new FileMessageChannelBinder(null, fileMessageBinderProvisioner, fileExtendedBindingProperties);
	}
	@Bean
	public FileProducerProperties fileConsumerProperties() {
		return new FileProducerProperties();
	}
}