A Look at DataSynapse GridServer (With Example)

A while back, I got the opportunity to work on a DataSynapse GridServer (service-oriented integration) service and client implementation using Java. This article is intended to teach you about this experience.

GridServer Architecture

• It's a highly scalable infrastructure that enables application services to operate in a virtualized fashion.

• Client applications submit requests asynchronously and in parallel to the GridServer.

• GridServer dynamically creates multiple service instances to respond to the requests.

• Service is a self-contained business implementation distributed to the engines.

• GridServer is based on service-oriented architecture.

There are two major components of a GridServer:

GridClients

This is responsible for submitting the requests to the grid with help of GridDriver. GridDriver provides the API to interact with GridServer. ResponseHandler is a callback implementation that gets the responses from GridServer.

GridServers

This is having two main parts: GridServerManager and Engines. GridServerManager is comprised of Director and Broker, where Director is responsible for routing, authentication, and load-balancing, and Broker is responsible for request queuing, scheduling, and communication. Engines are comprised of EngineDaemon and Engine, where EngineDaemon is responsible for managing the engine's (engine life cycle) and Engine is the actual guy who hosts and run the service to full-fill the request.

Let's look at the diagram below to get a better understanding: 

Now, let's look at an example.

The below example demonstrates the service and client implementation that will allow the caller to make many asynchronous requests in parallel.

Note: Grid Libraries (JAR in our case) are the enterprise-level method of deploying resources to Engines, which should have the grid-library.xml and service implemented class files.

grid-library.xml:

<?xml version="1.0" encoding="UTF-8"?>
<grid-library>
	<grid-library-name>GHDCaculator</grid-library-name>
	<grid-library-version>1.0</grid-library-version>
	<jar-path>
		<pathelement>jars</pathelement>
	</jar-path>
	<!--dependency><grid-library-name>library which may required </grid-library-name><grid-library-version>lib version</grid-library-version></dependency-->
</grid-library>

Here, GHDCaculator.zip will be the grid library that should be placed in the resources/gridlib directory of the Manager using the GridServer Admin tool. The service type should be registered from the Admin tool. In our case, the service name is GHDService.

Server-Side Implementation (Create a Service)

GridEuclidGcdService.java:

/**
 * @author arun.pandey
 * 
 * Euclid's GCD implementation as Service 
 */
public class GridEuclidGcdService {

 public String calculateGcd(long num1, long num2) {
   return "GCD of " + num1 + ":" + num2 + " = " + String.valueOf(calculateGcdRecursively(num1, num2));
  }
  /**
   * To calculate the GCD
   * @param num1
   * @param num2
   * @return
   */
 public long calculateGcdRecursively(long num1, long num2) {

  if (num1 % num2 == 0)
   return num2;
  return calculateGcdRecursively(num2, num1 % num2);
 }
}

GridEuclidGcdServiceHandler.java:

import java.io.IOException;
import java.io.Serializable;
import java.io.UncheckedIOException;
import java.io.Writer;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

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

import com.datasynapse.gridserver.client.ServiceInvocationException;
import com.datasynapse.gridserver.client.ServiceInvocationHandler;
import com.datasynapse.gridserver.engine.EngineProperties;
import com.datasynapse.gridserver.engine.EngineSession;

public class GridEuclidGcdServiceHandler implements ServiceInvocationHandler {

 private static final Log LOG = LogFactory.getLog(GridEuclidGcdServiceHandler.class);

 private static AtomicInteger totalNumberProcessed = new AtomicInteger();

 private ConcurrentHashMap < Integer, String > taskMap;
 private Path path;

 public GridEuclidGcdServiceHandler(ConcurrentHashMap < Integer, String > taskMap,
  String ghdResultPath) {
  this.taskMap = taskMap;
  this.path = Paths.get(ghdResultPath);
 }

 @Override
 public void handleError(ServiceInvocationException e, int id) {

  LOG.error("Error from Task ID " + id + ": " + e + " ENGINE HOSTNAME: " +
   EngineSession.getProperties().getProperty(EngineProperties.USERNAME) + " ENGINE INSTANCE: " + EngineSession.getProperties().getProperty(EngineProperties.INSTANCE));

  LOG.error("Task ID: " + id + " handled Numbers to calculate GHD : " + taskMap.get(id));
  LOG.info("Total Number Processed: " + totalNumberProcessed.incrementAndGet());
  taskMap.remove(id);
 }

 @Override
 public void handleResponse(Serializable response, int id) {
  if (null == response)
   LOG.info("Response is null for TASK ID: " + id);
  else
   LOG.info("Response from TASK ID: " + id + ": " + response + " getClass: " + response.getClass().getName());

  String ghdResultString = (String) response;

  if (null == ghdResultString || ghdResultString.isEmpty())
   LOG.info("Response from TASK ID: " + id + ": either null or empty");
  else
   try (Writer writer = Files.newBufferedWriter(path)) {
    writer.write(ghdResultString);
   } catch (UncheckedIOException ex) {
    LOG.error("Exception occured while writing result in File... " + ex.getMessage());
   } catch (IOException e) {
    LOG.error("Exception occured while writing result in File... " + e.getMessage());
   }

  LOG.info("Total Numbers Processed: " + totalNumberProcessed.incrementAndGet());
  taskMap.remove(id);
 }
}

Client-Side Implementation (Calling the Service)

GridClient.java:

import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.datasynapse.gridserver.client.Service;
import com.datasynapse.gridserver.client.ServiceException;
import com.datasynapse.gridserver.client.ServiceFactory;

/**
 * @author arun.pandey
 */
public class GridClient {
 private static final Log LOG = LogFactory.getLog(GridClient.class);

 private static String ghdResultFilePath;
 private static ConcurrentHashMap < Integer, String > taskMap = new ConcurrentHashMap < > ();
 private static Service s;
 public GridClient() {}

 public static void main(String[] args) throws ServiceException {

  GridClient grid = new GridClient();

  /** Here createService method is to create a Service instance for the Service, 
   * If 'GHDService' will not be registered, createService will throw a GridServerException 
   **/
  s = ServiceFactory.getInstance().createService("GHDService");

  /** This is to create a callback object that implements the ServiceInvocationHandler interface,
   * which will be passed to Grid while submitting the task
   **/
  GridEuclidGcdServiceHandler gridHandler = new GridEuclidGcdServiceHandler(taskMap, ghdResultFilePath);

  try {
   for (long i = 1; i < 1000; i++)
    grid.calculateGhd(i, i + 1, gridHandler);

   s.waitUntilInactive(0);
   s.destroyWhenInactive();
   System.exit(0);
  } catch (Exception e) {
   LOG.error("Error calling the GridService...", e);
   System.exit(1);
  }
 }

 /**
  * Calling the service - GridEuclidGcdService.gcd
  * 
  * @param num1
  * @param num2
  * @param gridHandler
  * @throws ServiceException
  */
 public void calculateGhd(long num1, long num2, GridEuclidGcdServiceHandler gridHandler) throws ServiceException {

  /** It prepares the arguments to be submitted to the Service, here two long numbers **/
  Object[] methodArguments = new Object[] {
   num1,
   num2
  };

  /** Submitting the task to Grid, and submit method returns an integer that uniquely 
   * identifies the particular call, which is getting used by GridEuclidGcdServiceHandler to match with response
   **/
  int taskId = s.submit("calculateGcd", methodArguments, gridHandler);
  LOG.info("Task submitted via GridClient, TaskId: " + taskId);

  taskMap.put(taskId, num1 + ":" + num2);
  return;
 }
}

Once the service deployment is complete, run GridClient and it will submit the tasks to Grid in an asynchronous and parallel way. The result will be handled by GridEuclidGcdServiceHandler, which will write the result to a file.

Happy learning!