tags/uima-as-2.4.2/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/UimacppServiceManagement.java - uima-async-scaleout - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.uima.aae.controller;

 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.io.PrintWriter;
 import java.net.Socket;
 import java.util.HashMap;
 import java.util.StringTokenizer;

 /**
  * This class implements the JMX MBean interface to provide operational statistics about the C++
  * service.
  */
 public class UimacppServiceManagement implements UimacppServiceManagementMBean {

   private static final long serialVersionUID = -2507413276728501209L;

   private static final long MAX_TIME_VALID = 500;

   private static long lastRefreshTime;

   HashMap<String, String> jmxInfo;

   private String uniqueMBeanName;

   private String aeDescriptor;

   private int aeInstances;

   private String queueBrokerURL;

   private String queueName;

   Socket socket;

   BufferedReader rdr;

   PrintWriter writer;

   public UimacppServiceManagement(String domainName, Socket sock, String aeDescriptor,
           int numInstances, String brokerURL, String queueName) throws IOException {

     if (domainName == null || domainName.length() == 0) {
       domainName = "org.apache.uima:type=ee.jms.services,s=" + queueName + " Uima EE Service,";
     }
     uniqueMBeanName = domainName + "name=" + queueName + "_Service";
     socket = sock;
     rdr = new BufferedReader(new InputStreamReader(socket.getInputStream()));
     writer = new PrintWriter(socket.getOutputStream(), true);
     this.aeDescriptor = aeDescriptor;
     this.queueBrokerURL = brokerURL;
     this.queueName = queueName;
     this.aeInstances = numInstances;
     this.jmxInfo = new HashMap();
     this.lastRefreshTime = 0;
   }

   synchronized public String quiesceAndStop() throws IOException {

 	  if (socket != null) {
 		// System.out.println("UimacppServiceManagement::quiesceAndStop()
 		// Sending QUIESCEANDSTOP");

 		writer.write("QUIESCEANDSTOP");
 		writer.flush();

 		BufferedReader in = new BufferedReader(new InputStreamReader(socket
 				.getInputStream()));

 		StringBuffer sb = new StringBuffer();
 		int c = in.read();
 		while (c >= 0) {
 		  sb.append((char) c);
 		  c = in.read();
 		  if (c == '\n') {
 			break;
 		  }
 		}
 		System.out.println("UimacppServiceManagement service reports QuiesceAndStop " + sb.toString());
 		return sb.toString();
 		} else {
 		  throw new IOException("Error: no socket connection.");
 		}
 	}


   synchronized public String getStatisticsAsString() throws IOException {

     if (socket != null) {
       // System.out.println("UimacppServiceManagement::getStatisticsAsString() Sending GETSTATS");

       writer.write("GETSTATS");
       writer.flush();

       // OutputStream os = socket.getOutputStream();
       // os.flush();
       // os.write(getstats.getBytes());

       // System.out.println("getStatistics() Sent GETSTATS Waiting for reply");
       BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

       StringBuffer sb = new StringBuffer();
       int c = in.read();
       while (c >= 0) {
         sb.append((char) c);
         c = in.read();
         if (c == '\n') {
           break;
         }
       }
       this.lastRefreshTime = System.currentTimeMillis();

       return sb.toString();
     } else {
       throw new IOException("Error: no socket connection.");
     }
   }

   synchronized public void getStatistics() throws IOException {

     String stats = getStatisticsAsString();
     String name = null;
     String value = null;
     if (stats != null) {
       StringTokenizer tokenizer = new StringTokenizer(stats);
       while (tokenizer.hasMoreTokens()) {
         String aStat = tokenizer.nextToken();
         StringTokenizer st2 = new StringTokenizer(aStat, "=");
         name = null;
         value = null;
         if (st2.hasMoreElements()) {
           name = st2.nextToken();
         }
         if (st2.hasMoreElements()) {
           value = st2.nextToken();
         }
         if (name == null) {
           throw new IOException("Statistic name is not set.");
         }
         if (value == null) {
           throw new IOException("Statistic value is not set for " + name);
         }
         jmxInfo.put(name, value);
       }
     }
   }

   synchronized public String getQueueBrokerURL() throws IOException {
     // System.out.println("QueueBrokerURL");
     /*
      * We are assuming this method gets called first by the MBeanServer. So we send a request to C++
      * service to get latest stats.
      */

     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }

     return queueBrokerURL;
   }

   synchronized public String getQueueName() throws IOException {
     // System.out.println("QueueName");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     return queueName;
   }

   synchronized public String getAEDescriptor() throws IOException {
     // System.out.println("aeDescriptor");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     return aeDescriptor;
   }

   synchronized public int getAEInstances() throws IOException {
     // System.out.println("aeInstances");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     return aeInstances;

   }

   synchronized public long getErrorsGetMeta() throws IOException {
     // System.out.println("GETMETAERRORS");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("GETMETAERRORS");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getErrorsProcessCas() throws IOException {
     // System.out.println("errorsProcessCas");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("PROCESSCASERRORS");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getErrorsCPC() throws IOException {
     // System.out.println("errorsCPC");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("CPCERRORS");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTotalNumCasProcessed() throws IOException {
     // System.out.println("TotalNumCasProcessed");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("NUMCASPROCESSED");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingGetMeta() throws IOException {
     // System.out.println("getTimingGetMeta");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("GETMETATIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingCPC() throws IOException {
     // System.out.println("getTimingCPC");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("CPCTIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingSerialization() throws IOException {
     // System.out.println("getTimingSerialization");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("SERIALIZETIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingAnnotatorProcess() throws IOException {
     // System.out.println("getTimingAnnotatorProcess");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("ANNOTATORTIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingDeserialization() throws IOException {
     // System.out.println("getTimingDeserialization");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("DESERIALIZETIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingMessageProcessing() throws IOException {
     // System.out.println("getTimingMessageProcessing");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("MESSAGEPROCESSTIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public long getTimingIdle() throws IOException {
     // System.out.println("getTimingIdle");
     if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
       getStatistics();
     }
     String v = jmxInfo.get("IDLETIME");
     if (v == null) {
       return -1;
     } else {
       return Long.valueOf(v);
     }
   }

   synchronized public void resetStats() throws IOException {
     writer.write("RESET");
     writer.flush();
     getStatistics();
   }

   public void increaseAEInstances(int num) {
     // TODO Auto-generated method stub

   }

   public void decreaseAEInstances(int num) {
     // TODO Auto-generated method stub

   }

   synchronized public void shutdown() throws IOException {
     if (this.socket != null) {
       // System.out.println("UimacppServiceManagement sending shutdown message");
       writer.write("SHUTDOWN");
       writer.flush();
       // System.out.println("UimacppServiceManagement sent shutdown message");
   		BufferedReader in = new BufferedReader(new InputStreamReader(socket
   				.getInputStream()));

   		StringBuffer sb = new StringBuffer();
   		int c = in.read();
   		while (c >= 0) {
   		  sb.append((char) c);
   		  c = in.read();
   		  if (c == '\n') {
   			break;
   		  }
   		}
   	  System.out.println("UimacppServiceManagement service reports shutdown " + sb.toString());
       return;
     } else {
       System.err.println("Error no connection");
     }
   }

   public String getUniqueMBeanName() {
     return uniqueMBeanName;
   }

 	synchronized public void quiesceAndShutdown() throws IOException {
 		quiesceAndStop();
 	}

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.uima.aae.controller;

	import java.io.BufferedReader;
	import java.io.IOException;
	import java.io.InputStreamReader;
	import java.io.OutputStream;
	import java.io.PrintWriter;
	import java.net.Socket;
	import java.util.HashMap;
	import java.util.StringTokenizer;

	/**
	* This class implements the JMX MBean interface to provide operational statistics about the C++
	* service.
	*/
	public class UimacppServiceManagement implements UimacppServiceManagementMBean {

	private static final long serialVersionUID = -2507413276728501209L;

	private static final long MAX_TIME_VALID = 500;

	private static long lastRefreshTime;

	HashMap<String, String> jmxInfo;

	private String uniqueMBeanName;

	private String aeDescriptor;

	private int aeInstances;

	private String queueBrokerURL;

	private String queueName;

	Socket socket;

	BufferedReader rdr;

	PrintWriter writer;

	public UimacppServiceManagement(String domainName, Socket sock, String aeDescriptor,
	int numInstances, String brokerURL, String queueName) throws IOException {

	if (domainName == null \|\| domainName.length() == 0) {
	domainName = "org.apache.uima:type=ee.jms.services,s=" + queueName + " Uima EE Service,";
	}
	uniqueMBeanName = domainName + "name=" + queueName + "_Service";
	socket = sock;
	rdr = new BufferedReader(new InputStreamReader(socket.getInputStream()));
	writer = new PrintWriter(socket.getOutputStream(), true);
	this.aeDescriptor = aeDescriptor;
	this.queueBrokerURL = brokerURL;
	this.queueName = queueName;
	this.aeInstances = numInstances;
	this.jmxInfo = new HashMap();
	this.lastRefreshTime = 0;
	}

	synchronized public String quiesceAndStop() throws IOException {

	if (socket != null) {
	// System.out.println("UimacppServiceManagement::quiesceAndStop()
	// Sending QUIESCEANDSTOP");

	writer.write("QUIESCEANDSTOP");
	writer.flush();

	BufferedReader in = new BufferedReader(new InputStreamReader(socket
	.getInputStream()));

	StringBuffer sb = new StringBuffer();
	int c = in.read();
	while (c >= 0) {
	sb.append((char) c);
	c = in.read();
	if (c == '\n') {
	break;
	}
	}
	System.out.println("UimacppServiceManagement service reports QuiesceAndStop " + sb.toString());
	return sb.toString();
	} else {
	throw new IOException("Error: no socket connection.");
	}
	}


	synchronized public String getStatisticsAsString() throws IOException {

	if (socket != null) {
	// System.out.println("UimacppServiceManagement::getStatisticsAsString() Sending GETSTATS");

	writer.write("GETSTATS");
	writer.flush();

	// OutputStream os = socket.getOutputStream();
	// os.flush();
	// os.write(getstats.getBytes());

	// System.out.println("getStatistics() Sent GETSTATS Waiting for reply");
	BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));

	StringBuffer sb = new StringBuffer();
	int c = in.read();
	while (c >= 0) {
	sb.append((char) c);
	c = in.read();
	if (c == '\n') {
	break;
	}
	}
	this.lastRefreshTime = System.currentTimeMillis();

	return sb.toString();
	} else {
	throw new IOException("Error: no socket connection.");
	}
	}

	synchronized public void getStatistics() throws IOException {

	String stats = getStatisticsAsString();
	String name = null;
	String value = null;
	if (stats != null) {
	StringTokenizer tokenizer = new StringTokenizer(stats);
	while (tokenizer.hasMoreTokens()) {
	String aStat = tokenizer.nextToken();
	StringTokenizer st2 = new StringTokenizer(aStat, "=");
	name = null;
	value = null;
	if (st2.hasMoreElements()) {
	name = st2.nextToken();
	}
	if (st2.hasMoreElements()) {
	value = st2.nextToken();
	}
	if (name == null) {
	throw new IOException("Statistic name is not set.");
	}
	if (value == null) {
	throw new IOException("Statistic value is not set for " + name);
	}
	jmxInfo.put(name, value);
	}
	}
	}

	synchronized public String getQueueBrokerURL() throws IOException {
	// System.out.println("QueueBrokerURL");
	/*
	* We are assuming this method gets called first by the MBeanServer. So we send a request to C++
	* service to get latest stats.
	*/

	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}

	return queueBrokerURL;
	}

	synchronized public String getQueueName() throws IOException {
	// System.out.println("QueueName");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	return queueName;
	}

	synchronized public String getAEDescriptor() throws IOException {
	// System.out.println("aeDescriptor");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	return aeDescriptor;
	}

	synchronized public int getAEInstances() throws IOException {
	// System.out.println("aeInstances");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	return aeInstances;

	}

	synchronized public long getErrorsGetMeta() throws IOException {
	// System.out.println("GETMETAERRORS");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("GETMETAERRORS");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getErrorsProcessCas() throws IOException {
	// System.out.println("errorsProcessCas");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("PROCESSCASERRORS");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getErrorsCPC() throws IOException {
	// System.out.println("errorsCPC");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("CPCERRORS");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTotalNumCasProcessed() throws IOException {
	// System.out.println("TotalNumCasProcessed");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("NUMCASPROCESSED");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingGetMeta() throws IOException {
	// System.out.println("getTimingGetMeta");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("GETMETATIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingCPC() throws IOException {
	// System.out.println("getTimingCPC");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("CPCTIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingSerialization() throws IOException {
	// System.out.println("getTimingSerialization");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("SERIALIZETIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingAnnotatorProcess() throws IOException {
	// System.out.println("getTimingAnnotatorProcess");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("ANNOTATORTIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingDeserialization() throws IOException {
	// System.out.println("getTimingDeserialization");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("DESERIALIZETIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingMessageProcessing() throws IOException {
	// System.out.println("getTimingMessageProcessing");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("MESSAGEPROCESSTIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public long getTimingIdle() throws IOException {
	// System.out.println("getTimingIdle");
	if (System.currentTimeMillis() - this.lastRefreshTime > MAX_TIME_VALID) {
	getStatistics();
	}
	String v = jmxInfo.get("IDLETIME");
	if (v == null) {
	return -1;
	} else {
	return Long.valueOf(v);
	}
	}

	synchronized public void resetStats() throws IOException {
	writer.write("RESET");
	writer.flush();
	getStatistics();
	}

	public void increaseAEInstances(int num) {
	// TODO Auto-generated method stub

	}

	public void decreaseAEInstances(int num) {
	// TODO Auto-generated method stub

	}

	synchronized public void shutdown() throws IOException {
	if (this.socket != null) {
	// System.out.println("UimacppServiceManagement sending shutdown message");
	writer.write("SHUTDOWN");
	writer.flush();
	// System.out.println("UimacppServiceManagement sent shutdown message");
	BufferedReader in = new BufferedReader(new InputStreamReader(socket
	.getInputStream()));

	StringBuffer sb = new StringBuffer();
	int c = in.read();
	while (c >= 0) {
	sb.append((char) c);
	c = in.read();
	if (c == '\n') {
	break;
	}
	}
	System.out.println("UimacppServiceManagement service reports shutdown " + sb.toString());
	return;
	} else {
	System.err.println("Error no connection");
	}
	}

	public String getUniqueMBeanName() {
	return uniqueMBeanName;
	}

	synchronized public void quiesceAndShutdown() throws IOException {
	quiesceAndStop();
	}

	}