SimpleHTTPJob

Thanks to my collegue Alex Collins who implemented this in our project.

Below is the description in his own words.

SimpleHTTPJob was implemented in order to allow for load-balanced service invocation from the scheduler instead of Remote EJB invocation. Previously, the CustomerJobScheduler class as described later in the article was used to lookup and invoke an EJB. This has the disadvantage that we can't redirect such invocations without making significant changes in the container. Example xml we are using

<schedule>
        <job>
            <name>exampleProcessorHTTP</name>
            <group>MYJOB_GROUP</group>
 
            <description>Calls the PCOV3 architecture over HTTP.</description>
            <job-class>test.SimpleHTTPJob</job-class>
            <job-data-map allows-transient-data="false">
 
                <entry>
                    <key>endpoint</key>
                    <value>http://localhost:8000/demo-SNAPSHOT/ExampleTrigger/?jndiName=exampleProcessor_V3</value>
                </entry>
 
  <entry>
   <key>http-method</key>
   <value>POST</value>
  </entry>
 
   <entry>
   <key>http-body</key>
   <value>ThisBody=IsIgnored</value>
     </entry>
     
     <!-- Relative to config/. If present, http-body is ignored -->
     <entry>
   <key>http-body-file</key>
   <value>request.xml</value>
    </entry>
 
        <trigger>
            <cron>
                <cron-expression>0/30 * * * * ?</cron-expression>
            </cron>
        </trigger>
    </schedule> 

As you can see the endpoint configured triggers a ExampleTriggerServlet

public class ExampleTriggerServlet extends HttpServlet
{
  protected void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException
  {
    String errStr;
    OutputStream out = res.getOutputStream();
    PrintWriter writer = new PrintWriter(out);
    String jndiName = String.valueOf(req.getParameter("jndiName"));
    boolean success = false;
 
    Collection errCol = new ArrayList(1);
    try
    {
        InitialContext ctx = new InitialContext();
        ResourceController resource = (ResourceController)ctx.lookup(jndiName);
        Map parameterMap = req.getParameterMap();
        Map resourceParamMap = new HashMap(parameterMap.keySet().size());
 
        for (Map.Entry entry : parameterMap.entrySet()) {
          String[] value = (String[])entry.getValue();
 
          if ((value != null) && (value.length > 0))
          {
            resourceParamMap.put(entry.getKey(), value[0]);
          }
        }
        resource.checkResource(resourceParamMap);
        success = true;
      }
    }
    catch (Exception e) {
    } finally {
      res.setContentType("application/json");
      writer.printf("{'success': %s, ", new Object[] { Boolean.valueOf(success) });
      writer.printf("'errors': [", new Object[0]);
 
      for (String errStr : errCol) {
        writer.printf("'%s',", new Object[] { URLEncoder.encode(errStr, "UTF-8") });
      }
 
      writer.printf("]", new Object[0]);
      writer.printf("}", new Object[0]);
      writer.flush();
    }
  }
} 
 
SimpleHTTPJob.java
 
public class SimpleHTTPJob implements Job {
 
    private final HttpClient client;
 
    public SimpleHTTPJob() {
        client = new DefaultHttpClient();
    }
 
    private static enum HTTPMethod {
        GET, POST, PUT, DELETE;
    }
 
    private static final class HTTPRequest {
        private final URL endpoint;
        private final int connectionTimeoutMs;
        private final HTTPMethod method;
        private final String body;
 
        private HTTPRequest(final URL url, final int milliseconds, final HTTPMethod httpMethod, final String httpBody) {
            this.endpoint = url;
            this.connectionTimeoutMs = milliseconds;
            this.method = httpMethod;
            this.body = httpBody;
        }
 
        @Override
        public String toString() {
            return "HTTPRequest{" + "endpoint=" + endpoint + ", connectionTimeoutMs=" + connectionTimeoutMs +
                    ", method=" + method + ", body='" + body + '\'' + '}';
        }
    }
 
    @Override
    public void execute(final JobExecutionContext context) throws JobExecutionException {
 
  final HTTPRequest request = parseParameters(context.getMergedJobDataMap());
 
        try {
            final HttpUriRequest httpRequest = newRequest(request.endpoint, request.method, request.body);
            final ResponseHandler<String> responseHandler = new BasicResponseHandler();
            client.getParams().setParameter(CoreConnectionPNames.CONNECTION_TIMEOUT, request.connectionTimeoutMs);
 
            final String response = client.execute(httpRequest, responseHandler);
 
        } catch (final Exception e) {
            throw new JobExecutionException(e);
        } finally {
            client.getConnectionManager().shutdown();
        }
    }
 
    private HttpUriRequest newRequest(final URL url, final HTTPMethod method, final String httpBody)
            throws URISyntaxException, JobExecutionException, UnsupportedEncodingException {
        switch (method) {
            case GET:
                return new HttpGet(url.toURI());
            case DELETE:
                return new HttpDelete(url.toURI());
            case POST:
                final HttpPost post = new HttpPost(url.toURI());
                post.setEntity(new StringEntity(httpBody));
                return post;
            case PUT:
                final HttpPut put = new HttpPut(url.toURI());
                put.setEntity(new StringEntity(httpBody));
                return put;
            default: throw new JobExecutionException("Unsupported HTTP Method");
        }
    }
 
   
    private HTTPRequest parseParameters(final JobDataMap mergedJobDataMap) throws JobExecutionException {
 
        final StringBuilder bodyFileSB = new StringBuilder(); // the SB to hold the body file's contents
        final String endpointStr = mergedJobDataMap.getString("endpoint");
        final String connectionTimeoutStr = mergedJobDataMap.getString("connection-timeout-ms");
        final String httpBodyFileName = mergedJobDataMap.getString("http-body-file");
        final String httpMethodStr = mergedJobDataMap.getString("http-method");
        final boolean httpMethodIsSet = StringUtils.isNotBlank(httpMethodStr);
        final boolean weHaveAnHttpBodyFile = StringUtils.isNotBlank(httpBodyFileName);
        final String httpBodyStr = mergedJobDataMap.getString("http-body");
 
        URL url;
        int connectionTimeoutMs = DEFAULT_TIMEOUT_MS; // 10 seconds default
        FileInputStream resource = null;
        Scanner scanner = null;
        HTTPMethod httpMethod = HTTPMethod.GET;
 
        try {
            url = new URL(endpointStr);
 
            if (connectionTimeoutStr != null && !"".equals(connectionTimeoutStr)) {
                connectionTimeoutMs = Integer.parseInt(connectionTimeoutStr);
            }
 
            if (weHaveAnHttpBodyFile) {
                LOGGER.info(format("Reading file '%s'", httpBodyFileName));
                resource = new FileInputStream(httpBodyFileName);
                scanner = new Scanner(resource, "UTF-8");
 
                while (scanner.hasNext()) {
                    bodyFileSB.append(scanner.next());
                }
            }
 
        } catch (final Exception e) {
        } finally {
           ..close all the objects
        }
 
        if (httpMethodIsSet) {
            try {
                httpMethod = HTTPMethod.valueOf(httpMethodStr);
            } catch (final IllegalArgumentException iie) {
                throw new JobExecutionException("Unsupported HTTP method", iie);
            }
        }
 
        return new HTTPRequest(url, connectionTimeoutMs, httpMethod, weHaveAnHttpBodyFile ? bodyFileSB.toString() : httpBodyStr);
    }
}

Finally here is the stateless Session EJB which listens to the JNDI name and implements interface ResourceController checkResource() method.

@Stateless(mappedName = "exampleProcessor_V3")
public class ExampleProcessor implements ResourceController {
 
 public void checkResource(Map<String, String> keyValueProperties) {
   ..... logic goes here
 } 
}
 

Previously we used to have a scheduler which triggers every 10 seconds and make sure it fires the EJB. Example xml which we had previously

<schedule>
        <job>
            <name>exampleProcessor</name>
            <group>MYJOB_GROUP</group>
 
            <description>The job description</description>
            <job-class>test.ConcurrencyThresholdAwareJob</job-class>
            <job-data-map allows-transient-data="false">
 
                <entry>
                    <key>jndi-location</key>
                    <value>exampleProcessor</value>
                </entry>
                
                <entry>
                 <key>maxConcurrentJobs</key>
     <value>24</value>
                </entry>
 
            </job-data-map>
        </job>
 
        <trigger>
            <cron>
                <cron-expression>0/10 * * * * ?</cron-expression>
            </cron>
        </trigger>

CustomSchedulerJobDelegator.java
 
public class CustomSchedulerJobDelegator implements Job {
 
    private static HashMap<String, ResourceController> jndiCache = new HashMap<String, ResourceController>();
 
    public void execute(final JobExecutionContext contextParam) throws JobExecutionException {
        ResourceController resourceController = null;
        String jndiLocation = null;
        try {
            jndiLocation = (String) contextParam.getMergedJobDataMap().get(Constants.REC_JNDI_CALL);
            resourceController = getInstanceOfController(jndiLocation, resourceController);
            executeService(resourceController, contextParam);
        } catch (NoSuchEJBException ex) {
            //most likely the jndi object from cache is not valid anymore becouse 
            //module being redeployed, so new one mast be put there
            System.err.println("WARNING Exception trying to execute service details:='" + ex.getMessage() +
                    "' Most likely object from the cache has been redeployed and now it is obsolete. ");
            //Logger.getLogger(CustomSchedulerJobDelegator.class.getName()).log(Level.WARNING, null, ex);
            //try to execute 2nd time
            //delete the obsolete object from the cache
            deleteObjectFromCache(jndiLocation);
            resourceController = getInstanceOfController(jndiLocation, resourceController);
            executeService(resourceController, contextParam);
        } catch (IllegalArgumentException ex) {
            System.err.println("EXCEPTION Trying to execute service details:='" + ex.getMessage() + "'");
            Logger.getLogger(CustomSchedulerJobDelegator.class.getName()).log(Level.SEVERE, null, ex);
        }
    }
 
    @SuppressWarnings("unchecked")
 private void executeService(ResourceController resourceController, JobExecutionContext contextParam) {
        resourceController.checkResource(contextParam.getMergedJobDataMap());
    }
 
    private ResourceController getInstanceOfController(String jndiLocation, ResourceController resourceController) {
        //the jndiCache is static , so the access must be locked via
        //class field
        synchronized (this.getClass()) {
            resourceController = jndiCache.get(jndiLocation);
            if (resourceController == null) {
                try {
                    final InitialContext context = new InitialContext();
                    resourceController = (ResourceController) context.lookup(jndiLocation);
                } catch (final NamingException ex) {
                }
                jndiCache.put(jndiLocation, resourceController);
            }
        }
        return resourceController;
    }
 
    private void deleteObjectFromCache(String objectName) {
        synchronized (this.getClass()) {
            jndiCache.remove(objectName);
        }
    }
}
 
We do have a ServletContextListener class which takes care of the house-keeping activities of start/stop scheduler.
 
Finally here is the stateless Session EJB which listens to the JNDI name and implements interface ResourceController checkResource() method.
 
@Stateless(mappedName = "exampleProcessor")
public class ExampleProcessor implements ResourceController {
 
 public void checkResource(Map<String, String> keyValueProperties) {
   ..... logic goes here
 } 
}