moves doRpc to RpcOverAmqp (and renames Settings.AMQPConn -> Settings AmqpConn) (5c6fcfc1) · Commits · ViaLactea / vlkb-soda

data-access/servlet/src/main/java/cutout/CutoutImpl.java

+1 −44

Original line number	Diff line number	Diff line
		@@ -379,7 +379,7 @@ class CutoutImpl implements Cutout
		jReq.add(countNullValues);
		jReq.add(extraCards);

		String outJson = doRpc( jReq.toString() );
		String outJson = RpcOverAmqp.doRpc( settings.amqpConn, jReq.toString() );

		cutResult = JsonDecoder.responseFromCutoutJson( outJson );
		}
		@@ -445,49 +445,6 @@ class CutoutImpl implements Cutout
		}



		private String doRpc(String InStr)
		{
		final String userName = "guest";
		final String password = "guest";
		// FIXME move these to Settings

		RpcOverAmqp rpc = new RpcOverAmqp(
		userName, password,
		settings.amqpConn.hostName(),
		settings.amqpConn.portNumber(),
		settings.amqpConn.routingKey());

		rpc.initConnectionAndReplyQueue();

		String OutStr = null;

		try
		{
		LOGGER.info("Sent request : " + InStr);
		OutStr = rpc.callAndWaitReply(InStr);
		LOGGER.info("Got response : " + OutStr);
		}
		catch (Exception e)
		{
		e.printStackTrace();
		}
		finally
		{
		try
		{
		rpc.close();
		}
		catch (Exception ignore)
		{
		LOGGER.info("ignoring exception on rpc.close():" + ignore.getMessage());
		}
		}

		return OutStr;
		}


		private String generateSubimgPathname(String relPathname, int hdunum)
		{
		String cutfitsname = "vlkb-cutout";

data-access/servlet/src/main/java/cutout/DatasetsAmqp.java

+6 −50

Original line number	Diff line number	Diff line
		@@ -78,57 +78,13 @@ class DatasetsAmqp implements Datasets
		LOGGER.info("doMCutout over AMQP");
		String updatedJsonString = JdlMCutout.resolveAndUpdateJsonRequest(jdlJson, settings, subsurveys);
		LOGGER.info("doMCutout over AMQP : " + updatedJsonString);
		String outJson = doRpc( JdlMCutout.mcutoutToJson(updatedJsonString) );
		String outJson = RpcOverAmqp.doRpc(settings.amqpConn, JdlMCutout.mcutoutToJson(updatedJsonString) );
		mCutResult = JdlMCutout.responseFromMCutoutJson(outJson);

		return mCutResult;
		}


		private String doRpc(String InStr)
		{
		LOGGER.info("trace");

		final String userName = "guest";
		final String password = "guest";
		// FIXME move these to Settings

		RpcOverAmqp rpc = new RpcOverAmqp(
		userName, password,
		settings.amqpConn.hostName(),
		settings.amqpConn.portNumber(),
		settings.amqpConn.routingKey());

		rpc.initConnectionAndReplyQueue();

		String OutStr = null;

		try
		{
		LOGGER.info("Sent request : " + InStr);
		OutStr = rpc.callAndWaitReply(InStr);
		LOGGER.info("Got response : " + OutStr);
		}
		catch (Exception e)
		{
		e.printStackTrace();
		}
		finally
		{
		try
		{
		rpc.close();
		}
		catch (Exception ignore)
		{
		LOGGER.info("ignoring exception on rpc.close():" + ignore.getMessage());
		}
		}

		return OutStr;
		}


		/* ================= MERGE =============================== */

		private String generateSubimgPathname(String relPathname, int hdunum)
		@@ -184,7 +140,7 @@ class DatasetsAmqp implements Datasets

		String inJson = jReq.toString();

		return JsonDecoder.responseFromCutoutJson( doRpc(inJson) );
		return JsonDecoder.responseFromCutoutJson( RpcOverAmqp.doRpc(settings.amqpConn, inJson) );
		}


		@@ -278,7 +234,7 @@ class DatasetsAmqp implements Datasets
		LOGGER.info("trace");

		String InJson = JsonEncoderMerge.mergefilesToJson( prefix, filestomerge);
		String OutJson = doRpc(InJson);
		String OutJson = RpcOverAmqp.doRpc(settings.amqpConn, InJson);
		return JsonDecoder.responseFromCutoutJson( OutJson );
		}

		@@ -384,7 +340,7 @@ class DatasetsAmqp implements Datasets
		LOGGER.info("trace");

		String InJson = JsonEncoderMerge.mergefilesCommonHeaderToJson(jobId, prefix, filestomerge);
		String OutJson = doRpc(InJson);
		String OutJson = RpcOverAmqp.doRpc(settings.amqpConn, InJson);
		String[] results = null; // FIXME JsonDecoder.responseFromJson(OutJson);

		return results;
		@@ -399,7 +355,7 @@ class DatasetsAmqp implements Datasets
		LOGGER.info("trace");

		String InJson = JsonEncoderMerge.mergefilesReprojectToJson(jobId, prefix, fitsfilename);
		String OutJson = doRpc(InJson);
		String OutJson = RpcOverAmqp.doRpc(settings.amqpConn, InJson);
		String[] results = null; // FIXME JsonDecoder.responseFromJson(OutJson);

		return results;
		@@ -413,7 +369,7 @@ class DatasetsAmqp implements Datasets
		LOGGER.info("trace");

		String InJson = JsonEncoderMerge.mergefilesAddReprojectedToJson(jobId, prefix);
		String OutJson = doRpc(InJson);
		String OutJson = RpcOverAmqp.doRpc(settings.amqpConn, InJson);
		String[] results = null; // FIXME JsonDecoder.responseFromJson(OutJson);

		return results;

data-access/servlet/src/main/java/cutout/RpcOverAmqp.java

+44 −0

Original line number	Diff line number	Diff line
		@@ -13,6 +13,7 @@
		// of vlkb-requests from Exchange to the correct queue


		import java.util.logging.Logger;

		import com.rabbitmq.client.ConnectionFactory;
		import com.rabbitmq.client.Connection;
		@@ -24,6 +25,8 @@ import java.util.UUID;

		public class RpcOverAmqp
		{
		private static final Logger LOGGER = Logger.getLogger("RpcOverAmqp");

		private final boolean NO_ACK = true;
		// affects message consume from queue:
		// broker will remove msg right after delivery without waiting for confirmation
		@@ -43,6 +46,47 @@ public class RpcOverAmqp
		private int channelNumber;


		public static String doRpc(Settings.AmqpConn amqpConn, String InStr)
		{
		final String userName = "guest";
		final String password = "guest";
		// FIXME move these to Settings

		RpcOverAmqp rpc = new RpcOverAmqp(
		userName, password,
		amqpConn.hostName(),
		amqpConn.portNumber(),
		amqpConn.routingKey());

		rpc.initConnectionAndReplyQueue();

		String OutStr = null;

		try
		{
		LOGGER.info("Sent request : " + InStr);
		OutStr = rpc.callAndWaitReply(InStr);
		LOGGER.info("Got response : " + OutStr);
		}
		catch (Exception e)
		{
		e.printStackTrace();
		}
		finally
		{
		try
		{
		rpc.close();
		}
		catch (Exception ignore)
		{
		LOGGER.info("ignoring exception on rpc.close():" + ignore.getMessage());
		}
		}

		return OutStr;
		}

		RpcOverAmqp(String userName, String password, String hostName, int portNumber, String routingKey)
		{
		this.userName = userName;

data-access/servlet/src/main/java/webapi/Settings.java

+6 −6

Original line number	Diff line number	Diff line
		@@ -66,7 +66,7 @@ class Settings
		}


		public static class AMQPConn
		public static class AmqpConn
		{
		private String hostName;
		private int portNum;
		@@ -89,7 +89,7 @@ class Settings

		public FITSPaths fitsPaths;
		public DBConn dbConn;
		public AMQPConn amqpConn;
		public AmqpConn amqpConn;
		public DefaultParamValues defaults;


		@@ -109,7 +109,7 @@ class Settings

		FITSPaths fitsPaths = loadFITSPaths(properties);
		DBConn dbConn = loadDBConn(properties);
		AMQPConn amqpConn = loadAMQPConn(properties);
		AmqpConn amqpConn = loadAmqpConn(properties);
		DefaultParamValues defaults = loadDefaults(properties);

		return new Settings(dbConn, amqpConn, fitsPaths, defaults);
		@@ -128,7 +128,7 @@ class Settings



		private Settings(DBConn dbConn, AMQPConn amqpConn,
		private Settings(DBConn dbConn, AmqpConn amqpConn,
		FITSPaths fitsPaths, DefaultParamValues defaults)
		{
		this.fitsPaths = fitsPaths;
		@@ -169,9 +169,9 @@ class Settings
		return defaults;
		}

		private static AMQPConn loadAMQPConn(Properties properties)
		private static AmqpConn loadAmqpConn(Properties properties)
		{
		AMQPConn amqpconn = new AMQPConn();
		AmqpConn amqpconn = new AmqpConn();
		amqpconn.hostName = properties.getProperty("amqp_host_name", "").strip();
		String strPortNum = properties.getProperty("amqp_port", "5672").strip();
		amqpconn.portNum = Integer.parseInt(strPortNum);