import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
public IntegrationFlow amqpNewOrderEventInboundFlow() {
    return IntegrationFlows.from(Amqp.inboundAdapter(rabbitMQConnection(), queueNewRealizations))
            .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
            .transform(this::performNewOrderEventUnmarshaling)
            .publishSubscribeChannel(c -> c
                    .subscribe(s -> s
                            .<NewOrderEvent, Boolean>route(NewOrderEvent::isPriorityRealization, mapping -> mapping
                                    .subFlowMapping(false, sf -> sf
                                            .channel(IntegrationFlowKeys.NOT_PRIORITY_ORDER_PROCESSING_QUEUE))
                                    .subFlowMapping(true, sf -> sf
                                            .channel(IntegrationFlowKeys.PRIORITY_ORDER_PROCESSING_QUEUE))))
                    .subscribe(s -> s
                            .transform(orderReadyMapper::toOrderReady)
                            .channel(IntegrationFlowKeys.NEW_ORDER_OUT_CHANNEL)))
            .get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
private IntegrationFlowBuilder newOrderEventProcessingFlow(IntegrationFlowBuilder builder, boolean priorityQueue) {
    return builder
            .split(NewOrderEvent.class, NewOrderEvent::getItems)
            .transform(orderItemMapper::toOrderItemDs)
            .channel(MessageChannels.queue(10))
            .publishSubscribeChannel(c2 -> c2
                    .subscribe(s2 -> s2
                            .handle(m -> orderPreparationService.prepare((OrderItemDs) m.getPayload(), false)))
                    .subscribe(s2 -> s2
                            .aggregate(aggregator -> aggregator
                                    .outputProcessor(group -> group
                                            .getMessages()
                                            .stream()
                                            .map(m -> ((OrderItemDs) m.getPayload()).getItemName())
                                            .collect(Collectors.joining(","))))
                            .handle(m -> System.out.println(m.getPayload()))));
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
public IntegrationFlow amqpNewOrderEventInboundFlow() {
    return IntegrationFlows.from(Amqp.inboundAdapter(rabbitMQConnection(), queueOrderReady))
            .channel(MessageChannels.executor(Executors.newCachedThreadPool()))
            .transform(this::performOrderReadyUnmarshaling)
            .handle(m -> sentNotifiation((OrderReady) m.getPayload()))
            .get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
public IntegrationFlow flow() {
	String to = env.getProperty("queues.portfolio.creation");
	return IntegrationFlows
		.from(MessageChannels.publishSubscribe(Channels.PortfolioInitializationRequest))
		.transform(Transformers.toJson(mapper()))
		.handle(Amqp
			.outboundGateway(amqpTemplate)
			.routingKey(to))
		.transform(Transformers.fromJson(ContractPortfolioRelation.class, mapper()))
		.channel(MessageChannels.direct(Channels.PortfolioInitializationResponse))
		.get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
IntegrationFlow creationflow() {
	String to = env.getProperty("queues.contract.creation");
	return IntegrationFlows
		.from(MessageChannels.publishSubscribe(Channels.ContractCreationRequest))
		.transform(Transformers.toJson(mapper()))
		.handle(Amqp
			.outboundGateway(amqpTemplate)
			.routingKey(to)
		)
		.transform(Transformers.fromJson(Contract.class, mapper()))
		.channel(MessageChannels.direct(Channels.ContractCreationResponse))
		.get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
IntegrationFlow approbationflow() {
	String to = env.getProperty("queues.contract.approbation");
	return IntegrationFlows
		.from(MessageChannels.publishSubscribe(Channels.ContractApprobationRequest))
		.transform(Transformers.toJson(mapper()))
		.handle(Amqp
			.outboundGateway(amqpTemplate)
			.routingKey(to)
		)
		.transform(Transformers.fromJson(Contract.class, mapper()))
		.channel(MessageChannels.direct(Channels.ContractApprobationResponse))
		.get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
IntegrationFlow initialPaymentflow() {
	String to = env.getProperty("queues.payment.initial-payment-reception");
	return IntegrationFlows
		.from(MessageChannels.publishSubscribe(Channels.InitialPaymentReceptionRequest))
		.transform(Transformers.toJson(mapper()))
		.handle(Amqp
			.outboundGateway(amqpTemplate)
			.routingKey(to)
		)
		.transform(Transformers.fromJson(Order.class, mapper()))
		.channel(MessageChannels.direct(Channels.InitialPaymentReceptionResponse))
		.get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
	public IntegrationFlow myFlow() {
		return IntegrationFlows.from(this.integerMessageSource(), c -> c.poller(Pollers.fixedRate(100)))
				.channel(this.inputChannel())
				.filter((Integer p) -> p > 0)
				.transform(Object::toString)
				.channel(MessageChannels.queue("sampleQueue"))
//				.handle(System.out::println)
				.get();
	}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
public IntegrationFlow myFlow() {
	return IntegrationFlows.from(this.integerMessageSource(), c ->
			c.poller(Pollers.fixedRate(100)))
			.channel(this.inputChannel())
			.filter((Integer p) -> p > 0)
			.transform(Object::toString)
			.channel(MessageChannels.queue())
			.get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean(name= IntegrationFlowKeys.NEW_ORDER_OUT_CHANNEL)
 public MessageChannel newOrderOutChannel(){
     return MessageChannels.executor(Executors.newCachedThreadPool()).get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean(name= IntegrationFlowKeys.NEW_ORDER_OUT_CHANNEL)
public MessageChannel newOrderOutChannel(){
    return MessageChannels.executor(Executors.newCachedThreadPool()).get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
public ExecutorChannel outboundRequests() {
	return MessageChannels.executor("outboundRequests", new SimpleAsyncTaskExecutor()).get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel files() {
    return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel contractCreationErrorChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel orderInitializationChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel initialPaymentReceptionErrorChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel portfolioInitializationChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel createContractDocumentationChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel portfolioOrderErrorChannel() {
	return MessageChannels.direct().get();
}
 

import org.springframework.integration.dsl.channel.MessageChannels; //导入依赖的package包/类
@Bean
MessageChannel portfolioInitializationErrorChannel() {
	return MessageChannels.direct().get();
}