X-Git-Url: https://gerrit.onap.org/r/gitweb?p=dmaap%2Fmessagerouter%2Fmirroragent.git;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fcom%2Fatt%2Fnsa%2FdmaapMMAgent%2FMirrorMakerAgent.java;h=3f993db9f926edce12bdd7e7de6f763986957710;hp=b28fb075a912df2549d48528a25e5f1ce80a8e0f;hb=79f026709bde8b26fd9b5f31535ee429497b951d;hpb=57cc651fd62897fab90085ab94569354c27570d0 diff --git a/src/main/java/com/att/nsa/dmaapMMAgent/MirrorMakerAgent.java b/src/main/java/com/att/nsa/dmaapMMAgent/MirrorMakerAgent.java index b28fb07..3f993db 100644 --- a/src/main/java/com/att/nsa/dmaapMMAgent/MirrorMakerAgent.java +++ b/src/main/java/com/att/nsa/dmaapMMAgent/MirrorMakerAgent.java @@ -22,20 +22,16 @@ package com.att.nsa.dmaapMMAgent; -import java.io.BufferedReader; -import java.io.DataOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; -import java.io.InputStreamReader; import java.io.OutputStream; -import java.net.HttpURLConnection; -import java.net.URL; import java.util.ArrayList; import java.util.Properties; +import org.apache.log4j.Level; import org.apache.log4j.Logger; import org.jasypt.util.text.BasicTextEncryptor; @@ -48,8 +44,6 @@ import com.att.nsa.dmaapMMAgent.dao.UpdateWhiteList; import com.att.nsa.dmaapMMAgent.utils.MirrorMakerProcessHandler; import com.google.gson.Gson; import com.google.gson.internal.LinkedTreeMap; -import com.sun.org.apache.xerces.internal.impl.dtd.models.CMAny; -import com.sun.org.apache.xerces.internal.impl.dv.util.Base64; public class MirrorMakerAgent { static final Logger logger = Logger.getLogger(MirrorMakerAgent.class); @@ -61,6 +55,8 @@ public class MirrorMakerAgent { String topicname = ""; String mechid = ""; String password = ""; + TopicUtil topicUtil = new TopicUtil(); + public boolean exitLoop = false; private static String secret = "utdfpWlgyDQ2ZB8SLVRtmN834I1JcT9J"; public static void main(String[] args) { @@ -95,7 +91,8 @@ public class MirrorMakerAgent { input = new FileInputStream(mmagenthome + "/etc/mmagent.config"); logger.info("mmagenthome is set :" + mmagenthome + " loading properties at /etc/mmagent.config"); } catch (IOException ex) { - logger.error(mmagenthome + "/etc/mmagent.config not found. Set -DMMAGENTHOME and check the config file" + ex); + logger.error( + mmagenthome + "/etc/mmagent.config not found. Set -DMMAGENTHOME and check the config file" + ex); return false; } finally { if (input != null) { @@ -109,8 +106,11 @@ public class MirrorMakerAgent { loadProperties(); input = null; try { - /*input = new FileInputStream(kafkahome + "/bin/kafka-run-class.sh");*/ - if(false) { + /* + * input = new FileInputStream(kafkahome + + * "/bin/kafka-run-class.sh"); + */ + if (false) { throw new IOException(); } logger.info("kakahome is set :" + kafkahome); @@ -126,14 +126,14 @@ public class MirrorMakerAgent { } } } - String response = publishTopic("{\"test\":\"test\"}"); + String response = topicUtil.publishTopic(topicURL, topicname, mechid, password, "{\"test\":\"test\"}"); if (response.startsWith("ERROR:")) { logger.error("Problem publishing to topic, please verify the config " + this.topicname + " MR URL is:" + this.topicURL + " Error is: " + response); return false; } logger.info("Published to Topic :" + this.topicname + " Successfully"); - response = subscribeTopic("1"); + response = topicUtil.subscribeTopic(topicURL, topicname, "1", response, response); if (response != null && response.startsWith("ERROR:")) { logger.error("Problem subscribing to topic, please verify the config " + this.topicname + " MR URL is:" + this.topicURL + " Error is: " + response); @@ -167,14 +167,14 @@ public class MirrorMakerAgent { prop.store(out, ""); } catch (Exception e) { - logger.error("Exception at checkPropertiesFile " +e); + logger.error("Exception at checkPropertiesFile " + e); } } finally { if (input != null) { try { input.close(); } catch (IOException e) { - e.printStackTrace(); + logger.error("Exception occurred is " + e); } } if (out != null) { @@ -182,7 +182,7 @@ public class MirrorMakerAgent { out.close(); } catch (IOException e) { e.printStackTrace(); - logger.error("IOException" + e); + logger.error("Exception is : " + e); } } } @@ -199,7 +199,8 @@ public class MirrorMakerAgent { checkPropertiesFile(mm.name, "producer", mm.producer, false); if (mm.whitelist != null && !mm.whitelist.equals("")) { - logger.info("MirrorMaker " + mm.name + " is not running, restarting. Check Logs for more Details"); + logger.info( + "MirrorMaker " + mm.name + " is not running, restarting. Check Logs for more Details"); MirrorMakerProcessHandler.startMirrorMaker(this.mmagenthome, this.kafkahome, mm.name, mmagenthome + "/etc/" + mm.name + "consumer.properties", mmagenthome + "/etc/" + mm.name + "producer.properties", mm.whitelist); @@ -212,6 +213,7 @@ public class MirrorMakerAgent { try { Thread.sleep(1000); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } mirrorMakers.getListMirrorMaker().set(i, mm); } else { @@ -225,74 +227,13 @@ public class MirrorMakerAgent { // System.out.println(g.toJson(mirrorMakers)); } - private String subscribeTopic(String timeout) { - String response = ""; - try { - String requestURL = this.topicURL + "/events/" + this.topicname + "/mirrormakeragent/1?timeout=" + timeout - + "&limit=1"; - String authString = this.mechid + ":" + this.password; - String authStringEnc = Base64.encode(authString.getBytes()); - URL url = new URL(requestURL); - HttpURLConnection connection = (HttpURLConnection) url.openConnection(); - connection.setRequestMethod("GET"); - connection.setDoOutput(true); - connection.setRequestProperty("Authorization", "Basic " + authStringEnc); - connection.setRequestProperty("Content-Type", "application/json"); - InputStream content = (InputStream) connection.getInputStream(); - BufferedReader in = new BufferedReader(new InputStreamReader(content)); - String line; - - while ((line = in.readLine()) != null) { - response = response + line; - } - Gson g = new Gson(); - // get message as JSON String Array - String[] topicMessage = g.fromJson(response, String[].class); - if (topicMessage.length != 0) { - return topicMessage[0]; - } - } catch (Exception e) { - return "ERROR:" + e.getMessage() + " Server Response is:" + response; - } - return null; - } - - private String publishTopic(String message) { - try { - String requestURL = this.topicURL + "/events/" + this.topicname; - String authString = this.mechid + ":" + this.password; - String authStringEnc = Base64.encode(authString.getBytes()); - URL url = new URL(requestURL); - HttpURLConnection connection = (HttpURLConnection) url.openConnection(); - connection.setRequestMethod("POST"); - connection.setDoOutput(true); - connection.setRequestProperty("Authorization", "Basic " + authStringEnc); - connection.setRequestProperty("Content-Type", "application/json"); - connection.setRequestProperty("Content-Length", Integer.toString(message.length())); - DataOutputStream wr = new DataOutputStream(connection.getOutputStream()); - wr.write(message.getBytes()); - - InputStream content = (InputStream) connection.getInputStream(); - BufferedReader in = new BufferedReader(new InputStreamReader(content)); - String line; - String response = ""; - while ((line = in.readLine()) != null) { - response = response + line; - } - return response; - - } catch (Exception e) { - return "ERROR:" + e.getLocalizedMessage(); - } - } - - private void readAgentTopic() { + public void readAgentTopic() { try { int connectionattempt = 0; while (true) { logger.info("--------------------------------"); logger.info("Waiting for Messages for 60 secs"); - String topicMessage = subscribeTopic("60000"); + String topicMessage = topicUtil.subscribeTopic(topicURL, topicname, "60000", mechid, password); Gson g = new Gson(); LinkedTreeMap object = null; if (topicMessage != null) { @@ -301,53 +242,11 @@ public class MirrorMakerAgent { // Cast the 1st item (since limit=1 and see the type of // object - if (object.get("createMirrorMaker") != null) { - logger.info("Received createMirrorMaker request from topic"); - CreateMirrorMaker m = g.fromJson(topicMessage, CreateMirrorMaker.class); - createMirrorMaker(m.getCreateMirrorMaker()); - checkAgentProcess(); - mirrorMakers.setMessageID(m.getMessageID()); - publishTopic(g.toJson(mirrorMakers)); - mirrorMakers.setMessageID(""); - } else if (object.get("updateMirrorMaker") != null) { - logger.info("Received updateMirrorMaker request from topic"); - UpdateMirrorMaker m = g.fromJson(topicMessage, UpdateMirrorMaker.class); - updateMirrorMaker(m.getUpdateMirrorMaker()); - checkAgentProcess(); - mirrorMakers.setMessageID(m.getMessageID()); - publishTopic(g.toJson(mirrorMakers)); - mirrorMakers.setMessageID(""); - } else if (object.get("deleteMirrorMaker") != null) { - logger.info("Received deleteMirrorMaker request from topic"); - DeleteMirrorMaker m = g.fromJson(topicMessage, DeleteMirrorMaker.class); - deleteMirrorMaker(m.getDeleteMirrorMaker()); - checkAgentProcess(); - mirrorMakers.setMessageID(m.getMessageID()); - publishTopic(g.toJson(mirrorMakers)); - mirrorMakers.setMessageID(""); - } else if (object.get("listAllMirrorMaker") != null) { - logger.info("Received listALLMirrorMaker request from topic"); - checkAgentProcess(); - mirrorMakers.setMessageID((String) object.get("messageID")); - publishTopic(g.toJson(mirrorMakers)); - mirrorMakers.setMessageID(""); - } else if (object.get("updateWhiteList") != null) { - logger.info("Received updateWhiteList request from topic"); - UpdateWhiteList m = g.fromJson(topicMessage, UpdateWhiteList.class); - updateWhiteList(m.getUpdateWhiteList()); - checkAgentProcess(); - mirrorMakers.setMessageID(m.getMessageID()); - publishTopic(g.toJson(mirrorMakers)); - mirrorMakers.setMessageID(""); - } else if (object.get("listMirrorMaker") != null) { - logger.info("Received listMirrorMaker from topic, skipping messages"); - } else { - logger.info("Received unknown request from topic"); - } + readAgent(object, topicMessage); } catch (Exception ex) { connectionattempt++; if (connectionattempt > 5) { - logger.info("Can't connect to the topic, mmagent shutting down , " + topicMessage); + logger.info("Can't connect to the topic, mmagent shutting down , " + topicMessage + ex); return; } logger.info("Can't connect to the topic, " + topicMessage + " Retrying " + connectionattempt @@ -359,10 +258,61 @@ public class MirrorMakerAgent { connectionattempt = 0; checkAgentProcess(); } - + if (exitLoop) { + break; + } } } catch (Exception e) { - e.printStackTrace(); + logger.error("Exception at readAgentTopic : " + e); + } + + } + + public void readAgent(LinkedTreeMap object, String topicMessage) { + + Gson g = new Gson(); + + if (object.get("createMirrorMaker") != null) { + logger.info("Received createMirrorMaker request from topic"); + CreateMirrorMaker m = g.fromJson(topicMessage, CreateMirrorMaker.class); + createMirrorMaker(m.getCreateMirrorMaker()); + checkAgentProcess(); + mirrorMakers.setMessageID(m.getMessageID()); + topicUtil.publishTopic(topicMessage, topicMessage, topicMessage, topicMessage, g.toJson(mirrorMakers)); + mirrorMakers.setMessageID(""); + } else if (object.get("updateMirrorMaker") != null) { + logger.info("Received updateMirrorMaker request from topic"); + UpdateMirrorMaker m = g.fromJson(topicMessage, UpdateMirrorMaker.class); + updateMirrorMaker(m.getUpdateMirrorMaker()); + checkAgentProcess(); + mirrorMakers.setMessageID(m.getMessageID()); + topicUtil.publishTopic(topicMessage, topicMessage, topicMessage, topicMessage, g.toJson(mirrorMakers)); + mirrorMakers.setMessageID(""); + } else if (object.get("deleteMirrorMaker") != null) { + logger.info("Received deleteMirrorMaker request from topic"); + DeleteMirrorMaker m = g.fromJson(topicMessage, DeleteMirrorMaker.class); + deleteMirrorMaker(m.getDeleteMirrorMaker()); + checkAgentProcess(); + mirrorMakers.setMessageID(m.getMessageID()); + topicUtil.publishTopic(topicMessage, topicMessage, topicMessage, topicMessage, g.toJson(mirrorMakers)); + mirrorMakers.setMessageID(""); + } else if (object.get("listAllMirrorMaker") != null) { + logger.info("Received listALLMirrorMaker request from topic"); + checkAgentProcess(); + mirrorMakers.setMessageID((String) object.get("messageID")); + topicUtil.publishTopic(topicMessage, topicMessage, topicMessage, topicMessage, g.toJson(mirrorMakers)); + } else if (object.get("updateWhiteList") != null) { + logger.info("Received updateWhiteList request from topic"); + UpdateWhiteList m = g.fromJson(topicMessage, UpdateWhiteList.class); + updateWhiteList(m.getUpdateWhiteList()); + checkAgentProcess(); + mirrorMakers.setMessageID(m.getMessageID()); + topicUtil.publishTopic(topicMessage, topicMessage, topicMessage, topicMessage, g.toJson(mirrorMakers)); + mirrorMakers.setMessageID(""); + } else if (object.get("listMirrorMaker") != null) { + logger.info("Received listMirrorMaker from topic, skipping messages"); + } else { + logger.info("Received unknown request from topic"); } } @@ -386,7 +336,7 @@ public class MirrorMakerAgent { } else if (exists == false && mirrorMakers == null) { mirrorMakers = new ListMirrorMaker(); ArrayList list = mirrorMakers.getListMirrorMaker(); - list = new ArrayList(); + list = new ArrayList(); list.add(newMirrorMaker); mirrorMakers.setListMirrorMaker(list); } @@ -441,15 +391,17 @@ public class MirrorMakerAgent { try { Thread.sleep(1000); } catch (InterruptedException e) { + logger.log(Level.WARN, "Interrupted!", e); + Thread.currentThread().interrupt(); } } catch (IOException ex) { - ex.printStackTrace(); + logger.error(" IOException Occered " + ex); } finally { if (out != null) { try { out.close(); } catch (IOException e) { - e.printStackTrace(); + logger.error(" IOException Occered " + e); } } } @@ -484,9 +436,11 @@ public class MirrorMakerAgent { try { Thread.sleep(1000); } catch (InterruptedException e) { + logger.log(Level.WARN, "Interrupted!", e); + Thread.currentThread().interrupt(); } } catch (IOException ex) { - ex.printStackTrace(); + logger.error("Exception at updateWhiteList : " + ex); } finally { if (out != null) { try { @@ -576,7 +530,8 @@ public class MirrorMakerAgent { BasicTextEncryptor textEncryptor = new BasicTextEncryptor(); textEncryptor.setPassword(secret); - //this.password = textEncryptor.decrypt(mirrorMakerProperties.getProperty("password")); + // this.password = + // textEncryptor.decrypt(mirrorMakerProperties.getProperty("password")); this.password = mirrorMakerProperties.getProperty("password"); } catch (IOException ex) { // ex.printStackTrace();