Channel adapters

In previous articles we discovered some alternative ways to send messages. Another alternative feature to do that are channel adapters, covered here.

At the begin of this article we'll talk theoritecaly about channel adapters, as about components of entreprise integration patterns. After we'll focus on theirs implementation in Spring Integration. At the end we'll show sample test cases with channel adapters.

What channel adapter is ?

Basically, channel adapter allows an application to connect to the messaging system. In this way, connected application is able to receive messages treated by messaging system. Channel adapter can also work on received messages and send them again to the messaging system.

Channel adapter is also part of another entreprise integration pattern called messaging bridge. This pattern defines the way of connect two complete messaging systems. The solution is based on channel adapters which are used to connect a pair of corresponding channels.

Channel adapters in Spring Integration

Two basic channel adapters are represented in Spring Integration configuration by x-channel-adapter tags, where x can be replaced by inbound or outbound. Channel adapters are in fact message endpoints which enable the connection with exterior systems, as HTTP, JDBC, e-mail or web services. Spring provides some of default channel adapters for listed elements.

At the begin we discovered the existence of two channel adapters: inbound and outbound. The first one is used by Spring-managed objects which handle incoming message and can return it modified to receiver channel. An important element of inbound channel adapter is poller. It will receive message worked by channel adapter and try to send it further, to receiver. In the other hand, outbound message channel is used to simply handle message, without returning it into receiver. However, both adapters play well together in the way where inbound adapter gets sent message and passes it further, to channel associated with outbound adapter.

To have an idea how does it work, let's take a look on HTTP channel adapters. If we search the method constructing objects for HTTP's chanel adapters, we'll find two parsers for XML configuration: org.springframework.integration.http.config.HttpOutboundChannelAdapterParser and org.springframework.integration.http.config.HttpInboundEndpointParser classes. The first one produces outbound channel adapter which is an instance of HttpRequestExecutingMessageHandler. Its implementation for abstract handleRequestMessage(Message<?> requestMessage) method manipulates incoming message. It can return null or new message:

@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
  // ...
  try {
    HttpMethod httpMethod = this.determineHttpMethod(requestMessage);
    // ...
    Object expectedResponseType = this.determineExpectedResponseType(requestMessage);

    HttpEntity<?> httpRequest = this.generateHttpRequest(requestMessage, httpMethod);
    Map<String, ?> uriVariables = this.determineUriVariables(requestMessage);
    UriComponents uriComponents = UriComponentsBuilder.fromUriString(uri).buildAndExpand(uriVariables);
    URI realUri = this.encodeUri ? uriComponents.toUri() : new URI(uriComponents.toUriString());
    ResponseEntity<?> httpResponse;
    if (expectedResponseType instanceof ParameterizedTypeReference<?>) {
      httpResponse = this.restTemplate.exchange(realUri, httpMethod, httpRequest, (ParameterizedTypeReference<?>) expectedResponseType);
    }
    else {
      httpResponse = this.restTemplate.exchange(realUri, httpMethod, httpRequest, (Class<?>) expectedResponseType);
    }
    if (this.expectReply) {
      HttpHeaders httpHeaders = httpResponse.getHeaders();
      Map<String, Object> headers = this.headerMapper.toHeaders(httpHeaders);
      AbstractIntegrationMessageBuilder<?> replyBuilder = null;
      if (httpResponse.hasBody()) {
        Object responseBody = httpResponse.getBody();
        replyBuilder = (responseBody instanceof Message<?>) ?
        this.getMessageBuilderFactory().fromMessage((Message<?>) responseBody) 
        : this.getMessageBuilderFactory().withPayload(responseBody);
      }
      else {
        replyBuilder = this.getMessageBuilderFactory().withPayload(httpResponse);
      }
      replyBuilder.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, httpResponse.getStatusCode());
      return replyBuilder.copyHeaders(headers).build();
    }
    return null;
  }
  catch (MessagingException e) {
    throw e;
  }
  catch (Exception e) {
    throw new MessageHandlingException(requestMessage, "HTTP request execution failed for URI [" + uri + "]", e);
  }
}

The second parser creates an instance of HttpRequestHandlingController if view-name or view-expression attributes are defined, or HttpRequestHandlingMessagingGateway if these attributes are not defined. This last object can work on incoming message and eventually return an ModelAndView object:

public final void handleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse)
throws ServletException, IOException {
  Object responseContent = null;
  Message<?> responseMessage;

  final ServletServerHttpRequest request = new ServletServerHttpRequest(servletRequest);
  final ServletServerHttpResponse response = new ServletServerHttpResponse(servletResponse);

  try {
    responseMessage = super.doHandleRequest(servletRequest, servletResponse);
    if (responseMessage != null) {
      responseContent = setupResponseAndConvertReply(response, responseMessage);
    }
  }
  catch (Exception e) {
    responseContent = handleExceptionInternal(e);
  }
  if (responseContent != null) {

    if (responseContent instanceof HttpStatus) {
    response.setStatusCode((HttpStatus) responseContent);
    }
    else {
    this.writeResponse(responseContent, response, request.getHeaders().getAccept());
    }
  }
  else {
    setStatusCodeIfNeeded(response);
  }
}

// Where super.doHandleRequest invokes:
protected final Message<?> doHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse)
throws IOException {
  if (this.isShuttingDown()) {
    return createServiceUnavailableResponse();
  }
  else {
    return actualDoHandleRequest(servletRequest, servletResponse);
  }
}

@SuppressWarnings({"rawtypes", "unchecked"})
private Message<?> actualDoHandleRequest(HttpServletRequest servletRequest, HttpServletResponse servletResponse)
throws IOException {
this.activeCount.incrementAndGet();
  try {
    ServletServerHttpRequest request = this.prepareRequest(servletRequest);

    Object requestBody = null;
    if (this.isReadable(request)) {
      requestBody = this.extractRequestBody(request);
    }
    HttpEntity httpEntity = new HttpEntity(requestBody, request.getHeaders());

    StandardEvaluationContext evaluationContext = this.createEvaluationContext();
    evaluationContext.setRootObject(httpEntity);

    evaluationContext.setVariable("requestAttributes", RequestContextHolder.currentRequestAttributes());

    MultiValueMap<String, String> requestParams = this.convertParameterMap(servletRequest.getParameterMap());
    evaluationContext.setVariable("requestParams", requestParams);

    evaluationContext.setVariable("requestHeaders", new ServletServerHttpRequest(servletRequest).getHeaders());

    Cookie[] requestCookies = servletRequest.getCookies();
    if (!ObjectUtils.isEmpty(requestCookies)) {
      Map<String, Cookie> cookies = new HashMap<String, Cookie>(requestCookies.length);
      for (Cookie requestCookie : requestCookies) {
        cookies.put(requestCookie.getName(), requestCookie);
      }
      evaluationContext.setVariable("cookies", cookies);
    }

    Map<String, String> pathVariables =
    (Map<String, String>) servletRequest.getAttribute(HandlerMapping.URI_TEMPLATE_VARIABLES_ATTRIBUTE);

    if (!CollectionUtils.isEmpty(pathVariables)) { 
      evaluationContext.setVariable("pathVariables", pathVariables);
    }

    Map<String, MultiValueMap<String, String>> matrixVariables =
    (Map<String, MultiValueMap<String, String>>) servletRequest
    .getAttribute(HandlerMapping.MATRIX_VARIABLES_ATTRIBUTE);

    if (!CollectionUtils.isEmpty(matrixVariables)) {
      evaluationContext.setVariable("matrixVariables", matrixVariables);
    }

    Map<String, Object> headers = this.headerMapper.toHeaders(request.getHeaders());
    Object payload = null;
    if (this.payloadExpression != null) {
      // create payload based on SpEL
      payload = this.payloadExpression.getValue(evaluationContext);
    }
    if (!CollectionUtils.isEmpty(this.headerExpressions)) {
      for (String headerName : this.headerExpressions.keySet()) {
        Expression headerExpression = this.headerExpressions.get(headerName);
        Object headerValue = headerExpression.getValue(evaluationContext);
        if (headerValue != null) {
          headers.put(headerName, headerValue);
        }
      }
    }

    if (payload == null) {
      if (requestBody != null) {
        payload = requestBody;
      }
      else {
        payload = requestParams;
      }
    }

    AbstractIntegrationMessageBuilder<?> messageBuilder = null;

    if (payload instanceof Message<?>) {
      messageBuilder = this.getMessageBuilderFactory().fromMessage((Message<?>) payload)
      .copyHeadersIfAbsent(headers);
    }
    else {
      messageBuilder = this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers);
    }

    Message<?> message = messageBuilder
    .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_URL,
    request.getURI().toString())
    .setHeader(org.springframework.integration.http.HttpHeaders.REQUEST_METHOD,
    request.getMethod().toString())
    .setHeader(org.springframework.integration.http.HttpHeaders.USER_PRINCIPAL,
    servletRequest.getUserPrincipal())
    .build();

    Message<?> reply = null;
    if (this.expectReply) {
      reply = this.sendAndReceiveMessage(message);
    }
    else {
      this.send(message);
    }
    return reply;
  }
  finally {
    this.postProcessRequest(servletRequest);
    this.activeCount.decrementAndGet();
  }
}

To resume these two methods, they are able to construct Message instance based on received HTTP parameters. AbstractIntegrationMessageBuilder plays an important role because it's this object which constructs Message. If you look carefully in these methods, you can also find some intergration-oriented objects, as RestTemplate object or invocation of protected Message<?> sendAndReceiveMessage(Object object) method defined in MessagingGatewaySupport inherited by HttpRequestHandlingMessagingGateway. Otherwise, the rest of operations consists mainly on analyzing incoming data and adapt it into channel adapter's configuration.

Example of channel adapters in Spring Integration

To see the general idea of channel adapters, we'll write a sample case when one channel sends a message handled by channel adapter. After that, this channel adapter sends the message to receiver channel. The configuration for this case looks like:

<context:annotation-config />
<context:component-scan base-package="com.waitingforcode"/>

<int:channel id="sender1" />
<int:channel id="receiver1">
  <int:queue capacity="30" />
</int:channel>

<!-- sample configuration for outbound channel adapter. Note that channel attribute here means
the channel sending the message.
-->
<int:outbound-channel-adapter channel="sender1" ref="productChannelAdapter" method="handleProduct" />

<!-- sample configuration for inbound channel adapter. This adapter is applied to channel receiving the messages -->
<int:inbound-channel-adapter ref="productChannelAdapter" method="prepareToFurtherSend" channel="receiver1">
  <int:poller fixed-rate="2000" />
</int:inbound-channel-adapter>

The configuration is quite basic. As you can see, two channel adapters uses the same bean, productChannelAdapter. When the message is sent by sender1 channel, it'll be worked by handleProduct. Further, the message will be passed to prepareToFurtherSend method. After that, this method'll sent the message with modified payload to receiver1 channel. ProductChannelAdapter class sets fix price for handled product, as you can observe it in following code:

/**
 * Sample channel adapter receiving messages in handleProduct and preparing them into further sending in prepareToFurtherSend.
 *
 * @author Bartosz Konieczny
 */
@Component
public class ProductChannelAdapter {

  private Queue<Product> received = new LinkedList<Product>();

  public static final double PRICE = 399.99d;

  public void handleProduct(Product product) {
    // even if we change the price here, it won't be sent to receiver's message channel
    product.setPrice(0);
    received.add(product);
  }

  public Product prepareToFurtherSend() {
    Product product = getLastProduct();
    product.setPrice(PRICE);
    return product;
  }

  public Product getLastProduct() {
    return received.poll();
  }

}

Test case using this adapter is:

/**
 * Test cases for channel adapters.
 *
 * @author Bartosz Konieczny
 */
@ContextConfiguration(locations = "classpath:META-INF/channel-adapter.xml")
@RunWith(SpringJUnit4ClassRunner.class)
public class ChannelAdaptersTest {

  @Autowired
  @Qualifier("sender1")
  private DirectChannel sender1;

  @Autowired
  @Qualifier("receiver1")
  private QueueChannel receiver1;

  @Autowired
  private ProductChannelAdapter productChannelAdapter;

  @Test
  public void testOutBound() {
    Product milk = new Product();
    milk.setName("milk");
    Message<Product> milkMsg = MessageBuilder.<Product>withPayload(milk).build();
    sender1.send(milkMsg, 2000);

    Product received = productChannelAdapter.getLastProduct();

    assertEquals("Bad message's payload was sent", milk.getName(), received.getName());
    receiver1.receive(2000);
  }

  @Test
  public void testInBound() {
    Product milk = new Product();
    milk.setName("milk");
    Message<Product> milkMsg = MessageBuilder.<Product>withPayload(milk).build();
    sender1.send(milkMsg, 2000);

    Message<?> receivedMsg = receiver1.receive(2000);
    Product receivedProduct = (Product) receivedMsg.getPayload();
    assertEquals("Message was not passed through inbound-channel-adapter before coming to receiver's channel",
            ProductChannelAdapter.PRICE, receivedProduct.getPrice(), 0);

    // this try will fail because of poller's sending rate (3 seconds) and receiver's waiting time (1 seconds)
    Product coffee = new Product();
    coffee.setName("coffee");
    Message<Product> coffeeMsg = MessageBuilder.<Product>withPayload(coffee).build();
    receivedMsg = receiver1.receive(1000);
    assertNull("Message shouldn't be received because of poller's too small sending rate", receivedMsg);
  }

}

We covered here the concept of channel adapters. We discovered that these elements allow to connect an application to messaging system. An example of connected application can be simple website using special kind of adapters from org.springframework.integration.http package. We also saw two implementations of adapters in Spring Integration: inbound, returning a message to reply channel, and outbound not generating the response.