Chapter 4. Message Endpoints
In the previous chapter, we covered message channels in detail. You now know that some channels accept subscribers to be called in an event-driven way, whereas others require polling consumers. The former enable synchronous invocation of a handler and thereby enable transactions to span both the producer and consumer. The latter offer the flexibility of even more loosely coupled interaction, but break that transaction boundary across separate threads for the producer and sender.
From the programming perspective, the event-driven model is easier to grasp. For example, the MessageHandler interface that’s central to Spring Integration is as simple as it can be:
package org.springframework.integration.core;
public interface MessageHandler {
void handleMessage(Message<?> message);
}
All message-handling components in Spring Integration, such as transformers, splitters, and routers, implement that interface. Therefore those implementations are relatively straightforward, always reacting to a received message, much in the same way as a Java Message Service (JMS) MessageListener would. These components can be connected to any type of channel, but some of the channel types accept event-driven subscribers, whereas others must be polled. Clearly, to connect components to a channel, a certain amount of glue is necessary. In Spring Integration, that glue comes in the form of an adapter that understands how to interact with a given channel while delegating to a MessageHandler. The generic term for that adapter is a message endpoint.
A few different kinds of adaptation are required depending on the type of channel and the role of the handler component being connected to that channel. If the channel is subscribable, the handler is invoked within the thread of the sender or by the channel’s TaskExecutor if one is configured. For such a channel, no polling is required. If on the other hand the channel is one that buffers messages in a queue, such as a QueueChannel, polling is necessary.
In addition to the distinction between subscription and polling, you must consider whether a reply message is expected. If a reply is expected, would that reply be sent back to a caller or passed to the next component within a linear pipeline? Likewise, when interacting with an external system, you must consider whether a component is unidirectional (a channel adapter) or bidirectional (a gateway). One of the key benefits of a messaging system is the ability to support these different interaction models and to even switch among them by changing configuration rather than directly impacting the underlying components. In other words, it should be trivial to change from a request-reply interaction to a pipeline if you determine that an additional transformation step is necessary before sending a reply.
In this chapter, we explore the different endpoints available in Spring Integration and see how they can be used to decouple message-producing and message-consuming components from the channel type and response strategy. First we describe the difference between polling and event-driven components in detail. Then we look at those differences in the context of inbound and outbound endpoints. Finally we discuss the different response strategies to clarify the distinction between channel adapters and gateways.
Spring Integration an Implementation of the Enterprise Integration Patterns?
Spring Integration stays true to enterprise integration patterns (EIP) naming wherever sensible. Not all patterns have their direct counterpart in the API—not every pattern maps directly to an implementation. Some patterns are more conceptual, and others describe a broad category within which implementations may be classified. The message endpoint is an example of such a pattern. The endpoint is too generic to be covered by a single concrete implementation. Instead, we opted for different endpoint implementations in a class hierarchy and explicitly named patterns like transformer, splitter, and router defined in an XML schema. Likewise, taking into account the various external endpoints, the named patterns gateway and channel adapter are defined in an XML schema for each supported type of system (file, JMS, and so on). Other patterns that are supported but not literally implemented include Request-Reply, Selective Consumers, and Return Address. The first two are composite configurations of other components, and the return address pattern is supported by a message header that’s recognized by any reply-producing message handler that doesn’t have an explicit output channel configured.
Before we can talk about endpoints and decide which one is best applied to a given problem, it’s important to understand the different properties of endpoints a little better. The next section goes over different characteristics of endpoints that will keep popping up in the rest of the book.
4.1. What can you expect of an endpoint?
As noted in the introduction of this chapter, endpoint is a broad term. We need to be precise in our understanding of the properties of a certain type of endpoint before we can have a meaningful discussion about it.
One advantage of Spring Integration is the consistent naming in the namespace support. This section explores what makes endpoints different enough to give them different names. Several characteristics are important:
- Polling or event-driven
- Inbound or outbound
- Unidirectional or bidirectional
- Internal or external
Considering all the possible ways we can combine the switches, we end up with 16 candidates for unique names. Multiply this by the number of supported protocols in case of an external endpoint, and the numbers become dizzying.
Luckily, only certain combinations make sense for a given protocol, which greatly reduces the number of options. Also, Spring Integration can implicitly take care of the polling concerns. You still have plenty of options to consider—polling or event-driven, inbound or outbound (from the perspective of the Spring Integration application), unidirectional or bidirectional (one-way or request-reply respectively), internal or external (with respect to the application context)—so let’s look at the most important examples in table 4.1.
Table 4.1. Differentiating characteristics of Endpoints
|
Polling/Event-driven |
Inbound/Outbound |
Unidirectional/Bidirectional |
Internal/External |
|
|---|---|---|---|---|
| <inbound-channel-adapter> | Polling | Inbound | Unidirectional | Internal |
| <outbound-channel-adapter> | Either | Outbound | Unidirectional | Internal |
| <gateway> | Event-driven | Inbound | Bidirectional | Internal |
| <service-activator> | Either | Outbound | Bidirectional | Internal |
| <http:outbound-gateway> | Either | Outbound | Bidirectional | External |
| <amqp:inbound-channel-adapter> | Event-driven | Inbound | Unidirectional | External |
As you might guess from the table, there are some naming conventions for endpoints. Wherever possible, the names were chosen to closely mirror a corresponding pattern name. In addition, the properties are signified by consistently applied tokens. For example, a gateway is always capable of bidirectional communication. This doesn’t change between HTTP and JMS or between inbound and outbound. In contrast, a channel adapter is always unidirectional: it’s either the beginning (inbound) or the end (outbound) of a message flow. Now that you know the choices you can make before selecting an endpoint, it’s time to look into the implications of those choices. First we ask what may be the most important question when designing a messaging system.
4.1.1. To poll or not to poll?
Those of you who’ve read Hamlet know that Shakespeare got some of it right: the important decision is whether you should be a slave to your invokers or take matters into your own hands. We shouldn’t stretch the analogy to suicide, so let’s get practical.
When a user of your application is entering a lot of information, you’d like to save that input somewhere as soon as possible. Modern browsers support client-side storage facilities, but in many applications that need to be compatible with lesser browsers, this isn’t an option. Even if client-side storage is an option technically, in many cases it isn’t an option functionally. Imagine a web application that allows users to collaborate, as they would with an online word processor. This would require changes to flow from one user to another in two directions with the server as a referee in the middle. Although it’s important to get the changes from one user to another quickly, it’s much more important to get the changes from a single user to their client quickly. In other words, if a sentence reaches another user a second after one user typed it, it would be fine. If the cursor were trailing a sentence behind the current user’s typing in the client, nobody would use the application.
In most cases when you’re typing, an application should respond to a key press within 0.1 seconds; otherwise the application would be unusable. Changes from somebody else may take up to a second before you would notice that they’re delayed. You’ve probably seen a few applications that allow you to collaborate within these boundaries; the increased use of this type of collaborative editor came with the rise of Ajax. With the boundaries of network latency, the only option to make an application like this work is to asynchronously send one user’s changes to another user’s client without making the first user wait for confirmation. As discussed in chapter 3, we use messages as the central concept to transfer data from one place to another.
If messages are handed off asynchronously, the assumption is that they should eventually be received by a polling consumer. Because the poller is an active process, it requires a separate worker thread. For this purpose, Spring Integration integrates with any implementation of the core Spring TaskExecutor abstraction. Thread management is then a responsibility of that implementation and can even be delegated to a WorkManager when running in an application server environment that requires threads to be managed by the container. We discuss the details of scheduling and concurrency considerations in chapter 15. The problem at hand is to make sure the separate worker thread is available and used properly to continue processing messages that have been handed off asynchronously. Even more important, we need to understand how to switch between synchronous and asynchronous invocation.
Polling Endpoints
A polling endpoint will actively request new data to process. This data may be messages in a QueueChannel, files in a directory, messages in a JMS destination, and so on. The endpoint needs at least a single thread to perform this polling. You could hand-code an endpoint like this by creating an infinite loop and invoking receive periodically. Although it may be easier to understand the inner workings of an active endpoint because the behavior is self-contained, coding components like this yourself has some serious downsides. First of all, it requires you to write threading code, which is notoriously hard to do right. Second, it makes the component much harder to test in a unit test. Third, it becomes troublesome to integrate the component where a passive component is needed. In Spring Integration, you don’t have to write different code depending on whether you want your component to be active. Instead, whether the component should be active or passive is inferred from the configuration and handled by the framework. Configuring the components that will take the responsibility for the polling concerns is still up to you.
Event-Driven Endpoints
Asynchronous handoff usually isn’t required, and in those cases, an event-driven model should be used. This can be as simple as wrapping a plain old Java object (POJO). Exposing a web service is also a good example. The essential thing about passive, or event-driven, components is that they don’t take responsibility for thread management. They’re still responsible for proper thread safety, but this could be as simple as not maintaining any mutable state.
4.1.2. Inbound endpoints
Just as important as getting information out of the messaging system is getting information into it. This is done through inbound endpoints, typically receiving information in a format that’s not native to Spring Integration. Examples are a web service invocation, an inbound email, or a file written to a directory. A Java method invocation is also a plausible inbound integration point. The generic algorithm for an inbound endpoint is shown in figure 4.1.
Figure 4.1. Behavior of an inbound endpoint. Input is taken from an external source and then converted into a message, which is sent to a channel. This image doesn’t show the optional reply message.
Polling Inbound Endpoints
Whether an inbound endpoint needs to poll or can be event-driven depends on the architecture external to the message system. Some technical constraints, such as the lack of filesystem events in Java, affect this decision. But in most cases it depends on the system with which you’re integrating. If the external integration point is passive, the inbound endpoint becomes responsible for actively polling for updates.
In Spring Integration, several polling endpoints are provided: inbound channel adapters for files, JMS, email, and a generic method-invoking variant. The latter can be used to create custom channel adapters, such as for querying Twitter or a Really Simple Syndication (RSS) feed. Twitter and RSS adapters are both available as of Spring Integration 2.0, but before they were added, users could simply configure the generic method-invoking adapter to call their own implementations.
Event-Driven Inbound Endpoints
Often, an external system actively invokes a service on the messaging system. When that happens, the responsibility of thread management can be left out of the messaging solution and left to the invoker (in the case of a local Java method call) or to the container (in the case of a web service invocation). But as soon as QueueChannels are to be used internally, thread management becomes a concern of the messaging system again.
Examples of event-driven endpoint types supported in Spring Integration include web services (through Spring WS), Remote Method Invocation (RMI), and JMS (where Spring’s listener container takes care of the polling). Again, you can generically create a custom event-driven endpoint using the @Gateway annotation.
4.1.3. Outbound endpoints
At some point in a message flow, you’ll likely need to invoke a service that’s external to the messaging system. This may be a local method call, a web service invocation, or a message sent on another messaging system such as JMS or Extensible Messaging and Presence Protocol (XMPP). An endpoint is responsible for making this call. The general responsibilities of an outbound endpoint are depicted in figure 4.2.
Figure 4.2. Behavior of an outbound endpoint. First a message is received from a channel; then the message is converted into something the external component understands. Finally the external API is invoked. The optional result is ignored in this image.
The algorithm shown in this figure usually must implement details such as transactions and security, which we cover later. The details of conversion and the invocation of external services are also assumed to be implementation details of the specific endpoint. But the reception of the message and whether a result is generated are concepts that belong to the base API.
Polling Outbound Endpoints
If an outbound endpoint is connected to a PollableChannel, it must invoke the receive method on its input channel to receive messages. That requires scheduling an active process to do the polling periodically. In other words, step 1 in the algorithm is triggered by the endpoint. Spring Integration automatically wraps any passive components in a polling endpoint if they’re configured to be on the receiving end of a PollableChannel.
Event-Driven Outbound Endpoints
When an outbound endpoint is connected to a SubscribableChannel, it can be passive. The channel ensures that a thread exists to invoke the endpoint when a message arrives. In many cases, the thread that invoked the send method on the channel is used, but it could also be handled by a thread pool managed at the channel level. The advantage of using a thread pool is that the sender doesn’t have to wait, but the disadvantage is that a transactional boundary would be broken because the transaction context is associated with the thread. This is discussed in detail in section 4.2.
4.1.4. Unidirectional and bidirectional endpoints
The previous examples all assume unidirectional communication. Endpoints using this style of communication are called channel adapters. Unidirectional communication is often enough to establish a working solution, but bidirectional requirements are common. For these scenarios, you can use gateways. The EIP definition of gateway isn’t as clear as we would’ve liked, so we did a bit of interpretation. Mainly, the lack of clear distinction between synchronous invocation and asynchronous handoff makes the gateway concept too broad and widely overlapping with other concepts, such as channel adapter. In Spring Integration, a gateway is synonymous with synchronous two-way communication. If you need an asynchronous gateway, you should compose it from other base components (for example, an inbound and outbound channel adapter).
The Return Address
When a message reaches an endpoint, through polling or otherwise, two things can happen:
- The message is consumed and no response is generated.
- The message is processed and a response is generated.
The first case is where you’d use a channel adapter, as shown in the previous sections. The second case becomes more complicated, but luckily Spring Integration has gateways to help you support it. The complexity lies in that you must do something with the response.
Usually you’ll want to either send it further along its way to the next endpoint or send a confirmation back to the original sender when all processing for this message is done. For the first option, you set an output channel, and for the second, you omit the output channel and Spring Integration uses the REPLY_CHANNEL header to find the channel on which the original sender wants the confirmation to go.
At this point you should have a good understanding of the different high-level categories of endpoints. We explored the distinctions between polling and event-driven consumers, and we discussed both unidirectional and bidirectional behavior. In the next section, we add one of the most important pieces to the puzzle: transactions. Unfortunately, although their importance is undeniable, transactions can be confusing for many developers. We’ll do our best to remedy that situation, at least in the context of Spring Integration applications. If you’re an expert on transaction management, you might like to skip parts of the next section and just skim for Spring Integration particulars.
4.2. Transaction boundaries around endpoints
A common mistake in moving from thread per request to asynchronous handoff is to make incorrect assumptions about transaction boundaries. Another common mistake is to incorrectly assume that the security context is set in the thread that’s receiving the handoff. Both incorrect assumptions occur because security and transaction contexts are traditionally stored in a ThreadLocal. This makes sense for most web applications, and you should practice caution if you want to break this convention. Security contexts shouldn’t be shared lightly, and transactions should be kept short.
This section goes into the details of transaction management around Spring Integration endpoints. This isn’t a general work on transactions, so we focus on endpoint-related concerns here. Before we explore the technicalities of transaction management and draw parallels to security, we must establish the rationale of thread-bound contexts.
4.2.1. Why sharing isn’t always a good thing
Have you ever been working on a shared file and found out that someone else just fixed the same problem you worked on all morning? We bet you have. The first reaction to a situation like that is usually to request immediate notification of any changes that others plan to make. Or your team decides to plan better or to use a new tool.
Have you ever tried to work on something and been constantly distracted by team members who were trying to work on the same thing and tried to align with you? Have you ever been stuck in planning meetings all day? We bet you have.
These are two sides of the same problem. You must carefully consider the consequences of living with the overhead of either sharing or merging changes. This is exactly the choice you must make when establishing the extent of a transaction. From a traditional web application point of view, you want to give your user a consistent view on reality (open transaction in view or even a conversation), but you also want to ensure the smallest possible chance of your user running into a conflict with another user’s changes.
Good practice is to keep transactions small and to use compensation instead of rollback to recover from mistakes. For this reason, it makes sense to have a transaction context confined to the thread that’s servicing a single HTTP request. One commonly suggested approach is to store transaction contexts in a message instead, but that might considerably increase the scope and duration of the transaction, so it’s not natively supported by Spring Integration.
We talked about sharing a view of the world from the perspective of an editor, but how about sharing only with people you trust?
Security contexts don’t become stale over time or open the door to merge conflicts. Once properly authorized and kept in sight at all times, a user doesn’t need to reenter a password. The problem is that it’s easy to let someone slip out of sight for a second. If you share a security context with multiple threads, you need to be careful to ensure that the security details are visible only from the context of the user. If multiple users are using the application concurrently, the least error-prone isolation mechanism is to allow access only from the thread currently processing the request.
Pass Security Context Information Along with A Message
Bind contextual information from the security context to a message header before handing off to another thread. Usually, the authentication is no longer needed; just a username will do.
Classic web applications usually don’t need to worry about asynchronous handoff to work. Spring Integration comes in when asynchronous handoff becomes relevant, and at that point, you must jump through some hoops. Of course, the first thing a crafty engineer asks before jumping through hoops is whether they can get away with not doing it.
4.2.2. What are transactions, and can we get by without them?
Go to the most senior database administrator in your organization and ask this question. (It’ll be fun, we promise!) After you’ve listened carefully to his lecture, remember at least one thing: there’s no excuse for being clueless about transactions. That said, there’s no excuse for being a zealot, either. Transactions aren’t the answer to all your problems, and overusing them can be detrimental to performance and scalability.
ACID, as we all know, stands for atomic, consistent, isolated, and durable. A transaction is supposed to give you all of this. Don’t be fooled: with a lot of ifs and buts, a transaction will give you something quite close to ACID, or an exception. The tricky part of ACID is isolation in combination with global consistency. Let’s look at two examples of multiple reasonable users in a standard web application.
The Seat Selection Conflict
Jill and Jack must check in for their afternoon flight to New York. They’re on the same flight, and they have independently decided to try to find a seat next to each other. The flight is at 4:20, and they’re expecting a cab to the airport at 3:00. It’s now 2:49, so they’re in a hurry.
Jill hits the seat selection application first. She tags that she wants to travel with a companion and fills in Jack’s details. The application renders a page to Jill that shows Jack’s position on the plane. Jack is in an aisle seat on row 15. Jill stops to check the make of the plane and the seat arrangement to decide whether she wants the aisle seat opposite to Jack or whether she wants to take the middle seat next to him.
Now Jack hits the site, follows the same steps, and (this is a no-brainer) selects the middle seat next to Jill’s window seat on row 8. It won’t be as comfortable as the other seat, but at least they’ll sit together. Jack’s transaction completes fine, and he runs for his cab. Jill is done as well, hits Submit, sees that her seat number has changed, and continues to synchronize her mail before she goes offline.
It doesn’t take a lot of mental effort to predict that Jack and Jill are in for a surprise.
The most important thing to take away from this example is that the only way to prevent this scenario is to stop the world as soon as a passenger starts seat selection. But how long are you prepared to wait for that passenger to check the seating arrangement before you allow other passengers to get on? Having a globally consistent view of the world is commonly too impractical to work.
Are use cases like the story of Jill and Jack impossible to accommodate? Surely not! Anything is possible; you just need to find an effective way to deal with conflicting changes. We explore several ways next.
Fix the Problem When Things Go Wrong
If the program could figure out that Jill and Jack want to sit next to each other, it could warn them if their effort failed. For example, after Jill submits her change, the application could show an alert warning her that she’s moving away from her companion. This will come as a surprise to Jill; in her reality, she was moving toward Bill, not away from him. A properly chosen alert message would make sense to Jill. Let’s say the message is, “Your companion has moved in the meantime. Are you sure you want to move away from him/her?” Jill would be able to comprehend this and revise her seat selection.
The scenario just described is a form of eventual consistency achieved by compensating transactions. After the transaction is committed, stakeholders are invited to make compensating changes based on the new reality.
Express Intent in the Message; Leave Implementation to the Mediator
What if Jill and Jack didn’t change their seats but asked to be seated together? Then the scenario would play out in one go. Jack asks to be seated next to Jill first and is moved. Then Jill asks to be seated next to Jack, and the server knows that no move is needed.
The concession here is that Jill and Jack no longer control their locations. Allowing them to do so would require a much more complex application and, most important, would make the application more difficult to use. This isn’t always desired, but it can be powerful.
If you take a trivially simple implementation of this system, you can make it work as long as you’re prepared to send many requests. Let’s say the server understands “Please move me next to my companion,” and it blindly executes this directive without any transaction. In this case, conflicting changes such as in the example might occur, but if you keep sending the message until you see the right result as a client, the system also implements eventual consistency.
Eventual consistency is guaranteed through a redelivery policy combined with an idempotent receiver.
The two flavors of eventual consistency are both ultimately based on their not having to support a rollback. Rollbacks are the Achilles’ heel of scalability. If you can get by without them, you can loosen the transactional requirements to allow massive improvements in scalability.
But hold your horses; we’re not done yet. If you understand the reasoning behind relaxing ACID constraints, it becomes even more important to understand where you need to replace transactional boundaries with eventual consistency strategies. If you miss a beat here, you’ll end up with lost updates, dirty reads, and all their two-faced friends and relatives. There’s no excuse for being clueless about transactions, remember? Let’s look at the details of transactions around endpoints.
Where is My Transaction?
In chapter 3, we looked at transactions around channels. Remember that the transaction boundary is broken as soon as you add a task executor or a queue to a channel. In this section, we look at the beginning and end of the transaction from within the endpoint.
As we learned in the first section of this chapter, there are many different types of endpoints. As much as possible, we’ve aimed to maintain transactions within the endpoint as a rule. This makes it easier for you, as the user, to identify transaction boundaries on channels. It also allows you to extend a transaction over multiple endpoints by using the right channel in between.
A transaction is started by a poller just before it pulls a message from a Message-Source. To be precise, the poller starts a transaction only if it was configured with a transaction manager. The transaction is then committed as soon as the send method of the channel the poller is sending to returns. If the MessageSource is transactional, it participates in the same transaction as the downstream endpoints as long as the boundary isn’t broken by an asynchronous handoff.
Similarly, a poller equipped with a transaction manager starts a transaction before pulling a message from a QueueChannel. The business transaction in the downstream endpoint is wrapped by a transaction that includes the reception of the message from the channel. It also includes sending the message to the output channel. Figure 4.3 shows the two standard transaction scopes to keep in mind when designing a system.
Figure 4.3. Transaction boundaries are determined by the scope of a given thread’s responsibility, so the transactional context doesn’t propagate across an asynchronous channel.
When letting a poller take care of the transaction, as shown in the figure, transaction management is simple. The only thing left to keep an eye on is endpoints that can break transactional boundaries around your message. For example, when you use an aggregator, the messages are stored in the endpoint until they’re complete or a timeout occurs. This means the message going in doesn’t have to continue on the same thread, so it usually doesn’t participate in the same transaction.
Transaction management around an endpoint isn’t always related to the poller that invokes it. If a different thread is involved, the transactional context doesn’t typically propagate to that thread. In general, this is the case whenever an endpoint can be configured with a task executor.
Although this chapter is relatively technical in nature, much of the discussion has been theoretical. Even when discussing concerns specific to Spring Integration, the examples have been mostly at the namespace-based configuration level. In the next section, we again take a look under the hood so that you can learn more about the implementation details of message endpoints, polling consumers, and event-driven consumers. You’ll have a better understanding of the types of components that are created by the namespace parser. If you’re not interested in that level of detail, feel free to skip ahead to the summary that follows.
4.3. Under the hood
Early in this chapter, you read that Spring Integration can take care of polling concerns for you. We haven’t explained in detail how this works, and in daily life you don’t need to worry about it. If ever you find yourself debugging a Spring Integration application, though, you might benefit from knowing how this works under the hood. Things will get technical now, and if you feel you’ve seen enough at any point, feel free to skip this section. If you’re not afraid to look inside, this section is for you.
The base class for all endpoints is AbstractEndpoint, and several subtypes exist to take care of the differences between polling and event-driven behavior. Polling-Consumer wraps a MessageHandler and decorates it with a poller so that it can be connected to any PollableChannel. EventDrivenConsumer wraps a MessageHandler and connects it to any SubscribableChannel. As you can imagine, the latter is trivial. The AbstractPollingEndpoint is more complicated because it has to find a poller and handle exceptions asynchronously. Albeit more complex, this is still nothing more than a wrapper around a MessageHandler.
Things become more interesting when we start looking into what happens during the parsing of the application context where the decisions between polling and event-driven behavior are made.
4.3.1. Endpoint parsing
Each XML namespace supported by Spring Integration and the other Spring projects is typically described as a domain-specific language (DSL). Here our domain is enterprise integration. The domain model consists of messages, channels, transformers, routers, and so on. The advantage is that the elements in the namespace are closer to the concepts that we’re working with than is, say, a simple Spring bean element. You can always drop down to a lower level and configure everything as simple beans as long as you’re familiar with the API. But considering most Spring Integration users rely largely on the schemas defined in the various namespaces, we include the role of parsers in our discussion. Let’s begin with the following configuration of two transformers and two channels:
<channel id="channel1"/>
<channel id="channel2">
<queue/>
</channel>
<transformer id="transformer1"
input-channel="channel1"
expression="payload.toUpperCase()"
output-channel="nullChannel"/>
<transformer id="transformer2"
input-channel="channel2"
expression="payload.toLowerCase()"/>
The first thing to notice is that channel1 has no queue. It’s the simplest channel type available in Spring Integration. That element leads to the creation of an instance that implements the SubscribableChannel interface, and any endpoint referencing channel1 as its input channel would be invoked directly when a sender sends a message to that channel. The channel2 element leads to the creation of an instance that implements the PollableChannel interface, and internally that channel buffers messages that are sent to it. Those messages must then be explicitly received by some active poller. Let’s now walk through the result of parsing the associated endpoints.
The endpoints in this case are both message transformers. Each has a trivial expression to evaluate, and for purposes of this example, we can assume that the messages will all have a string typed payload. Each element supported by a Spring namespace handler is mapped to the implementation of Spring’s BeanDefinitionParser, which is responsible for working with that element. We avoid going into detail about the internals of the parser implementations, but for those who are curious, feel free to check out the code for AbstractConsumerEndpointParser, the common base class for several endpoint parsers in Spring Integration. Its primary role is to connect MessageHandler objects to the correct input channels, as defined in the configuration.
At parsing time, a limited amount of information is known. For example, when parsing transformer1, the parser only knows that the channel named in the input-channel attribute (channel1) should be present within the same context and that the endpoint it creates must be injected with a reference to that channel. The endpoint parser doesn’t know yet whether that’s a pollable or subscribable channel. As a result, the parser can’t know whether to create a polling or event-driven consumer. To handle that limitation, Spring Integration’s parsers rely on another common Spring feature: the FactoryBean interface. Any object defined in a Spring context which implements that interface is treated differently than other objects. Instead of instantiating the object and adding it to the context, when Spring encounters an object that implements the FactoryBean interface, it creates that factory and then invokes its getObject() method to create the object that should be added directly to the context. One advantage of that technique is that much more information is available when getObject() is called than is available at parsing time. This is how Spring Integration’s endpoint parsers avoid the limitation mentioned previously.
The implementation in this case is called ConsumerEndpointFactoryBean. As its name suggests, it’s a generic creator of consumer endpoints. It can produce either polling or event-driven consumers. Because the input channel instance referenced by name in the configuration is available at the time the factory is invoked, the decision can be made on demand. So if the referenced channel is pollable, ConsumerEndpointFactoryBean creates a PollingConsumer instance, but if the channel is subscribable, it creates an EventDrivenConsumer instance instead. Let’s now look at each of those two implementations.
4.3.2. Endpoint instantiation
The previous section described the two types of consumer endpoints: Polling-Consumer and EventDrivenConsumer. The distinction between them is based on the type of input channel: PollableChannel or SubscribableChannel, respectively. This distinction becomes clear when we investigate the constructors for each of these objects. Both take a MessageHandler that handles the messages sent by a producer to the input channel, but the interface of the expected channel is different. The PollingConsumer expects a PollableChannel:
PollingConsumer expects a PollableChannel:
package org.springframework.integration.endpoint;
public class PollingConsumer extends AbstractPollingEndpoint {
public PollingConsumer(PollableChannel inputChannel,
MessageHandler handler) {
this.inputChannel = inputChannel;
this.handler = handler;
}
...
}
An EventDrivenConsumer expects a SubscribableChannel:
package org.springframework.integration.endpoint;
public class EventDrivenConsumer extends AbstractEndpoint {
public EventDrivenConsumer(SubscribableChannel inputChannel,
MessageHandler handler) {
this.inputChannel = inputChannel;
this.handler = handler;
}
...
}
You can see that it’s not too difficult to create these as simple beans if you don’t want to use the namespace support or even the ConsumerEndpointFactoryBean for some reason. For example, here’s a bean definition for an EventDrivenConsumer:
<beans:bean id="eventDrivenConsumer"
class="org.springframework.integration.endpoint.EventDrivenConsumer">
<beans:constructor-arg ref="someSubscribableChannel"/>
<beans:constructor-arg ref="someMessageHandler"/>
</beans:bean>
We’ve mentioned several times that the primary responsibility of these consumer endpoints is to connect the MessageHandler to an input channel, but what does that mean? It has to do with the lifecycle management of these components. As you can imagine, that’s also a different issue for PollingConsumers and EventDrivenConsumers because the former requires an active poller and the latter is passive. We end our under-the-hood exploration with a quick investigation of what lifecycle management means in the context of these two consumer types.
The foundation for managing the lifecycle of any components in Spring Integration is the Spring Framework. The core framework defines a Lifecycle interface:
package org.springframework.context;
public interface Lifecycle {
voidstart();
voidstop();
boolean isRunning();
}
As you can see, methods are available for starting and stopping a component. For an EventDrivenConsumer, the start operation consists of calling the Subscribable-Channel’s subscribe method while passing the MessageHandler. Likewise, the unsubscribe method is called from within the stop operation. PollingConsumers are more complicated. The start operation of a PollingConsumer schedules the poller task with the trigger that’s set on that consumer. The trigger implementation may be either a PeriodicTrigger (for fixed delay or fixed rate) or a CronTrigger. The stop operation of a PollingConsumer cancels the poller task so that it stops running.
You may wonder who’s responsible for calling these lifecycle methods. Stopping is more straightforward than starting, because when a Spring ApplicationContext is closing, any lifecycle component whose isRunning method returns true is stopped automatically. When it comes to starting, Spring 3.0 added an extension called Smart-Lifecycle to the Lifecycle interface. SmartLifecycle adds a boolean method, isAutoStartup(), to indicate whether startup should be automatic. It also extends the new Phased interface with its getPhase() method to provide the concept of life-cycle phases so that starting and stopping can be ordered.
Spring Integration consumer endpoints make use of this new interface. By default, they all return true from the isAutoStartup method. Also, the phase values are such that any EventDrivenConsumer is started (subscribed to its input channel) before any PollingConsumer is activated. That’s important because PollingConsumers often produce reply messages that may flow downstream to EventDrivenConsumers.
4.4. Summary
This chapter dove deeply into Spring Integration internals that you likely won’t need to think about on a regular basis. The higher-level components such as transformer, router, and channel adapter are the typical focal points. But the information you learned here provides a strong foundation for understanding how those higher-level components work. You now know that message endpoints break down into either polling consumers or event-driven consumers. You also know that regardless of the type of a given consumer, it delegates to a MessageHandler. The lifecycle of an endpoint consists of managing the connection between the MessageHandler and the Message-Channel where messages are received. The way that connection is managed (either by a poller or a simple subscription) is the distinguishing factor for the types of consumer implementation.
Now that we’ve explored the low-level details of endpoints in this chapter and of messages and channels in the previous chapter, we’re ready to move to a higher level. In the chapters that follow, you’ll learn about several components that map to the enterprise integration patterns. These include routers, splitters, aggregators, and a wide variety of channel adapters. We begin this journey by considering the role of Spring Integration in the larger context of a real-world application.
Business concerns are usually the most critical part of an application, and that fact motivates Spring Integration and the Spring Framework to promote a clean separation between integration and business concerns. The goal of these frameworks is to allow developers to focus on implementing the business functionality in their application’s domain without spending excessive time on infrastructure. The next chapter provides several examples in the flight management domain of our sample application.