package my.springframework.messaging;

/**
 * A generic message representation with headers and body.
 */
public interface Message<T> {

    /**
     * Return the message payload.
     */
    T getPayload();

    /**
     * Return message headers for the message.
     */
    MessageHeaders getHeaders();

}

package my.springframework.messaging;

import java.io.Serializable;
import java.util.*;

import org.springframework.util.AlternativeJdkIdGenerator;
import org.springframework.util.IdGenerator;

public class MessageHeaders implements Map<String, Object>, Serializable {

    /**
     * The key for the Message ID. This is an automatically generated UUID and
     * should never be explicitly set in the header map.
     */
    public static final String ID = "id";

    public static final String REPLY_CHANNEL = "replyChannel";

    private static final IdGenerator defaultIdGenerator =
        new AlternativeJdkIdGenerator();

    private final Map<String, Object> headers;

    public MessageHeaders(Map<String, Object> headers) {
        this.headers = (headers != null ?
            new HashMap<String, Object>(headers) :
            new HashMap<String, Object>());
        this.headers.put(ID, defaultIdGenerator.generateId());
    }

    public UUID getId() {
        return get(ID, UUID.class);
    }

    public Object getReplyChannel() {
        return get(REPLY_CHANNEL);
    }

    @SuppressWarnings("unchecked")
    public <T> T get(Object key, Class<T> type) {
        Object value = this.headers.get(key);
        if (value == null) {
            return null;
        }
        if (!type.isAssignableFrom(value.getClass())) {
            throw new IllegalArgumentException("Incorrect type");
        }
        return (T) value;
    }


    // Delegating Map implementation

    public boolean containsKey(Object key) {
        return this.headers.containsKey(key);
    }

    // + Same for containsValue, entrySet, get, isEmpty, keySet, size, values

    // Unsupported Map operations

    /**
     * Since MessageHeaders is immutable, the call to this method
     * will result in {@link UnsupportedOperationException}.
     */
    public Object put(String key, Object value) {
        throw new UnsupportedOperationException("MessageHeaders is immutable");
    }

    // + Same for putAll, remove, and clear operations

}

package my.springframework.messaging.support;

import java.io.Serializable;
import java.util.Map;

import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;

import my.springframework.messaging.Message;
import my.springframework.messaging.MessageHeaders;

public class GenericMessage<T> implements Message<T>, Serializable {

    private final T payload;

    private final MessageHeaders headers;

    public GenericMessage(T payload) {
        this(payload, new MessageHeaders(null));
    }

    public GenericMessage(T payload, Map<String, Object> headers) {
        this(payload, new MessageHeaders(headers));
    }

    public GenericMessage(T payload, MessageHeaders headers) {
        Assert.notNull(payload, "Payload must not be null");
        Assert.notNull(headers, "MessageHeaders must not be null");
        this.payload = payload;
        this.headers = headers;
    }

    public T getPayload() {
        return this.payload;
    }

    public MessageHeaders getHeaders() {
        return this.headers;
    }

    // + equals, hashcode and equals

}

package my.springframework.integration.support;

import java.util.*;
import org.springframework.util.*;
import my.springframework.messaging.*;

public final class MessageBuilder<T> {

    private final T payload;
    private final Map<String, Object> headers; // MessageHeaders is immutable,
                                               // we should create a Map
    private final Message<T> originalMessage;

    /**
     * Private constructor to be invoked from the static factory methods only.
     */
    private MessageBuilder(T payload, Message<T> originalMessage) {
        Assert.notNull(payload, "payload must not be null");
        this.payload = payload;
        this.originalMessage = originalMessage;
        this.headers = new HashMap<>();
        if (originalMessage != null) {
            this.headers.putAll(originalMessage.getHeaders());
        }
    }

    public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
        Assert.notNull(message, "message must not be null");
        MessageBuilder<T> builder = new MessageBuilder<T>(
                message.getPayload(), message);
        return builder;
    }

    public static <T> MessageBuilder<T> withPayload(T payload) {
        MessageBuilder<T> builder = new MessageBuilder<T>(payload, null);
        return builder;
    }

    private Object getHeader(String headerName) {
        return headers.get(headerName);
    }

    public MessageBuilder<T> setHeader(String name, Object value) {
        if (!ObjectUtils.nullSafeEquals(value, getHeader(name))) {
            if (value != null) {
                headers.put(name, value);
            }
            else {
                headers.remove(name);
            }
        }
        return this;
    }

    public MessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy) {
        if (headersToCopy != null) {
            for (Map.Entry<String, ?> entry : headersToCopy.entrySet()) {
                if (!isReadOnly(entry.getKey())) {
                    setHeader(entry.getKey(), entry.getValue());
                }
            }
        }
        return this;
    }

    public MessageBuilder<T> setReplyChannel(MessageChannel replyChannel) {
        return setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
    }

    protected boolean isReadOnly(String headerName) {
        return MessageHeaders.ID.equals(headerName);
    }

    public Message<T> build() {
        return new GenericMessage<T>(
            payload,
            new HashMap<String, Object>(headers));
    }

}

Message<String> hello = MessageBuilder.withPayload("hello").build();

// Proxy Gateway create a temporary channel to send the response
Message<String> proxy = MessageBuilder.withPayload("proxyGateway").
                            setReplyChannel(aTemporaryChannel).
                            build();

// Component could add a new header by duplicating a message
Message<String> helloWorld = MessageBuilder.fromMessage(hello).
                                 setHeader("name", "World").
                                 build();

package my.springframework.messaging;

public interface MessageChannel {

    /**
     * Send a {@link Message} to this channel. If the message is sent,
     * the method returns {@code true}. If the message cannot be sent due to a
     * non-fatal reason, the method returns {@code false}.
     * To provide a maximum wait time, use {@link #send(Message, long)}.
     */
    boolean send(Message<?> message);

    /**
     * Send a message, blocking until either the message is accepted or the
     * specified timeout period elapses.
     */
    boolean send(Message<?> message, long timeout);

}

package my.springframework.messaging;

public interface PollableChannel extends MessageChannel {

    /**
     * Receive a message from this channel, blocking indefinitely if necessary.
     */
    Message<?> receive();

    /**
     * Receive a message from this channel, blocking until
     * either a message is available
     * or the specified timeout period elapses.
     */
    Message<?> receive(long timeout);

}

package my.springframework.messaging;

public interface SubscribableChannel extends MessageChannel {

    boolean subscribe(MessageHandler handler);

    boolean unsubscribe(MessageHandler handler);

}

package my.springframework.messaging;

public interface MessageHandler {

    void handleMessage(Message<?> message);

}

package my.springframework.integration.channel;

import java.util.*;
import my.springframework.messaging.*;

public class DirectChannel implements SubscribableChannel {

    private List<MessageHander> handlers = new ArrayList<>();

    public boolean subscribe(MessageHandler handler) {
        this.handlers.add(handler);
    }

    @Override
    public boolean send(Message<?> message) {
        handlers.iterator().next().send(message);
        return true;
    }

}

package my.springframework.integration.channel;

import java.util.concurrent.CopyOnWriteArrayList;

import my.springframework.messaging.MessageHandler;
import my.springframework.messaging.SubscribableChannel;

public abstract class AbstractSubscribableChannel
        implements SubscribableChannel {

    protected final CopyOnWriteArrayList<MessageHandler> handlers =
        new CopyOnWriteArrayList<MessageHandler>();

    @Override
    public boolean subscribe(MessageHandler handler) {
        handlers.add(handler);
        return true;
    }

    @Override
    public boolean unsubscribe(MessageHandler handler) {
        handlers.remove(handler);
        return true;
    }

}

package my.springframework.integration.channel;

import java.util.*;
import my.springframework.messaging.*;

public class DirectChannel extends AbstractSubscribableChannel {

    @Override
    public boolean send(Message<?> message) {
        return send(message, -1);
    }

    @Override
    public boolean send(Message<?> message, long timeout) {
        boolean success = false;
        Iterator<MessageHandler> handlerIterator = handlers.iterator();
        List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
        while (!success && handlerIterator.hasNext()) {
            MessageHandler handler = handlerIterator.next();
            try {
                handler.handleMessage(message);
                success = true; // we have a winner.
            }
            catch (Exception e) {
                RuntimeException runtimeException =
                    wrapExceptionIfNecessary(message, e);
                exceptions.add(runtimeException);
                if (!handlerIterator.hasNext()) {
                    throw new MessagingException(message,
                        "All attempts to deliver Message failed.");
                }
            }
        }
        return success;
    }

    private RuntimeException wrapExceptionIfNecessary(
            Message<?> message, Exception e) {
        RuntimeException runtimeException = (e instanceof RuntimeException)
                ? (RuntimeException) e
                : new MessagingException(message, "Dispatcher failed.", e);
        return runtimeException;
    }

}

package my.springframework.messaging;

/**
 * The base exception for any failures related to messaging.
 */
@SuppressWarnings("serial")
public class MessagingException extends RuntimeException {

    private final Message<?> failedMessage;

    public MessagingException(Message<?> message, String description) {
        super(description);
        this.failedMessage = message;
    }

    public MessagingException(Message<?> message, String description,
                              Throwable cause) {
        super(description, cause);
        this.failedMessage = message;
    }

    public MessagingException(String description, Throwable cause) {
        this(null, description, cause);
    }

    public Message<?> getFailedMessage() {
        return this.failedMessage;
    }

}

package my.springframework.integration.channel;

import my.springframework.messaging.*;

public class PublishSubscribeChannel extends AbstractSubscribableChannel {

    @Override
    public boolean send(Message<?> message) {
        return send(message, -1);
    }

    @Override
    public boolean send(Message<?> message, long timeout) {
        for (MessageHandler handler : handlers) {
            handler.handleMessage(message);
        }
        return true;
    }

}

package my.springframework.integration.channel;

import java.util.concurrent.Executor;

import my.springframework.messaging.Message;
import my.springframework.messaging.MessageHandler;

public class PublishSubscribeChannel extends AbstractSubscribableChannel {

    private static Executor CALLER_RUNS = runnable -> runnable.run();

    private Executor executor;

    public PublishSubscribeChannel() {
        this(CALLER_RUNS);
    }

    public PublishSubscribeChannel(Executor executor) {
        this.executor = executor;
    }

    @Override
    public boolean send(Message<?> message) {
        return send(message, -1);
    }

    @Override
    public boolean send(Message<?> message, long timeout) {
        for (MessageHandler handler : handlers) {
            executor.execute(() -> handler.handleMessage(message));
        }
        return true;
    }

}

private static Executor CALLER_RUNS = runnable -> runnable.run();

public class CallerRunsExecutor implements java.util.concurrent.Executor {

    @Override
    public void execute(Runnable runnable) {
        runnable.run();
    }
}

package my.springframework.integration.channel;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.springframework.util.Assert;

import my.springframework.messaging.Message;
import my.springframework.messaging.MessagingException;
import my.springframework.messaging.PollableChannel;

public class QueueChannel implements PollableChannel {

    private final BlockingQueue<Message<?>> queue =
           new LinkedBlockingQueue<Message<?>>();

    @Override
    public final Message<?> receive() {
        return receive(-1);
    }

    /**
     * Receive the first available message from this channel. If the channel
     * contains no messages, this method will block until the allotted timeout
     * elapses. If the specified timeout is 0, the method will return
     * immediately. If less than zero, it will block indefinitely (see
     * {@link #receive()}).
     */
    @Override
    public final Message<?> receive(long timeout) {
        try {
            if (timeout > 0) {
                return this.queue.poll(timeout, TimeUnit.MILLISECONDS);
            }
            if (timeout == 0) {
                return this.queue.poll();
            }

            return this.queue.take();
        }
        catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return null;
        }
    }


    @Override
    public final boolean send(Message<?> message) {
        return this.send(message, -1);
    }

    /**
     * Send a message on this channel. If the channel is at capacity, this
     * method will block until either the timeout occurs or the sending thread
     * is interrupted. If the specified timeout is 0, the method will return
     * immediately. If less than zero, it will block indefinitely (see
     * {@link #send(Message)}).
     */
    @Override
    public final boolean send(Message<?> message, long timeout) {
        Assert.notNull(message, "message must not be null");
        Assert.notNull(message.getPayload(), "payload must not be null");

        try {
            try {
                if (timeout > 0) {
                    return this.queue.offer(message, timeout,
                              TimeUnit.MILLISECONDS);
                }
                if (timeout == 0) {
                    return this.queue.offer(message);
                }
                this.queue.put(message);
                return true;
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }
        catch (Exception e) {
            throw new MessagingException(message, "Failed", e);
        }
    }

}

// A message is immutable. We should use MessageBuilder to create it:
Message<String> message = MessageBuilder.withPayload("Hello World!").build();

// The message consumption differs whether the channel type.

// Example using a SubscribableChannel:
SubscribableChannel pubSubChannel = new DirectChannel();
pubSubChannel.subscribe(m -> System.out.println("New message received: " + m));
pubSubChannel.send(message);

// Example using a PollableChannel:
PollableChannel pollableChannel = new QueueChannel();
pollableChannel.send(message);
Message<String> m = (Message<String>) pollableChannel.receive();
System.out.println("New message received: " + m);

<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://my.springframework.org/schema/integration"
    xmlns:xsd="http://www.w3.org/2001/XMLSchema"
    targetNamespace="http://my.springframework.org/schema/integration"
    elementFormDefault="qualified" attributeFormDefault="unqualified">

    <xsd:element name="channel">
        <xsd:complexType>
            <xsd:complexContent>
                <xsd:extension base="channelType">
                    <xsd:sequence>
                        <xsd:choice minOccurs="0" maxOccurs="1">
                            <xsd:element name="queue" type="queueType" />
                        </xsd:choice>
                    </xsd:sequence>
                </xsd:extension>
            </xsd:complexContent>
        </xsd:complexType>
    </xsd:element>

    <xsd:element name="bridge">
        <xsd:complexType>
            <xsd:attribute name="id" type="xsd:string" />
            <xsd:attributeGroup ref="inputOutputChannelGroup" />
        </xsd:complexType>
    </xsd:element>

    <xsd:complexType name="queueType">
        <xsd:attribute name="capacity" type="xsd:string" />
    </xsd:complexType>

    <xsd:element name="publish-subscribe-channel">
        <xsd:complexType>
            <xsd:complexContent>
                <xsd:extension base="channelType">
                    <xsd:attribute name="task-executor" type="xsd:string" />
                </xsd:extension>
            </xsd:complexContent>
        </xsd:complexType>
    </xsd:element>

    <xsd:complexType name="channelType">
        <xsd:attribute name="id" type="xsd:string" use="required" />
    </xsd:complexType>

    <xsd:element name="gateway">
        <xsd:complexType>
            <xsd:sequence minOccurs="0" maxOccurs="1">
                <xsd:element name="method" minOccurs="0"
                             maxOccurs="unbounded" />
            </xsd:sequence>
            <xsd:attribute name="id" type="xsd:string" use="optional" />
            <xsd:attribute name="service-interface" type="xsd:string"
                use="optional" />
            <xsd:attribute name="default-request-channel" type="xsd:string" />
            <xsd:attribute name="default-reply-channel" type="xsd:string" />
        </xsd:complexType>
    </xsd:element>

    <xsd:element name="service-activator">
        <xsd:complexType>
            <xsd:attribute name="id" type="xsd:string" />
            <xsd:attribute name="ref" type="xsd:string" />
            <xsd:attribute name="method" type="xsd:string" />
            <xsd:attributeGroup ref="inputOutputChannelGroup" />
        </xsd:complexType>
    </xsd:element>

    <xsd:attributeGroup name="inputOutputChannelGroup">
        <xsd:attribute name="output-channel" type="xsd:string" />
        <xsd:attribute name="input-channel" type="xsd:string" />
    </xsd:attributeGroup>
</xsd:schema>

package my.springframework.integration.config.xml;

import org.springframework.beans.factory.xml.NamespaceHandlerSupport;

public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {

    public void init() {
        registerBeanDefinitionParser("gateway",
            new GatewayParser());
        registerBeanDefinitionParser("channel",
            new PointToPointChannelParser());
        registerBeanDefinitionParser("publish-subscribe-channel",
            new PublishSubscribeChannelParser());
        registerBeanDefinitionParser("bridge",
            new BridgeParser());
        registerBeanDefinitionParser("service-activator",
            new ServiceActivatorParser());
    }

}

package my.springframework.integration.config.xml;

import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;

import my.springframework.integration.channel.DirectChannel;
import my.springframework.integration.channel.QueueChannel;

public class PointToPointChannelParser extends AbstractBeanDefinitionParser {


    @Override
    protected AbstractBeanDefinition parseInternal(
            Element element, ParserContext parserContext) {
        BeanDefinitionBuilder builder = null;

        // configure a queue-based channel if any queue sub-element is defined
        if ((DomUtils.getChildElementByTagName(element, "queue")) != null) {
            builder = BeanDefinitionBuilder.genericBeanDefinition(
                          QueueChannel.class);
        } else {
            builder = BeanDefinitionBuilder.genericBeanDefinition(
                          DirectChannel.class);
        }

        AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
        beanDefinition.setSource(parserContext.extractSource(element));
        return beanDefinition;
    }

}

...
import static m.s.i.config.xml.IntegrationNamespaceUtils.*;

public class ServiceActivatorParser extends AbstractBeanDefinitionParser {

    @Override
    protected AbstractBeanDefinition parseInternal(Element element,
            ParserContext parserContext) {
        BeanDefinitionBuilder handlerBuilder = BeanDefinitionBuilder.
            genericBeanDefinition(ServiceActivatingHandler.class);
        setReferenceIfAttributeDefined(
            handlerBuilder, element, "ref");
        setReferenceIfAttributeDefined(
            handlerBuilder, element, "output-channel");
        AbstractBeanDefinition handlerBeanDefinition =
            handlerBuilder.getBeanDefinition();
        String handlerBeanName = BeanDefinitionReaderUtils.generateBeanName(
            handlerBeanDefinition, parserContext.getRegistry());
        parserContext.registerBeanComponent(
            new BeanComponentDefinition(
                handlerBeanDefinition, handlerBeanName));

        BeanDefinitionBuilder builder = BeanDefinitionBuilder.
            genericBeanDefinition(ConsumerEndpointFactoryBean.class);
        builder.addPropertyReference("handler", handlerBeanName);
        String inputChannelName = element.getAttribute("input-channel");
        builder.addPropertyValue("inputChannelName", inputChannelName);
        AbstractBeanDefinition beanDefinition = builder.getBeanDefinition();
        String beanName = this.resolveId(
            element, beanDefinition, parserContext);
        parserContext.registerBeanComponent(
            new BeanComponentDefinition(beanDefinition, beanName));

        return null;
    }

}

public class ConsumerEndpointFactoryBean
        implements FactoryBean<AbstractEndpoint>,
                   BeanFactoryAware,
                   InitializingBean {

    private volatile MessageHandler handler;

    private volatile String inputChannelName;

    private volatile AbstractEndpoint endpoint;

    private volatile ChannelResolver channelResolver;


    public void setHandler(MessageHandler handler) {
        Assert.notNull(handler, "handler must not be null");
        this.handler = handler;
    }

    public void setInputChannelName(String inputChannelName) {
        this.inputChannelName = inputChannelName;
    }

    /*
     * BeanFactoryAware implementation
     */

    @Override
    public void setBeanFactory(BeanFactory beanFactory) {
        this.channelResolver = new ChannelResolver(beanFactory);
    }

    /*
     * InitializingBean implementation
     */

    @Override
    public void afterPropertiesSet() throws Exception {
        MessageChannel channel =
            channelResolver.resolveDestination(this.inputChannelName);

        if (channel instanceof SubscribableChannel) {
            this.endpoint = new EventDrivenConsumer(
                (SubscribableChannel) channel, this.handler);
        }
        else if (channel instanceof PollableChannel) {
            PollingConsumer pollingConsumer = new PollingConsumer(
                (PollableChannel) channel, this.handler);
            this.endpoint = pollingConsumer;
        }
    }

    /*
     * FactoryBean implementation
     */

    @Override
    public boolean isSingleton() {
        return true;
    }

    @Override
    public AbstractEndpoint getObject() throws Exception {
        return this.endpoint;
    }

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

}

package my.springframework.integration.core;

...

public class ChannelResolver {

    private BeanFactory beanFactory;

    public ChannelResolver(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }

    public MessageChannel resolveDestination(String destinationName)
            throws MessagingException {
        try {
            return this.beanFactory.getBean(destinationName,
                                            MessageChannel.class);
        }
        catch (BeansException ex) {
            throw new MessagingException(
                "Failed to find channel '" + destinationName + "'", ex);
        }
    }
}

if (channel instanceof SubscribableChannel) {
    this.endpoint = new EventDrivenConsumer(
        (SubscribableChannel) channel, this.handler);
}
else if (channel instanceof PollableChannel) {
    PollingConsumer pollingConsumer = new PollingConsumer(
        (PollableChannel) channel, this.handler);
    this.endpoint = pollingConsumer;
}

new EventDrivenConsumer(channel, this.handler);

public class ConsumerEndpointFactoryBean implements SmartLifecycle, ... {

    /*
     * SmartLifecycle implementation (delegates to the created endpoint)
     */

    @Override
    public boolean isAutoStartup() {
        return (this.endpoint == null) || this.endpoint.isAutoStartup();
    }

    @Override
    public int getPhase() {
        return (this.endpoint != null) ? this.endpoint.getPhase() : 0;
    }

    @Override
    public boolean isRunning() {
        return (this.endpoint != null) && this.endpoint.isRunning();
    }

    @Override
    public void start() {
        if (this.endpoint != null) {
            this.endpoint.start();
        }
    }

    @Override
    public void stop() {
        if (this.endpoint != null) {
            this.endpoint.stop();
        }
    }

    @Override
    public void stop(Runnable callback) {
        if (this.endpoint != null) {
            this.endpoint.stop(callback);
        }
    }

}

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns="http://my.springframework.org/schema/integration"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="
    http://my.springframework.org/schema/integration
    http://my.springframework.org/schema/integration/my-spring-integration.xsd
    http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd">

    <!-- App configuration -->
    <beans:bean id="orderProcessor" class="order.OrderProcessor" />

    <!-- Spring Integration configuration -->

    <gateway id="orderService" default-request-channel="newOrders"
             service-interface="order.OrderService" />

    <publish-subscribe-channel id="newOrders" />

    <service-activator id="serviceActivator" input-channel="newOrders"
                       ref="orderProcessor" />

    <channel id="pollableChannel">
        <queue/>
    </channel>

    <bridge id="bridge" input-channel="newOrders"
                        output-channel="pollableChannel" />

</beans:beans>