Salmon Run: rest

Saturday, March 27, 2010

Java REST Client Interface for Ehcache server

Ehcache is a popular caching library in the Java world. So far, I was aware only of the Ehcache library, using which you could build in-JVM caches. Lately (since late 2008 actually), they have come up with the Ehcache server, with which you can create remote caches that your application can access over HTTP.

This obviously has enormous implications for scalability. Granted, cache access times over a network are much higher than in-memory access, I think this is a fair tradeoff to make when you are dealing with potentially very large caches. Keeping your cache behind a server frees up your application JVM's memory. Also, if you want more cache, you can partition it out into more servers.

Our first cut of the ehcache setup was something like this. A bunch of applications would maintain their own in-JVM caches, but these caches would communicate with each other and replicate over RMI, as shown in the diagram below. This was mainly to test out the local replicated mode, which we used in our final setup for fault tolerance (see below).

The next step was to take the cache and put it behind the cache server. We used Ehcache Standalone Server version 0.8. It comes packaged within a Glassfish Application Server, so all we had to do was to expand the tarball, and update the ehcache.xml file in the war/WEB-INF/classes subdirectory with our local replicated cache definitions. To start the Ehcache server, use bin/startup.sh (this works fine, but needs cleaning up to log to a file, etc). Out of the box, its set to have the cache server listen on port 8080, you can change it to whatever you want instead.

Now, if we needed more cache memory, we could now add more server pairs (paired for fault tolerance, see diagram above) and partition the key space. Then on the client, we could do a simple hashmod of the key and direct it to the appropriate load balancer.

To make the transition seamless, I factored out all ehcache access code into a helper class - the application code was basically doing get(), put() and delete() calls on the cache - then switched the calls from local to remote cache. The code is shown below.

package com.mycompany.myapp.helpers;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;

import net.sf.ehcache.Element;

import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.methods.DeleteMethod;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.InputStreamRequestEntity;
import org.apache.commons.httpclient.methods.PutMethod;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.stereotype.Service;

@Service("cacheHelper")
public class CacheHelper {

  private final Logger logger = LoggerFactory.getLogger(getClass());
  
  @Autowired private String serviceUrl;
  @Autowired private int readTimeout;
  @Autowired private int connectTimeout;
  @Autowired private int maxRetries;
  @Autowired private HttpClient httpClient;
  
  public Element get(String cacheName, String key) throws Exception {
    String url = StringUtils.join(new String[] {
      serviceUrl, cacheName, key}, "/");
    GetMethod getMethod = new GetMethod(url);
    configureMethod(getMethod);
    ObjectInputStream oin = null;
    int status = -1;
    try {
      status = httpClient.executeMethod(getMethod);
      if (status == HttpStatus.SC_NOT_FOUND) {
        // if the content is deleted already
        return null;
      }
      InputStream in = getMethod.getResponseBodyAsStream();
      oin = new ObjectInputStream(in);
      Element element = (Element) oin.readObject();
      return element;
    } catch (IOException e) {
      logger.warn("GET Failed (" + status + ")", e);
    } finally {
      IOUtils.closeQuietly(oin);
      getMethod.releaseConnection();
    }
    return null;
  }
  
  public void put(String cacheName, String key, Serializable value) 
      throws Exception {
    Element element = new Element(key, value);
    String url = StringUtils.join(new String[] {
      serviceUrl, cacheName, key}, "/");
    PutMethod putMethod = new PutMethod(url);
    configureMethod(putMethod);
    ObjectOutputStream oos = null;
    int status = -1;
    try {
      ByteArrayOutputStream bos = new ByteArrayOutputStream();
      oos = new ObjectOutputStream(bos);
      oos.writeObject(element);
      putMethod.setRequestEntity(new InputStreamRequestEntity(
        new ByteArrayInputStream(bos.toByteArray())));
      status = httpClient.executeMethod(putMethod);
    } catch (Exception e) {
      logger.warn("PUT Failed (" + status + ")", e);
    } finally {
      IOUtils.closeQuietly(oos);
      putMethod.releaseConnection();
    }
  }
  
  public void delete(String cacheName, String key) throws Exception {
    String url = StringUtils.join(new String[] {
      serviceUrl, cacheName, key}, "/");
    DeleteMethod deleteMethod = new DeleteMethod(url);
    configureMethod(deleteMethod);
    int status = -1;
    try {
      status = httpClient.executeMethod(deleteMethod);
    } catch (Exception e) {
      logger.warn("DELETE Failed (" + status + ")", e);
    } finally {
      deleteMethod.releaseConnection();
    }
  }
  
  private void configureMethod(HttpMethod method) {
    if (readTimeout > 0) {
      method.getParams().setSoTimeout(readTimeout);
    }
    if (maxRetries > 0) {
      method.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, 
        new DefaultHttpMethodRetryHandler(maxRetries, false));
    }
  }
}

Calling code is no different from the code calling the local cache. Instead of cache.XXX() you now do cacheHelper.XXX() calls. The service URL is /ehcache/rest.

Here is the cache definition for one of our caches. You can repeat the cache block for multiple caches. The peer listener and peer provider factory defintions are shared across multiple caches.

<ehcache>
    ...
    <defaultCache .../>
    
    <cache name="my-cache"
        maxElementsInMemory="10000"
        eternal="true"
        diskPersistent="true"
        overflowToDisk="true">
      <cacheEventListenerFactory 
        class="net.sf.ehcache.distribution.RMICacheReplicatorFactory"/>
      <bootstrapCacheLoaderFactory 
        class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory"/>
    </cache>

    <cacheManagerPeerListenerFactory
      class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"/>
    <cacheManagerPeerProviderFactory 
      class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
      properties="hostName=localhost,
                  peerDiscovery=automatic,
                  multicastGroupAddress=224.0.0.1,
                  multicastGroupPort=4446,
                  timeToLive=1"/>

</ehcache>

Obviously, neither the code or configuration is rocket science, there are enough examples in the Ehcache site for someone to build this stuff themselves. But the code examples of using the Ehcache server are quite generic, they concentrate on caching strings, while most people who have used Ehcache in local mode tend to cache real Serializable objects. Also, most enterprise type places I know tend to use Jakarta's HTTPClient rather than Java's URLConnection. Also, it took me a fair bit of time to figure out and test the distributed cache configuration. So if you are looking for a quick ramp up to using Ehcache server, then this post may be helpful.

Saturday, July 26, 2008

A RSS Feed Client in Java

In his article, "Demystifying RESTful Data Coupling", Steve Vinoski says:

Developers who favor technologies that promote interface specialization typically raise two specific objections to the uniform-interface constraint designed into the Representational State Transfer (REST) architectural style. One is that different resources should each have specific interfaces and methods that more accurately reflect their precise functionality. The other objection to the concept of a uniform interface is that it merely shifts all coupling issues and other problems to the data exchanged between client and server.

We have faced similar concerns from clients of our RSS-2.0 based REST API. While the concerns are easier to address because our XML format is a well-known standard, and we can point them to several implementations of RSS feed parsers, such as Mark Pilgrim's Python Universal Feed Parser, the ROME Fetcher, or the Jakarta FeedParser, to name a few. In addition, because of the popularity of RSS, almost all major programming languages have built-in support or contributed modules to parse various flavors of RSS, so clients can usually find an off-the-shelf parser or toolkit that works well with their programming language of choice.

However, thinking through Steve Vinoski's comment a little more with reference to my particular context, I came up with the idea of using the ROME SyndFeed object as a Data Transfer Object (DTO). Since ROME is a popular project, its data structures are well documented, both on its own website and in various books such as Dave Johnson's "RSS and Atom in Action", client programmers can look at publicly available documentation to figure out how to convert the SyndFeed into objects that would be consumable by their application.

What makes the task easier is that ROME already has a Fetcher module, which takes care of the various nuances of parsing special headers from RSS feeds, local caching and such. While the generally available 0.9 release (at the time of this writing) does not have support for connection and read timeouts on the underlying HTTP client, the version in CVS (and probably releases following 0.9) would have this support, so I used that.

So what we would provide would be a "client library" consisting of a single class:

// ApiClient.java
package com.healthline.feeds.client;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLEncoder;
import java.util.Map;
import java.util.UUID;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.sun.syndication.feed.synd.SyndFeed;
import com.sun.syndication.fetcher.FetcherException;
import com.sun.syndication.fetcher.impl.HashMapFeedInfoCache;
import com.sun.syndication.fetcher.impl.HttpClientFeedFetcher;
import com.sun.syndication.io.FeedException;

/**
 * Client for API. Based on the ROME FeedFetcher project.
 * Provides a single execute() method to point to the RSS based webservice.
 * The response is RSS 2.0 XML, which is converted into a SyndFeed object and 
 * returned to the caller to parse as needed.
 */
public class ApiClient {

  private final Log log = LogFactory.getLog(getClass());
  
  private URL serviceUrl;

  private HttpClientFeedFetcher fetcher = null;
  
  /**
   * Constructs a ApiClient instance.
   * @param serviceUrl the location of the service.
   * @param useLocalCache true if you want to cache responses locally.
   * @param connectTimeout the connection timeout (ms) for the network connection.
   * @param readTimeout the read timeout (ms) for the network connection.
   */
  public ApiClient(URL serviceUrl, boolean useLocalCache, int connectTimeout, 
      int readTimeout) {
    super();
    this.serviceUrl = serviceUrl;
    fetcher = new HttpClientFeedFetcher();
    fetcher.setUserAgent("MyApiClientFetcher-1.0");
    fetcher.setConnectTimeout(connectTimeout);
    fetcher.setReadTimeout(readTimeout);
    if (useLocalCache) {
      fetcher.setFeedInfoCache(HashMapFeedInfoCache.getInstance());
    }
  }
  
  /**
   * Executes a service request and returns a ROME SyndFeed object.
   *
   * @param methodName the methodName to execute.
   * @param params a Map of name value pairs.
   * @return a SyndFeed object.
   */
  public SyndFeed execute(String methodName, Map<String,String> params) {
    URL feedUrl = buildUrl(methodName, params);
    SyndFeed feed = null;
    try {
      feed = fetcher.retrieveFeed(feedUrl);
    } catch (FetcherException e) {
      throw new RuntimeException("Failed to fetch URL:[" + 
        feedUrl.toExternalForm() + "]. HTTP Response code:[" + 
        e.getResponseCode() + "]", e);
    } catch (FeedException e) {
      throw new RuntimeException("Failed to parse response for URL:[" + 
        feedUrl.toString() + "]", e);
    } catch (IOException e) {
      throw new RuntimeException("IO Error fetching URL:[" + 
        feedUrl.toString() + "]", e);
    }
    return feed;
  }

  /**
   * Convenience method to build up the request URL from the method name and
   * the Map of query parameters.
   * @param methodName the method name to execute.
   * @param params the Map of name value pairs of parameters.
   * @return
   */
  private URL buildUrl(String methodName, Map<String,String> params) {
    StringBuilder urlBuilder = new StringBuilder(serviceUrl.toString());
    urlBuilder.append("/").append(methodName);
    int numParams = 0;
    for (String paramName : params.keySet()) {
      String paramValue = params.get(paramName);
      if (StringUtils.isBlank(paramValue)) {
        continue;
      }
      try {
        paramValue = URLEncoder.encode(paramValue, "UTF-8");
      } catch (UnsupportedEncodingException e) {
        // will never happen, but just in case it does, we throw the error up
        throw new RuntimeException(e);
      }
      urlBuilder.append(numParams == 0 ? "?" : "&").
      append(paramName).
      append("=").
      append(paramValue);
      numParams++;
    }
    try {
      if (log.isDebugEnabled()) {
        log.debug("Requesting:[" + urlBuilder.toString() + "]");
      }
      return new URL(urlBuilder.toString());
    } catch (MalformedURLException e) {
      throw new RuntimeException("Malformed URL:[" + urlBuilder.toString() + "]", e);
    }
  }
}

All the client has to do is instantiate this class with the parameters, then execute the service command. This is completely generic, by the way, not tied to our API service in any way. As an example, I tried hitting the RSS feed for the National Public Radio (NPR) Top Stories Page with the test code below:

Based on our original requirement, the objective is to convert the SyndFeed object returned from the call to ApiClient.execute() to an appropriate user object. We call our user object SearchResult, and it is a POJO as shown below:

// SearchResult.java
public class SearchResult {

  private String title;
  private String url;
  private String summary;
  // auto-generated getters and setters removed for brevity
  ...
}

// NprApiClient.java
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.mycompany.feeds.client.ApiClient;
import com.sun.syndication.feed.synd.SyndCategory;
import com.sun.syndication.feed.synd.SyndEntry;
import com.sun.syndication.feed.synd.SyndFeed;

public class NprApiClient {

  private static final String SERVICE_URL = "http://www.npr.org/rss";
  private static final boolean USE_CACHE = true;
  private static final int DEFAULT_CONN_TIMEOUT = 5000;
  private static final int DEFAULT_READ_TIMEOUT = 1000;
  
  private ApiClient apiClient;
  
  public NprApiClient() throws Exception {
    apiClient = new ApiClient(new URL(SERVICE_URL), USE_CACHE, DEFAULT_CONN_TIMEOUT, 
      DEFAULT_READ_TIMEOUT);
  }
  
  @SuppressWarnings("unchecked")
  public List<SearchResult> getTopStories() {
    Map<String,String> args = new HashMap<String,String>();
    args.put("id", "1001");
    SyndFeed feed = apiClient.execute("rss.php", args);
    List<SyndEntry> entries = feed.getEntries();
    List<SearchResult> results = new ArrayList<SearchResult>();
    for (SyndEntry entry : entries) {
      SearchResult result = new SearchResult();
      result.setTitle(entry.getTitle());
      result.setUrl(entry.getLink());
      result.setSummary(entry.getDescription().getValue());
      results.add(result);
    }
    return results;
  }
  
  public static void main(String[] args) {
    try {
      NprApiClient client = new NprApiClient();
      List<SearchResult> results = client.doTopStorySearch();
      for (SearchResult result : results) {
        System.out.println(result.getTitle());
        System.out.println("URL:" + result.getUrl());
        System.out.println(result.getSummary());
        System.out.println("--");
      }
    } catch (Exception e) {
      System.err.println(e.getMessage());
      throw new RuntimeException(e);
    }
  }
}

Here are the (partial) results from the run. I have truncated the results after the first few results but you get the idea.

Housing Bill Clears Senate, Awaits Bush's Signature
URL:http://www.npr.org/templates/story/story.php?storyId=92964747&ft=1&f=1001
The Senate met in a rare Saturday session and gave final congressional approval to a wide-ranging 
housing bill.  The bill aims to bolster the sagging housing market and includes measures aimed at 
shoring up Fannie Mae and Freddie Mac. The president says he'll sign it when it reaches his desk, 
early next week.
--
What's The Deal With The XM-Sirius Merger?
URL:http://www.npr.org/templates/story/story.php?storyId=92960423&ft=1&f=1001
The FCC has approved the merger of XM and Sirius satellite radio after 17 months of behind-
the-scenes negotiations. While some critics have said the merger represents a monopoly, it 
appears that the two weak companies may be combining to form one weak company.
--
Military Tribunals Begin At Guantanamo
URL:http://www.npr.org/templates/story/story.php?storyId=92960420&ft=1&f=1001
The first war crimes trials since World War II started this week at Guantanamo Bay. Andrew 
McBride, a former Justice Department official, discusses the trials, as well as how Guantanamo's 
war crimes compare with those of 1945.
--
...

Although the above code is good enough for a standard RSS feed parsing client, I was not able to get results out of our custom tags (for our RSS-2.0 based API I spoke about earlier). I plan to investigate this, since we use a variety of open-source RSS custom modules (such as Amazon's OpenSearch as well as our own home-grown custom module to satisfy several data requirements that cannot be accommodated by standard RSS 2.0. Because of this, it is important for our clients to be able to parse out our custom module and its contents from the SyndFeed object.

I will investigate this on my own and write about it in a future post. From what I see so far, the ROME Fetcher is not passing the custom module information through in the SyndFeed object it parses out of the XML. It is possible that I am just missing some configuration piece that would enable it. In the meantime, if you happen to know how to do this, would really appreciate you letting me know.

Salmon Run

Saturday, March 27, 2010

Java REST Client Interface for Ehcache server

Saturday, July 26, 2008

A RSS Feed Client in Java

Posts

Labels

Blogs I Read

About me

My Nerd Rating

Visitor Map

Contact Me